如何制作异步 Swift 函数“@synchronized”？

Question

我想创建一个异步函数，它本身使用异步调用。我还想确保在任何时刻只有一个呼叫被主动处理。所以我想要一个async @synchronized函数。

怎么做？将函数体包装在 内并dispatchQueue.sync {}不能像它期望的同步代码那样工作。另外，似乎DispatchQueue一般不设计有要执行的异步代码块/任务。

该代码与硬件通信，因此本质上是异步的，这就是为什么我想要为我的库提供异步接口。（我不想在通信阶段发生时阻止应用程序。）但是某些操作无法在硬件上并行执行，因此我必须进行同步，以便某些操作不会同时发生。

Answer 1

我原来的答案（我建议等待之前的任务）如下。这是一种运行良好的简单模式，但属于非结构化并发（使取消工作流程复杂化）。

\n

如今，我会使用AsyncSequence，例如AsyncStream。请参阅 WWDC 2021 视频认识 AsyncSequence。或者，对于类似队列的行为（我们最初设置队列，然后向其中添加项目），我通常会使用AsyncChannelSwift异步算法包。请参阅 WWDC 2022 视频认识 Swift 异步算法。

\n

例如，我可以AsyncChannel为我想要下载的 URL 创建一个：

\n

let urls = AsyncChannel<URL>()\n

\n

现在我有了一个通道，我可以设置一个任务来使用for- await-in循环串行处理它们：

\n

func processUrls() async {\n    for await url in urls {\n        await download(url)\n    }\n}\n

\n

而且，当我稍后想要向该通道添加要处理的内容时，我可以send向该通道添加：

\n

func append(_ url: URL) async {\n    await urls.send(url)\n}\n

\n\n

您可以让每一项Task等待前一项。您可以使用 actor 确保一次只运行一个。诀窍是，由于参与者可重入，您必须将“等待先前Task”逻辑放入同步方法中。

\n

例如，你可以这样做：

\n

actor Experiment {\n    private var previousTask: Task<Void, Error>?\n\n    func startSomethingAsynchronous() {\n        previousTask = Task { [previousTask] in\n            let _ = await previousTask?.result\n            try await self.doSomethingAsynchronous()\n        }\n    }\n\n    private func doSomethingAsynchronous() async throws {\n        let id = OSSignpostID(log: log)\n        os_signpost(.begin, log: log, name: "Task", signpostID: id, "Start")\n        try await Task.sleep(nanoseconds: 2 * NSEC_PER_SEC)\n        os_signpost(.end, log: log, name: "Task", signpostID: id, "End")\n    }\n}\n

\n

现在我正在使用，os_signpost这样我就可以从 Xcode Instruments 观察这个串行行为。无论如何，您可以像这样开始三个任务：

\n

import os.log\n\nprivate let log = OSLog(subsystem: "Experiment", category: .pointsOfInterest)\n\nclass ViewController: NSViewController {\n\n    let experiment = Experiment()\n\n    func startExperiment() {\n        for _ in 0 ..< 3 {\n            Task { await experiment.startSomethingAsynchronous() }\n        }\n        os_signpost(.event, log: log, name: "Done starting tasks")\n    }\n\n    ... \n}\n

\n

Instruments 可以直观地演示顺序行为（其中向\xe2\x93\xa2我们显示所有任务的提交完成的位置），但您可以在时间轴上看到任务的顺序执行：

\n

\n\n

我实际上喜欢将这种串行行为抽象为它自己的类型：

\n

actor SerialTasks<Success> {\n    private var previousTask: Task<Success, Error>?\n\n    func add(block: @Sendable @escaping () async throws -> Success) {\n        previousTask = Task { [previousTask] in\n            let _ = await previousTask?.result\n            return try await block()\n        }\n    }\n}\n

\n

然后，您需要此串行行为的异步函数将使用上述内容，例如：

\n

class Experiment {\n    let serialTasks = SerialTasks<Void>()\n\n    func startSomethingAsynchronous() async {\n        await serialTasks.add {\n            try await self.doSomethingAsynchronous()\n        }\n    }\n\n    private func doSomethingAsynchronous() async throws {\n        let id = OSSignpostID(log: log)\n        os_signpost(.begin, log: log, name: "Task", signpostID: id, "Start")\n        try await Task.sleep(nanoseconds: 2 * NSEC_PER_SEC)\n        os_signpost(.end, log: log, name: "Task", signpostID: id, "End")\n    }\n}\n

\n