Swift 引入 async/await 之后，异步操作的组合方式发生了明显变化。过去需要闭包回调或 Combine 订阅才能处理的数据流，现在可以通过 AsyncSequence 协议统一成可迭代的序列。只要一个类型遵循 AsyncSequence，就可以在 for await 循环中逐个获取元素，循环内部还能直接 await 其他异步函数。这种写法让异步数据流的消费逻辑变得和同步数组遍历一样直观。

AsyncSequence 协议要解决的问题与协议结构

同步的 Sequence 协议在 Swift 标准库中已经存在多年，for-in 循环依赖它来遍历数组、字典等集合。但是同步序列有一个天生的限制：它的 next 方法是一个同步函数，调用后必须立即返回下一个元素或者 nil。一旦数据源是异步产生的，比如网络响应、数据库查询结果、传感器采样值，同步序列就无法胜任，因为获取下一个元素可能需要等待一段时间。

AsyncSequence 协议正是为这个场景设计的。它把迭代器的 next 方法改成异步可抛出的版本，返回类型仍然是可选元素，但调用方需要等待异步操作完成。协议的核心定义大致如下：

public protocol AsyncSequence {
    associatedtype Element
    associatedtype AsyncIterator: AsyncIteratorProtocol where AsyncIterator.Element == Element
    __consuming func makeAsyncIterator() -> AsyncIterator
}
public protocol AsyncIteratorProtocol {
    associatedtype Element
    mutating func next() async throws -> Element?
}

可以看到，与 Sequence 相比，唯一的本质区别就是 next 方法多了 async throws 修饰。这意味着实现一个自定义 AsyncSequence 时，主要工作是提供一个异步迭代器，并在 next 中完成数据获取或等待。例如下面这个简单的倒计时序列，每次调用 next 都会等待 100 毫秒再返回一个数：

struct Countdown: AsyncSequence {
    typealias Element = Int
    var start: Int
    struct AsyncIterator: AsyncIteratorProtocol {
        var current: Int
        mutating func next() async throws -> Int? {
            guard current > 0 else { return nil }
            try await Task.sleep(nanoseconds: 100_000_000)
            let value = current
            current -= 1
            return value
        }
    }
    func makeAsyncIterator() -> AsyncIterator {
        AsyncIterator(current: start)
    }
}

这个例子虽然简单，但揭示了异步序列的核心逻辑：next 方法可以在内部执行 await，把控制权交还给并发运行时，等数据就绪后再恢复。因此任何异步数据源只要能够抽象成一个 next 方法，就可以包装成 AsyncSequence。

for await 循环的写法与执行细节

有了 AsyncSequence，消费数据最自然的方式就是使用 for await 循环。它的语法和 for-in 类似，只是把 in 换成 await in，表示每次迭代都可能挂起。如果序列本身不抛出错误，例如 AsyncStream，直接使用 for await 即可。下面代码创建了一个生成 1 到 5 的异步流并逐个打印：

let stream = AsyncStream<Int> { continuation in
    for i in 1...5 {
        continuation.yield(i)
    }
    continuation.finish()
}
Task {
    for await number in stream {
        print("收到 \(number)")
    }
    print("流结束")
}

for await 循环会在每次迭代前调用迭代器的 next 方法。如果 next 返回一个元素，循环体执行；如果返回 nil，循环正常退出；如果 next 抛出错误，循环会中断并把错误向上传播，此时语法上必须使用 for try await。AsyncStream 是标准库提供的最常用的 AsyncSequence 实现，它的 continuation 允许在闭包内随时向流中写入元素，并在结束时调用 finish 方法通知消费者。

这里要特别留意取消行为。当包裹 for await 循环的 Task 被取消时，AsyncStream 会自动响应取消信号，让当前的 next 调用结束并退出循环。这意味着你通常不需要在循环体内手动检查 Task.isCancelled，底层机制已经帮你处理了。但如果自定义 AsyncSequence 的迭代器不响应取消，for await 循环就可能一直挂起，所以自己实现时应该尽量在 next 中检查取消状态或者依赖支持取消的 API。

错误处理的另一个关键是区分 for await 和 for try await。AsyncStream 本身不会抛出错误，所以使用 for await 即可。如果数据源可能失败，比如从磁盘读取文件时遇到 IO 错误，就应当使用 AsyncThrowingStream，并且消费端写成 for try await，外层用 do/catch 捕获异常。

基于 AsyncStream 构建一个线程安全的 Channel

在并发编程中，Channel 是一种常见的消息传递原语，允许一个或多个生产者向一个或多个消费者发送数据。Swift 标准库没有直接提供名为 Channel 的类型，但我们可以利用 AsyncStream 的 continuation 机制自己封装一个。设计思路非常简单：创建一个 AsyncStream，同时保存它的 continuation，在外部通过 send 方法向 continuation 写入元素，stream 本身则暴露给消费方使用。

下面的代码实现了一个线程安全的 Channel 类，支持自定义缓冲区策略，默认使用无限缓冲：

