for await遍历异步数据流

AsyncSequence为异步数据流提供了统一的接口，而for await则是遍历这些数据流的钥匙。结合async/await，for await允许开发者以同步循环的直观方式处理异步生成的元素，避免了传统回调的复杂性。本节将详细讲解for await的语法规则、使用方法及其在实际场景中的应用，通过代码示例带你掌握这一现代并发技术的核心技能。

基本语法与机制

for await是Swift中专门为AsyncSequence设计的循环结构，用于迭代异步序列中的元素。它的基本语法如下：

for await element in asyncSequence {
    // 处理每个元素
}
  • await：表示每次迭代可能暂停，等待下一个元素就绪。
  • asyncSequence：符合AsyncSequence协议的对象。

由于for await涉及异步操作，它必须在异步上下文中使用，通常包裹在Task或异步函数中。例如：

struct SimpleSequence: AsyncSequence {
    typealias Element = Int
    
    func makeAsyncIterator() -> SimpleIterator {
        SimpleIterator()
    }
}

struct SimpleIterator: AsyncIteratorProtocol {
    private var count = 0
    
    mutating func next() async -> Int? {
        try? await Task.sleep(nanoseconds: 500_000_000) // 模拟0.5秒延迟
        guard count < 5 else { return nil }
        count += 1
        return count
    }
}

Task {
    for await number in SimpleSequence() {
        print("数字：\(number)")
    }
    print("序列结束")
}

输出（每0.5秒一行）：

数字：1
数字：2
数字：3
数字：4
数字：5
序列结束

机制要点

  • 异步迭代：每次调用next()时，任务挂起直到元素可用。
  • 结束条件：next()返回nil时，循环终止。
  • 线程切换：迭代器运行于后台线程，需手动切换到主线程处理UI。

处理错误

AsyncSequence的迭代器可能抛出错误，需用try结合for await：

struct ErrorSequence: AsyncSequence {
    typealias Element = String
    
    func makeAsyncIterator() -> ErrorIterator {
        ErrorIterator()
    }
}

struct ErrorIterator: AsyncIteratorProtocol {
    private var count = 0
    
    mutating func next() async throws -> String? {
        try await Task.sleep(nanoseconds: 500_000_000)
        guard count < 3 else { return nil }
        count += 1
        if count == 2 { throw IteratorError.failure }
        return "消息\(count)"
    }
}

enum IteratorError: Error {
    case failure
}

Task {
    do {
        for try await message in ErrorSequence() {
            print(message)
        }
        print("序列完成")
    } catch {
        print("错误：\(error)")
    }
}

输出：

消息1
错误：failure
  • try await：捕获每次迭代的潜在错误。
  • 错误抛出后，循环中止。

典型使用场景

以下是for await在实际开发中的几个常见应用：

1. 实时数据流

处理服务器推送的实时消息：

func messageStream() -> some AsyncSequence {
    AsyncStream<String> { continuation in
        // 模拟WebSocket推送
        Task {
            for message in ["新闻1", "新闻2", "新闻3"] {
                try? await Task.sleep(nanoseconds: 1_000_000_000)
                continuation.yield(message)
            }
            continuation.finish()
        }
    }
}

Task { @MainActor in
    for await message in messageStream() {
        print("收到：\(message)")
        // 更新UI
        newsLabel.text = message
    }
}
  • AsyncStream动态生成元素。
  • @MainActor确保UI更新安全。

2. 文件流读取

逐行读取大文件：

func lines(from fileURL: URL) -> some AsyncSequence {
    AsyncStream<String> { continuation in
        Task {
            guard let file = try? FileHandle(forReadingFrom: fileURL) else {
                continuation.finish()
                return
            }
            defer { try? file.close() }
            
            while let line = try? file.read(upTo: "\n".data(using: .utf8)!) {
                try? await Task.sleep(nanoseconds: 500_000_000) // 模拟延迟
                if let lineString = String(data: line, encoding: .utf8)?.trimmingCharacters(in: .newlines) {
                    continuation.yield(lineString)
                }
            }
            continuation.finish()
        }
    }
}

Task {
    let url = URL(fileURLWithPath: "sample.txt")
    for await line in lines(from: url) {
        print("行：\(line)")
    }
}
  • 假设sample.txt包含多行文本。
  • 按需读取，避免内存压力。

3. 分页API加载

连续加载分页数据：

struct PageSequence: AsyncSequence {
    typealias Element = [String]
    let baseURL: String
    let pageSize: Int
    
    func makeAsyncIterator() -> PageIterator {
        PageIterator(baseURL: baseURL, pageSize: pageSize)
    }
}

struct PageIterator: AsyncIteratorProtocol {
    let baseURL: String
    let pageSize: Int
    private var page = 0
    
    mutating func next() async throws -> [String]? {
        let url = URL(string: "\(baseURL)?page=\(page)&size=\(pageSize)")!
        let (data, _) = try await URLSession.shared.data(from: url)
        let items = try JSONDecoder().decode([String].self, from: data)
        guard !items.isEmpty else { return nil }
        page += 1
        return items
    }
}

Task {
    let sequence = PageSequence(baseURL: "https://api.example.com/items", pageSize: 2)
    do {
        for try await page in sequence {
            print("页面数据：\(page)")
        }
    } catch {
        print("加载失败：\(error)")
    }
}
  • 模拟分页API，每次返回一页数据。
  • 空页结束循环。

进阶技巧

  1. 中断循环
    使用break提前退出：

    for await item in sequence {
    if item == "停止" { break }
    print(item)
}

  2. 结合集合操作
    AsyncSequence支持map、filter等（需扩展支持）：

    let filtered = sequence.filter { await $0.count > 5 }

  3. 超时控制
    用withTimeout（假设自定义）限制等待：

    for await item in sequence.withTimeout(seconds: 10) { ... }

小结

for await将AsyncSequence的异步元素遍历转化为直观的循环形式，适用于实时数据、文件流和分页加载等场景。本节通过示例展示了其基本用法和实际应用，强调了错误处理和线程管理的注意点。掌握for await，你将能轻松处理动态数据流。下一节将探讨AsyncStream的自定义实现，进一步扩展你的异步序列技能。

内容说明

  • 结构：从语法到错误处理，再到场景和技巧，最后总结。
  • 代码：包含实时流、文件读取和分页示例，突出多样性。
  • 语气：讲解性且实践导向，适合技术书籍深入章节。
  • 衔接：承接前节（AsyncSequence简介），预告后续（AsyncStream）。
Last Updated::