草庐IT

Swift AsyncThrowingStream 和 AsyncStream 代码实例详解

Swift君 2023-03-28 原文

前言

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

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

什么是 AsyncThrowingStream？

你可以把 AsyncThrowingStream 看作是一个有可能导致抛出错误的元素流。他的值随着时间的推移而传递，流可以通过一个结束事件来关闭。一旦发生错误，结束事件既可以是成功，也可以是失败。

什么是 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

上述代码示例中的打印语句有助于你理解 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)")
}

// ..
}
}

回调在流终止时被调用，它将告诉你你的流是否还活着。

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

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 循环在实现层面上迭代值。

参考资料

[1]SE-314: ​​https://github.com/apple/swift-evolution/blob/main/proposals/0314-async-stream.md。​​

AsyncThrowingStreamAsyncStreamspanstylecolor移动开发iOS$Swift元素流

有关Swift AsyncThrowingStream 和 AsyncStream 代码实例详解的更多相关文章

  1. ruby - 如何在 buildr 项目中使用 Ruby 代码？ - 2

    如何在buildr项目中使用Ruby？我在很多不同的项目中使用过Ruby、JRuby、Java和Clojure。我目前正在使用我的标准Ruby开发一个模拟应用程序，我想尝试使用Clojure后端(我确实喜欢功能代码)以及JRubygui和测试套件。我还可以看到在未来的不同项目中使用Scala作为后端。我想我要为我的项目尝试一下buildr(http://buildr.apache.org/)，但我注意到buildr似乎没有设置为在项目中使用JRuby代码本身!这看起来有点傻，因为该工具旨在统一通用的JVM语言并且是在ruby中构建的。除了将输出的jar包含在一个独特的、仅限ruby​​

  2. ruby-on-rails - Rails 源代码 : initialize hash in a weird way? - 2

    在rails源中:https://github.com/rails/rails/blob/master/activesupport/lib/active_support/lazy_load_hooks.rb可以看到以下内容@load_hooks=Hash.new{|h,k|h[k]=[]}在IRB中，它只是初始化一个空哈希。和做有什么区别@load_hooks=Hash.new 最佳答案 查看rubydocumentationforHashnew→new_hashclicktotogglesourcenew(obj)→new_has

  3. ruby-on-rails - 如何使用 instance_variable_set 正确设置实例变量？ - 2

    我正在查看instance_variable_set的文档并看到给出的示例代码是这样做的:obj.instance_variable_set(:@instnc_var,"valuefortheinstancevariable")然后允许您在类的任何实例方法中以@instnc_var的形式访问该变量。我想知道为什么在@instnc_var之前需要一个冒号:。冒号有什么作用？ 最佳答案 我的第一直觉是告诉你不要使用instance_variable_set除非你真的知道你用它做什么。它本质上是一种元编程工具或绕过实例变量可见性的黑客攻击

  4. ruby 正则表达式 - 如何替换字符串中匹配项的第 n 个实例 - 2

    在我的应用程序中，我需要能够找到所有数字子字符串，然后扫描每个子字符串，找到第一个匹配范围(例如5到15之间)的子字符串，并将该实例替换为另一个字符串“X”。我的测试字符串s="1foo100bar10gee1"我的初始模式是1个或多个数字的任何字符串，例如，re=Regexp.new(/\d+/)matches=s.scan(re)给出["1","100","10","1"]如果我想用“X”替换第N个匹配项，并且只替换第N个匹配项，我该怎么做？例如，如果我想替换第三个匹配项“10”(匹配项[2])，我不能只说s[matches[2]]="X"因为它做了两次替换“1fooX0barXg

  5. ruby-on-rails - Rails - 从另一个模型中创建一个模型的实例 - 2

    我有一个正在构建的应用程序，我需要一个模型来创建另一个模型的实例。我希望每辆车都有4个轮胎。汽车模型classCar轮胎模型classTire但是，在make_tires内部有一个错误，如果我为Tire尝试它，则没有用于创建或新建的activerecord方法。当我检查轮胎时，它没有这些方法。我该如何补救？错误是这样的:未定义的方法'create'forActiveRecord::AttributeMethods::Serialization::Tire::Module我测试了两个环境:测试和开发，它们都因相同的错误而失败。 最佳答案

  6. ruby-on-rails - 浏览 Ruby 源代码 - 2

    我的主要目标是能够完全理解我正在使用的库/gem。我尝试在Github上从头到尾阅读源代码，但这真的很难。我认为更有趣、更温和的踏脚石就是在使用时阅读每个库/gem方法的源代码。例如，我想知道RubyonRails中的redirect_to方法是如何工作的:如何查找redirect_to方法的源代码？我知道在pry中我可以执行类似show-methodmethod的操作，但我如何才能对Rails框架中的方法执行此操作？您对我如何更好地理解Gem及其API有什么建议吗？仅仅阅读源代码似乎真的很难，尤其是对于框架。谢谢! 最佳答案 Ru

  7. ruby-on-rails - RSpec:避免使用允许接收的任何实例 - 2

    我正在处理旧代码的一部分。beforedoallow_any_instance_of(SportRateManager).toreceive(:create).and_return(true)endRubocop错误如下:Avoidstubbingusing'allow_any_instance_of'我读到了RuboCop::RSpec:AnyInstance我试着像下面那样改变它。由此beforedoallow_any_instance_of(SportRateManager).toreceive(:create).and_return(true)end对此:let(:sport_

  8. ruby - 模块嵌套代码风格偏好 - 2

    我的假设是moduleAmoduleBendend和moduleA::Bend是一样的。我能够从thisblog找到解决方案,thisSOthread和andthisSOthread.为什么以及什么时候应该更喜欢紧凑语法A::B而不是另一个，因为它显然有一个缺点？我有一种直觉，它可能与性能有关，因为在更多命名空间中查找常量需要更多计算。但是我无法通过对普通类进行基准测试来验证这一点。 最佳答案 这两种写作方法经常被混淆。首先要说的是，据我所知，没有可衡量的性能差异。(在下面的书面示例中不断查找)最明显的区别，可能也是最著名的，是你的

  9. ruby - 寻找通过阅读代码确定编程语言的ruby gem？ - 2

    几个月前，我读了一篇关于ruby​​gem的博客文章，它可以通过阅读代码本身来确定编程语言。对于我的生活，我不记得博客或gem的名称。谷歌搜索“ruby编程语言猜测”及其变体也无济于事。有人碰巧知道相关gem的名称吗？ 最佳答案 是这个吗:http://github.com/chrislo/sourceclassifier/tree/master 关于ruby-寻找通过阅读代码确定编程语言的rubygem？，我们在StackOverflow上找到一个类似的问题：

  10. ruby - Net::HTTP 获取源代码和状态 - 2

    我目前正在使用以下方法获取页面的源代码:Net::HTTP.get(URI.parse(page.url))我还想获取HTTP状态，而无需发出第二个请求。有没有办法用另一种方法做到这一点？我一直在查看文档，但似乎找不到我要找的东西。 最佳答案 在我看来，除非您需要一些真正的低级访问或控制，否则最好使用Ruby的内置Open::URI模块:require'open-uri'io=open('http://www.example.org/')#=>#body=io.read[0,50]#=>"["200","OK"]io.base_ur

随机推荐