我目前正在将我的应用程序迁移到使用Swift中的并发模型。我希望序列化任务以确保它们一个接一个地执行(没有并行)。在我的用例中,我想要监听由NotificationCenter发布的通知,并在发布新通知时执行任务。但我想确保没有先前的任务正在运行。这相当于使用maxConcurrentOperationCount = 1的OperationQueue。
例如,我在我的应用程序中使用CloudKit和Core Data,并使用持久历史跟踪来确定存储中发生了哪些更改。在这个同步本地存储到云端示例代码中,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()
}
}
在这个加载和显示大数据源的示例代码中,苹果公司使用Tasks来处理历史更改(在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来确保方法被串行调用吗?
我考虑过这样的实现,但我还不太熟悉:
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)).")
}
}
当收到新通知(AsyncSequence)时,processRemoteStoreChange方法将被调用:
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()
}
}