如何从NotificationCenter Publisher调用GlobalActor函数？求最优方案

问题背景 #

我在应用中使用了一个带有若干方法的GlobalActor，希望收到NotificationCenter的通知时调用该Actor的方法，但sink的闭包是同步的，无法直接使用await。当前实现是在闭包内创建Task调用Actor方法，但觉得这种方式不妥，还可能存在竞态风险，想知道是否有更优的实现方式，也曾尝试在Actor内部接收通知，但情况并未改善。

当前实现代码：

class MyClass {
    private var cancellables: [AnyCancellable] = []

    init() {
        NotificationCenter.default.publisher(for: NotificationName)
            .receive(on: DispatchQueue.global(qos: .utility))
            .compactMap { $0 as? SomeType }
            .sink { [weak self] val in
                Task { [weak self] in
                    await self?.someCallToActor(val)
                }
            }.store(in: &cancellables)
    }
    
    @SomeGlobalActor
    func someCallToActor(_ val: String) async {
        await SomeGlobalActor.shared.actorMethod(val)
    }
}

@globalActor
actor SomeGlobalActor {
    static var shared = SomeGlobalActor()

    func actorMethod(_ val: String) async {
        // 业务逻辑
    }
}

优化方案 #

1. 直接用Task.detached绑定全局Actor #

不需要额外封装someCallToActor方法，直接在sink里创建绑定到SomeGlobalActor的分离任务，任务会直接在Actor的执行上下文运行，避免多余的上下文切换：

.sink { val in
    Task.detached(priority: .utility) { @SomeGlobalActor [weak self] in
        guard let self = self else { return }
        await SomeGlobalActor.shared.actorMethod(val)
    }
}

这里指定priority: .utility和你之前receive(on:)的队列QoS对应，保持执行优先级一致。

2. 将订阅逻辑移入GlobalActor内部 #

如果之前在Actor内订阅没改善，大概率是没处理好上下文绑定。可以在Actor里直接订阅，并用receive(on:)指定Actor的串行执行上下文：

@globalActor
actor SomeGlobalActor {
    static var shared = SomeGlobalActor()
    private var cancellables: [AnyCancellable] = []

    private init() {
        NotificationCenter.default.publisher(for: NotificationName)
            .compactMap { $0 as? SomeType }
            // 强制事件在当前Actor的串行上下文处理
            .receive(on: self)
            .sink { [weak self] val in
                // 已处于Actor上下文，可直接调用异步方法
                Task {
                    await self?.actorMethod(val)
                }
            }.store(in: &cancellables)
    }

    func actorMethod(_ val: String) async {
        // 业务逻辑
    }
}

这种方式下所有通知事件都会被Actor串行处理，从根源上规避竞态风险——因为Actor本身的执行机制就是串行的，不会出现多线程同时访问内部状态的情况。

3. 自定义Combine Operator封装异步调用 #

如果需要在多个地方复用“通知触发Actor方法”的逻辑，可以封装一个Combine Operator：

extension Publisher where Output == String {
    func sinkToActor<T: GlobalActor>(_ actor: T.Type, action: @escaping (Output) async -> Void) -> AnyCancellable {
        sink { val in
            Task.detached(priority: .utility) { @T in
                await action(val)
            }
        }
    }
}

使用时可以大幅简化代码：

NotificationCenter.default.publisher(for: NotificationName)
    .compactMap { $0 as? SomeType }
    .sinkToActor(SomeGlobalActor.self) { val in
        await SomeGlobalActor.shared.actorMethod(val)
    }
    .store(in: &cancellables)

关于竞态风险的说明 #

你担心的竞态风险，本质是多个Task同时访问共享状态导致的问题。但GlobalActor本身是串行执行的，只要确保所有对共享状态的操作都在Actor上下文内进行，就不会出现竞态。之前的实现中，只要actorMethod内部的逻辑是在Actor里执行的，即使创建多个Task，也会被Actor串行调度处理，不会有竞态问题。

内容的提问来源于stack exchange，提问作者Plays2