深入理解 Swift SE-0388：AsyncStream.makeStream 便捷工厂方法的设计与实战

文档【免费下载链接】swift-evolutionThis maintains proposals for changes and user-visible enhancements to the Swift Programming Language.项目地址https://gitcode.com/gh_mirrors/sw/swift-evolution点击查看免费下载导读SE-0388Convenience Async[Throwing]Stream.makeStream methods为 Swift 标准库的AsyncStream与AsyncThrowingStream新增了静态工厂方法makeStream(of:bufferingPolicy:)让开发者可以一步同时获得「流」与其「续体Continuation」彻底告别从初始化闭包中逃逸续体的隐式解包可选IUO样板代码。本文以该提案为主体结合仓库中 SE-0314 AsyncStream 原始提案 与 SE-0406 背压演进提案 的源码级依据讲解其动机、API 设计、实现细节、缓冲策略参数与兼容性约束读完后你可以在 Swift 5.9 中直接写出更安全、更简洁的生产者/消费者并发代码。一、背景AsyncStream 与 Continuation 的角色划分AsyncStream与AsyncThrowingStream由 SE-0314 引入Swift 5.5 实现是标准库提供的「根级 AsyncSequence」类型。它们的作用是把基于回调callback或委托delegate的多次异步产出桥接进async/await世界例如QuakeMonitor.quakeHandler这种每次地震事件回调一次、可被持续调用的接口就可以包装成AsyncStreamQuake供for await消费。SE-0314 明确设计了两个互补的角色外层AsyncStreamElement消费端对外呈现AsyncSequence接口通过for await或迭代器的next()逐个取回元素内层AsyncStream.Continuation生产端Sendable类型可从任意并发上下文调用yield(_:)产出值、调用finish()结束序列且可被 yield 多次。两个角色之间的桥梁就是初始化器接收的那个build闭包——续体通过闭包参数被交给生产者。SE-0314 的原始 API 签名如下节选自 proposals/0314-async-stream.mdpublic init( _ elementType: Element.Type Element.self, bufferingPolicy limit: Continuation.BufferingPolicy .unbounded, _ build: (Continuation) - Void )二、痛点续体必须从闭包中逃逸出来在实际使用中一个常见的场景是把续体和流分别交给不同的位置续体交给生产者如网络层、事件源流交给消费者如 UI 层。这要求把闭包参数里的Continuation逃逸escape出初始化闭包而 SE-0388 提案指出这种写法存在三个层面的不便必须借助隐式解包可选IUO由于闭包在初始化完成后才被调用逃逸前只能用var cont: AsyncStreamInt.Continuation!占位触发送达性Sendability警告逃逸后需要再拷贝到let常量里才能消除警告语义误导闭包结构暗示「续体的生命周期被限定在闭包作用域内」而实际并非如此——续体需要长期存活直到流被 finish 或取消。提案给出了改造前的典型写法完整示例见 proposals/0388-async-stream-factory.mdvar cont: AsyncStreamInt.Continuation! let stream AsyncStreamInt { cont $0 } // We have to assign the continuation to a let to avoid sendability warnings let continuation cont await withTaskGroup(of: Void.self) { group in group.addTask { for i in 0...9 { continuation.yield(i) } continuation.finish() } group.addTask { for await i in stream { print(i) } } }这段代码虽然能运行但cont!这种先占位、后填充的舞步极易出错也让续体生命周期与闭包无关这一事实变得模糊。三、方案新增makeStream静态工厂方法SE-0388 的解决方案是在AsyncStream和AsyncThrowingStream上各新增一个静态方法makeStream一次性返回「流 续体」二元组。同样是上面的任务组示例新写法变成了let (stream, continuation) AsyncStream.makeStream(of: Int.self) await withTaskGroup(of: Void.self) { group in group.addTask { for i in 0...9 { continuation.yield(i) } continuation.finish() } group.addTask { for await i in stream { print(i) } } }对比之下makeStream方案消除了 IUO、消除了 Sendability 警告、也让生产端与消费端各持一端的意图一目了然。提案状态为Implemented (Swift 5.9)即从 Swift 5.9 起可用仓库 README.md 的发布记录同样确认 Swift 5.9 于 2023-09 发布。四、详细设计完整的 API 签名与实现4.1 AsyncStream 版本available(SwiftStdlib 5.1, *) extension AsyncStream { /// Initializes a new AsyncStream and an AsyncStream/Continuation. /// /// - Parameters: /// - elementType: The element type of the stream. /// - limit: The buffering policy that the stream should use. /// - Returns: A tuple containing the stream and its continuation. The continuation should be passed to the /// producer while the stream should be passed to the consumer. backDeployed(before: SwiftStdlib 5.9) public static func makeStream( of elementType: Element.Type Element.self, bufferingPolicy limit: Continuation.BufferingPolicy .unbounded ) - (stream: AsyncStreamElement, continuation: AsyncStreamElement.Continuation) { var continuation: AsyncStreamElement.Continuation! let stream AsyncStreamElement(bufferingPolicy: limit) { continuation $0 } return (stream: stream, continuation: continuation!) } }4.2 AsyncThrowingStream 版本available(SwiftStdlib 5.1, *) extension AsyncThrowingStream { /// Initializes a new AsyncThrowingStream and an AsyncThrowingStream/Continuation. /// /// - Parameters: /// - elementType: The element type of the stream. /// - failureType: The failure type of the stream. /// - limit: The buffering policy that the stream should use. /// - Returns: A tuple containing the stream and its continuation. The continuation should be passed to the /// producer while the stream should be passed to the consumer. backDeployed(before: SwiftStdlib 5.9) public static func makeStream( of elementType: Element.Type Element.self, throwing failureType: Failure.Type Failure.self, bufferingPolicy limit: Continuation.BufferingPolicy .unbounded ) - (stream: AsyncThrowingStreamElement, Failure, continuation: AsyncThrowingStreamElement, Failure.Continuation) where Failure Error { var continuation: AsyncThrowingStreamElement, Failure.Continuation! let stream AsyncThrowingStreamElement, Failure(bufferingPolicy: limit) { continuation $0 } return (stream: stream, continuation: continuation!) } }4.3 设计要点逐项解读参数of elementType元素类型默认值Element.self因此最常见的AsyncStream.makeStream(of: Int.self)与全默认调用AsyncStream.makeStream()都合法借助类型推断let (stream, cont) AsyncStreamInt.makeStream()也可以省略of:参数。参数throwing failureType仅 Throwing 版本失败类型约束where Failure Error这与 SE-0314 中AsyncThrowingStream.init只允许Failure Error的构造约束保持一致见 proposals/0314-async-stream.md。参数bufferingPolicy limit缓冲策略默认.unbounded与 SE-0314 初始化器的默认行为一致。其合法取值定义在 SE-0314 的Continuation.BufferingPolicy中见 proposals/0314-async-stream.md取值行为.unbounded无界缓冲所有未被消费的 yield 值都先入缓冲区默认值.bufferingOldest(Int)缓冲区满时丢弃新到达的元素保证保留最旧的 n 个值.bufferingNewest(Int)缓冲区满时丢弃最旧的元素保证保留最新的 n 个值需要指出AsyncStream的缓冲区只为尚未被迭代消费的值服务如果缓冲容量为 0则当没有任务正在await迭代器的next()时yield 的值会直接被丢弃SE-0314 的 dropping 行为。yield的返回值YieldResultenqueued(remaining:)/dropped(Element)/terminated正是对这三种情况的显式回报。返回值是带标签的元组(stream:..., continuation:...)这正是评审后从具体类型改为元组的关键设计详见下文备选方案分析。实现技巧方法内部依然使用 IUO——var continuation: ...!配合初始化闭包{ continuation $0 }完成填充再用continuation!解包返回。也就是说SE-0388 不是消灭了 IUO而是把 IUO 从用户代码收拢进标准库内部用户侧不再接触任何强制解包。注解组合方法整体以available(SwiftStdlib 5.1, *)标记沿用 SE-0314 类型自身的可用性同时以backDeployed(before: SwiftStdlib 5.9)标记使得该 API 可以向后部署到旧版本 Swift 标准库上运行详见第七节。五、实战生产者/消费者模式的三种典型写法5.1 无抛错流AsyncStreamlet (stream, continuation) AsyncStream.makeStream(of: Int.self) await withTaskGroup(of: Void.self) { group in group.addTask { for i in 0...9 { continuation.yield(i) } continuation.finish() } group.addTask { for await i in stream { print(i) } } }5.2 可抛错流AsyncThrowingStreamSE-0314 曾给出过一个把回调式买菜接口桥接为流的例子见 proposals/0314-async-stream.md用makeStream改写后生产端可以直接把续体交给回调接口无需闭包嵌套let (stream, continuation) AsyncThrowingStream.makeStream(of: Vegetable.self) buyVegetables( shoppingList: list, onGotVegetable: { veggie in continuation.yield(veggie) }, onAllVegetablesFound: { continuation.finish() }, onNonVegetable: { error in continuation.finish(throwing: error) } ) for try await veggie in stream { // 消费蔬菜 }注意AsyncThrowingStream的消费需要使用for try await因为迭代器next()可能抛出错误finish(throwing:)传入的错误会被迭代器原样抛出而finish()则表示正常结束。5.3 处理任务取消与资源清理makeStream返回的续体同样支持onTermination回调。SE-0314 规定见 proposals/0314-async-stream.md当迭代结束、流离开作用域或所在任务被取消时会触发onTermination其中Termination区分.finished与.cancelled两种终态。配合makeStream可写成let (stream, continuation) AsyncStream.makeStream(of: Quake.self) continuation.onTermination { termination in switch termination { case .finished: monitor.stopMonitoring() case .cancelled: monitor.stopMonitoring() } } monitor.quakeHandler { quake in continuation.yield(quake) } monitor.startMonitoring() for await quake in stream { // 处理地震事件 }六、从源码结构看为什么返回元组而不是具体类型提案在 Alternatives considered 一节完整记录了设计权衡这一决策过程对理解 API 形态至关重要。6.1 元组 vs 具体类型提案者最初的 pitch 使用元组作为工厂的返回类型评审前又改成了具体类型理由是便于写文档注释但在正式评审中多数反馈倾向于元组方案最终接受时改回元组见提案末尾 Revision historyAfter review: Changed the return type from a concrete type to a tuple。元组方案的两个主要收益引导解构destructuringlet (stream, continuation) ...的写法自然地把两个值拆开分别交给生产者与消费者各自持有——这正是期望的使用方式支持向后部署back deployment具体类型方案无法同样轻量地实现backDeployed。6.2 为什么不在init中直接传入续体pitch 阶段有人提议让用户直接把一个续体传给AsyncStreamElement.init()该方案被否决原因有二同一个续体可能被传入多个流造成一对多、语义混乱未被任何流持有的续体毫无用处。SE-0388 的结论是AsyncStream.Continuation与某个AsyncStream实例深度耦合一一对应因此 API 应当把这种耦合关系直接表达出来从类型层面杜绝误用。6.3 为什么不什么都不做提案也考虑了维持现状的选项但认为既然AsyncStream属于标准库就应该提供一个更体面的方式来同时创建流与续体而非要求用户手写 IUO 逃逸样板。七、兼容性与演进source compatibility、ABI 与 API resilienceSE-0388 对兼容性的影响全部是纯增量的Source compatibility新增静态方法不修改任何既有声明对源码完全无影响Effect on ABI stability仅引入新的并发库 ABImakeStream方法本身不影响既有声明的 ABIEffect on API resilience无影响既有弹性模型允许添加静态方法向后部署backDeployed(before: SwiftStdlib 5.9)意味着在 Swift 5.9 之前的标准库运行时上编译器会在调用点内联该方法的实现使旧系统也能获得这一便捷 API。八、后续演进SE-0406 中带背压的makeStream变体makeStream的工厂形态并非终点。仓库中的 SE-0406 AsyncStream backpressure 在其上进一步演进引入makeStream(of:backpressureStrategy:)返回(stream, source)source提供write(contentsOf:)、enqueueCallback等带背压的写入 API支持.watermark(low:high:)等策略见 proposals/0406-async-stream-backpressure.mdlet (stream, source) AsyncStream.makeStream( of: Int.self, backpressureStrategy: .watermark(low: 2, high: 4) )SE-0406 还明确将其定位为严格单播strict unicast异步序列并讨论了makeStream方法中提供onTerminate回调的方案。这说明「一次性返回流 生产端」这一工厂模式已成为 Swift 异步流 API 的标准骨架后续演进均在此基础上叠加能力而 SE-0388 正是这一模式的奠基者。九、总结SE-0388 的makeStream以极小的 API 面两个静态方法解决了AsyncStream使用中最常见的痛点续体逃逸样板。它把隐式解包可选收敛进标准库实现、用元组返回引导正确的解构语义、并以backDeployed实现跨版本可用。在 Swift 5.9 及以后let (stream, continuation) AsyncStream.makeStream(of: Element.self)应成为所有生产者/消费者桥接代码的默认起点若你需要更精细的背压控制则可进一步关注 SE-0406 的演进方案。相关完整资料可继续阅读仓库内的 SE-0388 原文、SE-0314 AsyncStream 设计 与 SE-0406 背压演进。赞分享文档【免费下载链接】swift-evolutionThis maintains proposals for changes and user-visible enhancements to the Swift Programming Language.项目地址https://gitcode.com/gh_mirrors/sw/swift-evolution点击查看免费下载相关推荐Swift Clock 的纪元Epoch体系深入解读 SE-0473 的 systemEpoch 设计Swift Clock 的纪元Epoch体系深入解读 SE 0473 的 systemEpoch 设计 导读 SE 0473《Clock Epochs》文档DDD示例工厂深入理解领域驱动设计的实战之旅DDD示例工厂深入理解领域驱动设计的实战之旅 项目介绍 DDDDomain Driven Design领域驱动设计通过强调业务领域和软件开发之间的紧密结Swift 标准库演进解读SE-0218 Dictionary.compactMapValues 的设计、实现与实战Swift 标准库演进解读SE 0218 Dictionary.compactMapValues 的设计、实现与实战 本文以 Swift Evolution文档创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考