认识 AsyncSequence

随时间迭代值序列现在和编写“for”循环一样简单。了解新的 AsyncSequence 协议如何实现自然、简单的语法，将来自通知的任何内容迭代成服务器流式传输的字节。我们还将向您展示如何修改现有代码，以提供自己的异步序列。为了充分了解本节内容，我们建议先观看“认识 Swift 中的 async/await”。

资源

相关视频

WWDC22

WWDC21

下载

♪低音音乐播放♪ ♪ 菲利普豪斯勒：嗨我叫菲利普很高兴向你介绍 Swift中的一个很酷的新功能 AsyncSequence 今天我们将讨论什么是异步序列以及它们背后的基本原理然后我们将讨论如何在代码中使用它们并介绍一些新的AsyncSequence API 在最后我们将探讨如何构建自己的异步序列所以让我们直接开始吧这是我编写的简单工具用来说明你能用AsyncSequence 做的一些很酷的新东西在这个工具中我们从一个指向端点的URL开始它列出了最近的地震现在通常下载东西实际上是一项可能需要一些时间的异步任务但是在这种情况下我们不想等待所有的下载相反地我们想要展示收到的东西所以我决定稍微改变一下并使用新的async/await功能来让代码行从端点获得响应我们正在获取的数据被格式化为逗号分隔的文本所以每一行都是一个完整的数据行由于代码行的AsyncSequence 在被收到时就会把每一行都发出去这意味着我们可能有一个非常大的下载量但是通过在我们得到它们时发出它们这些代码片段感觉非常合理最棒的是你可以在这个新的异步上下文常规序列中使用你熟悉的相同东西这意味着你可以使用新的 for-await-in语法来迭代以及map、filter 和reduce等功能或者…就像在这个范例中一样… dropFirst函数来操纵这些值那么这是如何运作的呢？好吧我今天要讨论的很多事情都是基于async/await座谈的但让我们回顾一下几个关键点通过使用await关键词异步函数使你无需回调即可编写并行代码调用异步函数将挂起然后在产生值或错误时恢复另一方面 AsyncSequence 当底层迭代器产生一个值或抛出时将在每个元素上挂起并恢复基本上顾名思义它们就像常规的序列但有一些关键的区别也就是说每个元素都是异步传递的但是因为它们是异步交付这意味着失败是绝对有可能的一些异步序列抛出但如果不允许失败其他的就不会这么做就像抛出的函数一样在迭代或组合时编译器将帮助确保错误得到处理一般来说异步序列是一种描述如何随着时间的推移生成值所以AsyncSequence 可能是零个或多个值然后通过从它的迭代器返回一个nil来表示完成就像序列一样当发生错误时这也是 AsyncSequence处于终止状态的点并且在错误发生后它们将在迭代器上对next的任何后续调用返回nil 因此让我们首先从常规迭代开始深入了解该定义的工作原理在这里我们有一个非常熟悉的模式这是一个for-in循环在这种情况下它会从一个序列中迭代地震然后调用一个函数当幅度超过某个值时编译器知道这个迭代应该如何工作但它的作用并不稀奇编译步骤实际上只是做了一些简单的转换让我们检查这些转换以便你了解它的异步形式是什么这大致上是编译器在构建前面的代码时所做的它首先通过创建一个迭代器变量开始然后使用一个while循环来获取每次调用next时由迭代器生成的地震要使用新的async/await功能这里可以做一个小改动就像将下一个函数更改为异步函数一样简单我们现在可以经由等待下一次地震来让迭代参与Swift的并行性让我们回过头来看看如果循环在AsyncSequence上会是什么样子如前所述我们需要等待 AsyncSequence中的每一项这反映在新的for-await-in语法中这一切都意味着如果你知道如何使用Sequence 那你就对如何使用AsyncSequence 有了很好的了解有几种方法可以利用异步序列正如我们刚刚介绍的那样你可以使用新的for-await-in语法或者如果AsyncSequence抛出你可以使用新的 for-try-await-in语法这使你可以轻松迭代异步产生的值无需处理闭包并使用你已经熟悉的语法来迭代它们甚至像break和continue 也能正常运作现在你已经听完了异步序列理论的介绍让我们进一步探索该迭代给定一个异步序列的源你可以通过使用 for-await-in语法来等待每个值这意味着它将等待迭代器生成的每个项目当它到达终端时它完成了循环在迭代异步序列时 break显然是从循环内部提前终止迭代的好方法这就像常规序列一样工作当地震的位置数据不存在时我们就中断了或者如果我们有一些想要跳过的值我们可以使用continue 在这种情况下如果深度大于某个值我们将跳过那些并继续等待下一次地震下载的下一次迭代会和以前一样运作但在这种情况下我们拥有的来源可能会引发错误就像抛出函数一样在被迭代的AsyncSequence 可以抛出时每个元素都需要try来处理并且就像抛出函数一样编译器会检测你何时错过了try 并为你提供Fix-it来纠正错误这意味着在AsyncSequence可能产生错误时你都会很安全因为语言迫使你要嘛抛出那个错误或抓住它第二次迭代会在第一个循环的迭代之后按顺序运行顺序运行代码并不总是需要的如果与其他正在发生的事情并行运行迭代很有用你可以创建一个新的异步任务来封装迭代当你知道正在使用的异步序列可能无限期运行时这会很有用现在即使该序列可能是不确定的但发生的概率要低得多但在异步行为的世界里这是更常见的事情以及使用它们时需要考虑的事项值得庆幸的是Swift中的并行功能让这件事变得非常简单和安全这也很有帮助当你想潜在从外部取消迭代时这里我们可以同时运行两次迭代并在稍后终止迭代可以轻易地让任务确定和某个容器的生命周期一样无限期的工作范围接下来让我们来看看一些AsyncSequence API 可用于macOS Monterey iOS 15、tvOS 15和watchOS 8上其中有很多但我将向你展示一些亮点从文件中读取通常是异步行为的主要用例 FileHandle现在有一个新的字节属性可以访问来自该FileHandle的异步字节序列这可以与AsyncSequence上的将异步字节序列转换为代码行的新扩展结合使用但是处理文件是如此普遍以至于我们决定URL 对于字节和代码行应该要有访问器这与我在初始范例中所使用的API相同它是URL上的一个便利属性将代码的AsyncSequence 从内容中返回无论是来自文件还是网络我确信这将使许多以前非常复杂的任务变得容易和安全有时从网络获取信息需要对响应进行更多控制和认证所以URLSession现在有一个字节函数获取给定URL或URLRequest的 AsyncSequence字节如果你想了解更多你一定要看看《通过URLSession 使用async/await》座谈来获取这方面的更多详细信息以及URLSession的更多新异步功能但是文件和网络并不是唯一对AsyncSequence有意义的东西现在可以使用新的通知API等待通知迭代并不是与AsyncSequence 交互的唯一方式在此范例中我们正在等待具有匹配存储UUID的远程更改的第一个通知使用firstWhere之类的方法以及AsyncSequence通知能允许一些简洁的新设计模式做出的代码可以将先前的复杂逻辑转为紧凑和易读如果所有这些还不够酷还有很多新的API 来用于异步操作来自异步序列的值这些应该很熟悉因为它们是一些在Sequence上可用的相同的功能到目前为止我们已经介绍了一些例如dropFirst和firstWhere 但还有很多不仅是这些几乎所有你能想到的在Sequence上使用的东西现在都有一个异步对应物能与AsyncSequence配合使用现在有这么多东西得搞懂你可能会想 “嘿那些新API真的很酷而且语法非常简洁但是我怎样才能制作自己的异步序列呢？” 好吧那就这么做吧！实现AsyncSequence的方法有好几种但我将专注于如何调整现有代码特别是有一些设计模式非常适合AsyncSequence 我们有一些很棒的功能可以让你已经拥有的东西与这个新概念进行交互其中一些设计模式就像多次调用的闭包但也有一些代理也可以很好地工作几乎任何不需要响应而且只需要通知发生新值的东西可以是制作AsyncSequence的主要候选者这些设计模式真的很常见你现在的应用程序中可能已经有了一些这是常见处理程序模式的范例它是一个具有处理程序属性和启动以及停止方法的类它似乎是AsyncSequence的完美候选者现有的用法可能是这样的其中创建了一个监视器并分配一个获取值的处理程序然后启动监视器以便可以将地震发送给处理程序稍后监视器可能会停止以取消正在生成的事件我们可以使用相同的接口来适应新的 AsyncStream类型的用法只需少量代码即可使用它并允许你构建AsyncSequence 在构建AsyncStream时元素类型和构造闭包是有明确指定的闭包有一个延续可以多次产生值、完成或处理终止所以这意味着在这种情况下可以在构造闭包内创建监视器然后可以分配处理程序来延续地震的产生然后onTermination可以处理取消和清理然后我们就可以开始监控了我们之前拥有的这个相同的监视器代码可以在AsyncStream的构造中轻松封装这减少了在每个使用站点中复制相同逻辑的需要这就是AsyncStream的使用方式你可以使用强大的转换功能例如说过滤器以及新的for-await-in语法这让你可以专注于代码的意图而无需担心复制簿记因为一切都集中在一个地方 AsyncStream还有很大的灵活性来创建你自己的异步序列这实际上只是一个例子还有更多其他例子可以让你在你的代码中采用 AsyncStream是将现有代码调整为 AsyncSequence的好方法它可以处理所有你期望 AsyncSequence的事情例如安全、迭代和取消但他们也处理缓冲 AsyncStream是一种可靠的方式来让你构建自己的异步序列以及来自你自己的API的合适的返回类型因为唯一元素的生产来源就是来自建设如果你需要表示抛出的错误呢？这个嘛我们给它准备了一个类型！ AsyncThrowingStream 就像AsyncStream 但可以处理错误它提供与AsyncStream 相同的灵活性和安全性但是可以通过从迭代中抛出处理失败 AsyncSequence 是一个非常强大的工具既安全又熟悉用于处理多个异步值如果你知道如何使用Sequence 你已经知道如何使用AsyncSequence了我们已经讨论了什么是异步序列以及它们是如何使用的并向你介绍了AsyncStream 我们已经深入研究了理论和它们的定义以及一些新引入的异步序列以及最后如何自己构建它们我热切地期待你们接下来能用它做出什么来 ♪

