Swift 5.5 并发：如何序列化异步任务以用 maxConcurrentOperationCount = 1 替换 OperationQueue？

Question

我目前正在迁移我的应用程序以使用 Swift 中的并发模型。我想序列化任务以确保它们一个接一个地执行（没有并行性）。在我的用例中，我想收听 NotificationCenter 发布的通知，并在每次发布新通知时执行一个任务。但我想确保没有以前的任务正在运行。这相当于使用 maxConcurrentOperationCount = 1 的 OperationQueue。

例如，我在我的应用程序中使用 CloudKit 和 Core Data，并使用持久历史跟踪来确定商店中发生了哪些变化。 在这个Synchronizing a Local Store to the Cloud 示例代码中，Apple 使用操作队列来处理历史处理任务（在 CoreDataStack 中）。此 OperationQueue 的最大操作数设置为 1。

private lazy var historyQueue: OperationQueue = {
    let queue = OperationQueue()
    queue.maxConcurrentOperationCount = 1
    return queue
}()

当收到一个 Core Data 通知时，一个新任务被添加到这个串行操作队列中。因此，如果收到很多通知，它们都会以串行方式一个接一个地执行。

@objc
func storeRemoteChange(_ notification: Notification) {
    // Process persistent history to merge changes from other coordinators.
    historyQueue.addOperation {
        self.processPersistentHistory()
    }
}

在这个Loading and Displaying a Large Data Feed 示例代码中，Apple 使用任务来处理历史更改（在 QuakesProvider 中）。

// Observe Core Data remote change notifications on the queue where the changes were made.
notificationToken = NotificationCenter.default.addObserver(forName: .NSPersistentStoreRemoteChange, object: nil, queue: nil) { note in
    Task {
        await self.fetchPersistentHistory()
    }
}

我觉得第二个项目有问题，因为任务可以按任何顺序发生，不一定按顺序发生（与第一个项目中 OperationQueue 作为 maxConcurrentOperationCount = 1 的情况相反）。

我们是否应该在某处使用演员来确保方法被串行调用？

我想过这样的实现，但我还不太适应：

actor PersistenceStoreListener {
    let historyTokenManager: PersistenceHistoryTokenManager = .init()
    private let persistentContainer: NSPersistentContainer

    init(persistentContainer: NSPersistentContainer) {
        self.persistentContainer = persistentContainer
    }

    func processRemoteStoreChange() async {
        print("\(#function) called on \(Date.now.formatted(date: .abbreviated, time: .standard)).")
    }
}

收到新通知时将调用 processRemoteStoreChange 方法的位置（AsyncSequence）：

notificationListenerTask = Task {
   let notifications = NotificationCenter.default.notifications(named: .NSPersistentStoreRemoteChange, object: container.persistentStoreCoordinator)
   
   for await _ in notifications {
        print("notificationListenerTask called on \(Date.now.formatted(date: .abbreviated, time: .standard)).")
        await self.storeListener?.processRemoteStoreChange()
    }
}

Answer 1

使用actor，它的同步方法已经为您协调好了，不需要其他任何东西。例如。在下面的 Instruments 屏幕快照中，前四个 ⓢ 路标是我等待该 actor 的同步方法的位置，绿色间隔表示该方法实际运行的时间。

现在，如果您正在调用 Actor 的异步方法，那么 Actor 重入会干扰您想要实现的“一次一个”行为。 （有关重入的讨论，请参阅 SE-0306。）请参阅后四个路标，我将其中的 actor 异步方法称为异步方法，结果间隔以橙色显示。

简而言之，如果您的 actor 仅执行同步任务（这是您的情况），那么 actor 会自动产生 maxConcurrentOperationCount = 1 类型的行为。

---

FWIW，这不是很相关，但这是我的 minalist actor 用于上述内容：

actor ActorExample {
    func synchronousTask() {
        Thread.sleep(forTimeInterval: 2)
    }

    func asynchronousTask() async throws {
        try await Task.sleep(nanoseconds: NSEC_PER_SEC * 2)
    }
}

并分别调用它们，如下所示：

let example = ActorExample()

func startSynchronous() {
    Task(priority: .background) {
        os_signpost(.event, log: pointsOfInterest, name: #function)

        await example.synchronousTask()
    }
}

func startAsynchronous() {
    Task(priority: .background) {
        os_signpost(.event, log: pointsOfInterest, name: #function)

        try await example.asynchronousTask()
    }
}

---

你问：

如果您想了解有关processPersistentHistory 方法的更多详细信息，可以在本文中找到或多或少的内容：https://avanderlee.com/swift/persistent-history-tracking-core-data。您会看到它创建了一个新的核心数据backgroundContext 并执行一些操作，但以同步方式 (.performAndWait)。使用新的 Swift 并发模型和核心数据，这个 performAndWait 现在是 await context.perform { ... }。但是如果我们使用Task { await context.perform { ... sync work ... } }，我仍然不清楚如何将它们序列化，因为Task 可以在任意线程上运行。

上面我重点介绍了您问题标题中提出的问题，如何序列化Tasks。但如果问题真的是如何序列化perform(_:)，看来您不必这样做。 documentation 表示它已经在队列上运行了已执行的请求：

在协调器的队列上异步执行闭包。

如果确实如此，那么您根本不必担心序列化它，因为 perform(_:) 会为您做到这一点。不需要额外的参与者或其他序列化机制。尽情享受 Swift 并发吧。