前言

AsyncThrowingStream 和 AsyncStream 是 Swift 5.5 中由 SE-314 引入的并发框架的一部分。异步流允许你替换基于闭包或 Combine 发布器的现有代码。

在深入研究围绕抛出流的细节之前，如果你还没有阅读我的文章，我建议你先阅读我的文章，内容包括async-await。本文解释的大部分代码将使用那里解释的API。

什么是 AsyncThrowingStream？

AsyncThrowingStream 符合 AsyncSequence 协议，提供了一种不需要手动实现异步迭代器就能创建异步序列的便利方法。异步流适用于将基于回调或委托的API适配为可以与async-await一起使用的方式。

与 AsyncStream 相比，这种类型可以从 awaited next() 中抛出错误，从而用抛出的错误终止流。

您可以使用闭包初始化 AsyncThrowingStream，该闭包接收 AsyncThrowingStream.Continuation。在此闭包中生成元素，然后通过调用 continuation的yield(_:) 方法将它们提供给流。当没有更多元素可生成时，请调用 continuation的finish() 方法。这会导致序列迭代器生成nil，从而终止序列。如果发生错误，请调用 continuation的finish(throwing:) 方法，这会导致迭代器的 next() 方法将错误抛出到等待调用点。continuation 是 Sendable 的，这允许从 AsyncThrowingStream 的迭代之外的并发上下文中调用它。

任意的元素来源可能会比调用者迭代它们的速度更快地生成元素。因此，AsyncThrowingStream 定义了缓冲行为，允许流缓冲特定数量的最旧或最新元素。默认情况下，缓冲限制为 Int.max，这意味着它是无界的。

调整现有代码以使用流

将现有的回调代码适配为使用 async-await，请使用回调将值提供给流，方法是使用 continuation的yield(_:) 方法。

考虑一个假想的 QuakeMonitor 类型，每当它检测到地震时，它会向调用者提供 Quake 实例。为了接收回调，调用者将自定义闭包设置为监视器的 quakeHandler 属性的值，监视器会根据需要回调该闭包。调用者还可以设置 errorHandler 来接收异步错误通知，例如监视器服务突然不可用。

class QuakeMonitor {
    var quakeHandler: ((Quake) -> Void)?
    var errorHandler: ((Error) -> Void)?

    func startMonitoring() {…}
    func stopMonitoring() {…}
}

为了适应使用async-await，请扩展QuakeMonitor以添加一个类型为AsyncThrowingStream的quakes属性。在此属性的getter中，返回一个AsyncThrowingStream，其build闭包 - 在运行时调用以创建流 - 使用continuation执行以下步骤：

1. 创建一个 QuakeMonitor 实例。
2. 将监视器的 quakeHandler 属性设置为一个闭包，该闭包接收每个 Quake 实例，并通过调用 continuation的yield(_:) 方法将其转发到流中。
3. 将监视器的 errorHandler 属性设置为一个闭包，该闭包接收来自监视器的任何错误，并通过调用 continuation 的 finish(throwing:) 方法将其转发到流中。这会导致流的迭代器抛出错误并终止流。
4. 将 continuation 的 onTermination 属性设置为一个闭包，该闭包在监视器上调用 stopMonitoring()。
5. 在 QuakeMonitor 上调用 startMonitoring()。

extension QuakeMonitor {

    static var throwingQuakes: AsyncThrowingStream<Quake, Error> {
        AsyncThrowingStream { continuation in
            let monitor = QuakeMonitor()
            monitor.quakeHandler = { quake in
                 continuation.yield(quake)
            }
            monitor.errorHandler = { error in
                continuation.finish(throwing: error)
            }
            continuation.onTermination = { @Sendable _ in
                monitor.stopMonitoring()
            }
            monitor.startMonitoring()
        }
    }
}

因为流是AsyncSequence，调用点使用for-await-in语法来处理流产生的每个Quake实例：

do {
    for try await quake in quakeStream {
        print("Quake: \(quake.date)")
    }
    print("Stream done.")
} catch {
    print("Error: \(error)")
}

什么是 AsyncStream?

AsyncStream 类似于抛出的变体，但绝不会导致抛出错误。一个非抛出型的异步流会根据明确的完成调用或流的取消而完成。

注意： 在这篇文章中，我们将解释如何使用AsyncThrowingStream。除了发生错误处理的部分，代码示例与AsyncStream类似。

AsyncThrowingStream

如何使用 AsyncThrowingStream

AsyncThrowingStream 可以很好地替代现有的基于闭包的代码，如进度和完成处理程序。为了更好地理解我的意思，我将向你介绍我们在 WeTransfer 应用程序中遇到的一个场景。

在我们的应用程序中，我们有一个基于闭包的现有类，叫做 FileDownloader：

struct FileDownloader {
    enum Status {
        case downloading(Float)
        case finished(Data)
    }

    func download(_ url: URL, progressHandler: (Float) -> Void, completion: (Result<Data, Error>) -> Void) throws {
        // .. Download implementation
    }
}

文件下载器接受一个URL，报告进度情况，并完成一个包含下载数据的结果或在失败时显示一个错误。