0:37 - QuakesTool

@main
struct QuakesTool {
    static func main() async throws {
        let endpointURL = URL(string: "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_month.csv")!

        // skip the header line and iterate each one 
        // to extract the magnitude, time, latitude and longitude
        for try await event in endpointURL.lines.dropFirst() {
            let values = event.split(separator: ",")
            let time = values[0]
            let latitude = values[1]
            let longitude = values[2]
            let magnitude = values[4]
            print("Magnitude \(magnitude) on \(time) at \(latitude) \(longitude)")
        }
    }
}

3:24 - Iterating a Sequence

for quake in quakes {
    if quake.magnitude > 3 {
        displaySignificantEarthquake(quake)
    }
}

3:52 - How the compiler handles iteration

var iterator = quakes.makeIterator()
while let quake = iterator.next() {
    if quake.magnitude > 3 {
        displaySignificantEarthquake(quake)
    }
}

4:11 - How the compiler handles asynchronous iteration

var iterator = quakes.makeAsyncIterator()
while let quake = await iterator.next() {
    if quake.magnitude > 3 {
        displaySignificantEarthquake(quake)
    }
}

4:28 - Iterating an AsyncSequence

for await quake in quakes {
    if quake.magnitude > 3 {
        displaySignificantEarthquake(quake)
    }
}