final class Channel<Element> {
    private let continuation: AsyncStream<Element>.Continuation
    let stream: AsyncStream<Element>
    init(bufferingPolicy: AsyncStream<Element>.Continuation.BufferingPolicy = .unbounded) {
        var capturedContinuation: AsyncStream<Element>.Continuation!
        stream = AsyncStream<Element>(bufferingPolicy: bufferingPolicy) { continuation in
            capturedContinuation = continuation
        }
        self.continuation = capturedContinuation
    }
    func send(_ element: Element) {
        continuation.yield(element)
    }
    func close() {
        continuation.finish()
    }
}

使用这个 Channel 时，生产者只需要调用 send 方法，消费者对 stream 执行 for await 循环。下面演示了一个经典的生产者-消费者模型，一个 Task 负责发送数据，另一个 Task 负责接收并打印：

let channel = Channel<Int>()
let consumer = Task {
    for await value in channel.stream {
        print("消费者收到 \(value)")
    }
    print("通道关闭，消费者结束")
}
channel.send(1)
channel.send(2)
channel.send(3)
channel.close()
await consumer.value

这里有一个值得注意的细节：AsyncStream.Continuation 是 Sendable 的，因此你可以在多个并发任务中安全地调用 send 方法，不会引发数据竞争。不过，多个生产者同时调用 send 时，元素的写入顺序并不保证与调用顺序完全一致。如果你的业务对消息顺序有严格要求，可以在 send 内部增加一个锁或者用 actor 来串行化写入。

缓冲区策略对 Channel 的行为影响很大。默认的 unbounded 策略表示缓冲区可以无限增长，生产者在 send 时永远不会被阻塞，但如果消费者处理速度跟不上，内存占用会持续上升。AsyncStream 还提供了 bufferingNewest 和 bufferingOldest 两种有限缓冲策略，前者在缓冲区满时丢弃最旧的元素，后者丢弃最新的元素。选择合适的策略可以为系统引入背压，避免消息堆积导致内存问题。

AsyncThrowingStream 与错误传递

实际开发中，异步数据流往往不只是正常结束，还可能因为网络故障、文件损坏或人为取消而异常终止。如果仍然使用 AsyncStream，错误就无法传递给消费者，只能通过旁路状态或者提前结束流来模拟。AsyncThrowingStream 正是为了解决这个问题，它允许迭代器在产生元素的过程中抛出错误，从而让 for try await 循环捕获并处理异常。

下面这个例子模拟了一个异步任务，在成功发送三个数字后抛出一个错误：

enum ChannelError: Error {
    case cancelled
}
let throwingStream = AsyncThrowingStream<Int, Error> { continuation in
    Task {
        for i in 1...3 {
            continuation.yield(i)
            try await Task.sleep(nanoseconds: 100_000_000)
        }
        continuation.finish(throwing: ChannelError.cancelled)
    }
}
Task {
    do {
        for try await value in throwingStream {
            print("值: \(value)")
        }
    } catch {
        print("流异常终止: \(error)")
    }
}

消费 AsyncThrowingStream 时必须使用 for try await，否则编译器会报错，因为迭代器的 next 方法是可抛出的。错误一旦被抛出，循环立即结束，之前未消费的元素也不会再被读取。如果序列正常结束，可以调用 finish 而不带参数，这样循环会自然退出，不会进入 catch 块。

取消行为与 AsyncStream 类似，AsyncThrowingStream 也会在包裹循环的 Task 被取消时自动中断当前 next 调用。对于已经缓冲但尚未消费的元素，取消后不会再被交付。因此在设计 Channel 时，你需要明确取消和关闭的区别：关闭是正常流程的一部分，取消则是外部强制终止，两者在流语义上可能有所不同。

AsyncSequence 与 Combine 的对比及实践建议

Combine 框架中的 Publisher 同样可以处理异步数据流，但它采用推模型：发布者主动向订阅者推送值，订阅者被动接收。AsyncSequence 则采用拉模型：消费者通过 for await 循环主动请求下一个元素。拉模型的一个明显优势是天然支持背压，消费者处理不完时可以暂停请求，而 Combine 需要额外的背压策略或者使用 buffer 操作符来平衡。此外，AsyncSequence 与 Task 的协作取消集成得更好，一旦任务取消，循环会自动退出，不需要手动管理订阅生命周期。

不过两者并非互斥。很多场景下，你可以把 Combine 发布者转换成 AsyncSequence，或者把 AsyncSequence 包装成发布者。Swift 并发生态中还提供了 swift-async-algorithms 这样的补充库，里面包含了很多实用的异步序列算法，例如合并、去抖、分组等。这些工具可以进一步简化复杂的异步流处理。

在实际项目中，如果数据源是一次性的、需要逐个消费的流，比如文件逐行读取、网络分块下载、传感器连续采样，使用 AsyncSequence 配合 for await 循环通常是最清晰的方案。如果数据流需要一对多广播，你需要注意 AsyncStream 默认是单消费者的，多个 for await 同时消费同一个 stream 会导致元素被竞争性获取。要实现多播，需要自己构建广播结构或者使用专门的异步通道类型。最后提醒一点，for await 循环体内尽量避免执行同步阻塞调用，应该使用 await 版本的 API 来保持并发性能。