文件下载器在文件下载过程中报告一个数值流。在这种情况下，它报告的是一个状态值流，以报告正在运行的下载的当前状态。FileDownloader 是一个完美的例子，你可以重写一段代码来使用 AsyncThrowingStream。然而，重写需要你在实现层面上也重写你的代码，所以让我们定义一个重载方法来代替：

extension FileDownloader {
    func download(_ url: URL) -> AsyncThrowingStream<Status, Error> {
        return AsyncThrowingStream { continuation in
            do {
                try self.download(url, progressHandler: { progress in
                    continuation.yield(.downloading(progress))
                }, completion: { result in
                    switch result {
                    case .success(let data):
                        continuation.yield(.finished(data))
                        continuation.finish()
                    case .failure(let error):
                        continuation.finish(throwing: error)
                    }
                })
            } catch {
                continuation.finish(throwing: error)
            }
        }
    }
}

正如你所看到的，我们把下载方法包裹在一个 AsyncThrowingStream 里面。我们将流的值 Status 的类型描述为一个通用的类型，允许我们用状态更新来延续流。

只要有错误发生，我们就会通过抛出一个错误来完成流。在完成处理程序的情况下，我们要么通过抛出一个错误来完成，要么用一个不抛出的完成回调来跟进数据的产生。

switch result {
case .success(let data):
    continuation.yield(.finished(data))
    continuation.finish()
case .failure(let error):
    continuation.finish(throwing: error)
}

在收到最后的状态更新后，不要忘记 finish() 回调，这一点至关重要。否则，我们将保持流的存活，而实现层面的代码将永远不会继续。

我们可以通过使用另一个 yield 方法来重写上述代码，接受一个 Result 枚举作为参数：

continuation.yield(with: result.map { .finished($0) })
continuation.finish()

重写后的代码简化了我们的代码，并去掉了 switch-case 代码。我们必须映射我们的 Reslut 枚举以匹配预期的 Status 值。如果我们产生一个失败的结果，我们的流将在抛出包含的错误后结束。

AsyncThrowingStream 迭代

一旦你配置好你的异步抛出流，你就可以开始在数值流上进行迭代。在我们的 FileDownloader 例子中，它将看起来如下所示：

do {
    for try await status in download(url) {
        switch status {
        case .downloading(let progress):
            print("Downloading progress: \(progress)")
        case .finished(let data):
            print("Downloading completed with data: \(data)")
        }
    }
    print("Download finished and stream closed")
} catch {
    print("Download failed with \(error)")
}

我们处理任何状态的更新，并且我们可以使用 catch 闭包来处理任何发生的错误。你可以使用基于 AsyncSequence 接口的 for ... in 循环进行迭代，这对 AsyncStream 来说是一样的。

如果你遇到了类似的编译错误:

‘async’ in a function that does not support concurrency

你可能想读一读我的文章，其中Swift 中的 async/await ——代码实例详解。

上述代码示例中的打印语句有助于你理解 AsyncThrowingStream 的生命周期。你可以替换打印语句来处理进度更新和处理数据，为你的用户实现可视化。

调试 AsyncStream

如果一个流不能报告数值，我们可以通过放置断点来调试流产生的回调。虽然也可能是上面的 “Download finished and stream closed” 的打印语句不会调用，这意味着你在实现层的代码永远不会继续。后者可能是一个未完成的流的结果。

为了验证，我们可以利用 onTermination 回调：

func download(_ url: URL) -> AsyncThrowingStream<Status, Error> {
    return AsyncThrowingStream { continuation in

        ///  配置一个终止回调，以了解你的流的生命周期。
        continuation.onTermination = { @Sendable status in
            print("Stream terminated with status \(status)")
        }

        // ..
    }
}

回调在流终止时被调用，它将告诉你你的流是否还活着。我推荐你阅读 Sendable 和 @Sendable 闭包代码实例详解 来理解 @Sendable 属性。

如果出现了错误，输出结果可能如下:

Stream terminated with status finished(Optional(FileDownloader.FileDownloadingError.example))

上述输出只有在使用 AsyncThrowingStream 时才能实现。如果是一个普通的 AsyncStream，完成的输出看起来如下：

Stream terminated with status finished

而取消的结果对这两种类型的流来说都是这样的:

Stream terminated with status cancelled

你也可以在流结束后使用这个终止回调进行任何清理。例如，删除任何观察者或在文件下载后清理磁盘空间。

取消一个 AsyncStream

一个 AsyncStream 或 AsyncThrowingStream 可以由于一个封闭的任务被取消而取消。一个例子可以如下：

let task = Task.detached {
    do {
        for try await status in download(url) {
            switch status {
            case .downloading(let progress):
                print("Downloading progress: \(progress)")
            case .finished(let data):
                print("Downloading completed with data: \(data)")
            }
        }
    } catch {
        print("Download failed with \(error)")
    }
}
task.cancel()

一个流在超出范围或包围的任务取消时就会取消。如前所述，取消将相应地触发 onTermination 回调。

结论

AsyncThrowingStream 或 AsyncStream 是重写基于闭包的现有代码到支持 async-awai t的替代品的好方法。你可以提供一个连续的值流，并在成功或失败时完成一个流。你可以使用基于 AsyncSequence APIs 的 for 循环在实现层面上迭代值。