5:36 - Terminating iteration early by breaking

for await quake in quakes {
    if quake.location == nil {
        break
    }
    if quake.magnitude > 3 {
        displaySignificantEarthquake(quake)
    }
}

5:51 - Skipping values by continuing

for await quake in quakes {
    if quake.depth > 5 {
        continue
    }
    if quake.magnitude > 3 {
        displaySignificantEarthquake(quake)
    }
}

6:05 - AsyncSequence might throw

do {
    for try await quake in quakeDownload {
        ...
    }
} catch {
    ...
}

7:15 - Concurrently iterating inside an async task

let iteration1 = Task {
    for await quake in quakes {
        ...
    }
}

let iteration2 = Task {
    do {
        for try await quake in quakeDownload {
            ...
        }
    } catch {
        ...
    }
}

//... later on  
iteration1.cancel()
iteration2.cancel()

7:56 - Reading bytes from a FileHandle

for try await line in FileHandle.standardInput.bytes.lines {
    ...
}

8:16 - Reading lines from a URL

let url = URL(fileURLWithPath: "/tmp/somefile.txt")
for try await line in url.lines {
    ...
}

8:49 - Reading bytes from a URLSession

let (bytes, response) = try await URLSession.shared.bytes(from: url)

guard let httpResponse = response as? HTTPURLResponse,
      httpResponse.statusCode == 200 /* OK */
else {
    throw MyNetworkingError.invalidServerResponse
}

for try await byte in bytes {
    ...
}

9:12 - Notifications

let center = NotificationCenter.default
let notification = await center.notifications(named: .NSPersistentStoreRemoteChange).first {
    $0.userInfo[NSStoreUUIDKey] == storeUUID
}

11:10 - Using an AsyncStream

class QuakeMonitor {
    var quakeHandler: (Quake) -> Void
    func startMonitoring()
    func stopMonitoring()
}

let quakes = AsyncStream(Quake.self) { continuation in
    let monitor = QuakeMonitor()
    monitor.quakeHandler = { quake in
        continuation.yield(quake)
    }
    continuation.onTermination = { @Sendable _ in
        monitor.stopMonitoring()
    }
    monitor.startMonitoring()
}

let significantQuakes = quakes.filter { quake in
    quake.magnitude > 3
}

for await quake in significantQuakes {
    ...
}

正在查找特定内容？在上方输入一个主题，就能直接跳转到相应的精彩内容。