Swift Combine：如何让Publisher在指定队列启动执行？

我知道可以用.subscribe(on: ...)指定订阅操作的队列，但如果要让Publisher本身在指定队列工作该怎么处理？

示例代码如下：

class ViewModel {
    
    @Published var count: Int = 0

    func doSomething2() { // <-- 可能在任意队列执行
        
        guard Thread.isMainThread else { // <-- 切换到主队列，能否优化？
            DispatchQueue.main.async {
                self.doSomething2()
            }
            return
        }
        
        Just(count) // <-- 必须在主队列访问`count`
            .map { value in
                count = max(0, value - 1)
                return count
            }
            .first { value in
                value > 0
            }
            .subscribe(on: DispatchQueue.global())
            .sink { _ in
                print("Running a long operation 2")
                self.doSomething2()
            }
    }
}

// 以下视图代码仅为示例可执行性，与问题无关
struct MyView: View {
    
    var viewModel = ViewModel()
    
    var body: some View {
        
        VStack {
            Button("Try me") {
                viewModel.count = 3
                viewModel.doSomething2()
            }
        }
    }
}

核心问题：

• 在该示例中，能否去掉guard Thread.isMainThread...的代码，转而通过某种方式指定Publisher在主队列工作？
• 更通用地，是否可以像为订阅者使用.subscribe(on: ...)那样，以声明式方式让任意Publisher在指定队列工作？还是说每次都需要自定义Publisher来实现？

问题解答 #

直接优化示例代码 #

完全可以去掉那段线程判断逻辑，利用Combine的原生操作符就能实现需求。优化后的代码如下：

class ViewModel {
    
    @Published var count: Int = 0
    private var cancellables = Set<AnyCancellable>() // 存储订阅，防止被释放

    func doSomething2() { // 可在任意队列调用
        
        Just(count)
            .receive(on: DispatchQueue.main) // 确保map操作在主队列执行，安全访问/修改count
            .map { [weak self] value in
                guard let self = self else { return 0 }
                self.count = max(0, value - 1)
                return self.count
            }
            .first { $0 > 0 }
            .subscribe(on: DispatchQueue.global()) // 把耗时的sink操作放到全局队列
            .sink(receiveCompletion: { _ in }, receiveValue: { [weak self] _ in
                print("Running a long operation 2")
                self?.doSomething2()
            })
            .store(in: &cancellables)
    }
}

关键操作符解析 #

要搞清楚两个核心操作符的区别，就能灵活控制队列：

• .subscribe(on:)：控制订阅启动过程的执行队列，比如Publisher的初始化、上游订阅的触发逻辑会在这个队列执行，适合把订阅的启动移到后台，避免阻塞当前线程。
• .receive(on:)：控制**下游所有操作（map、filter、sink等）**的执行队列，用来切换后续闭包的运行线程，比如把需要访问UI或主线程变量的操作切回主队列。

通用队列控制方案 #

如果要让任意Publisher的全部逻辑都在指定队列执行，只需组合这两个操作符即可，不需要自定义Publisher：

// 让整个Publisher流程在targetQueue执行
somePublisher
    .subscribe(on: targetQueue)
    .receive(on: targetQueue)
    .sink { /* 处理逻辑 */ }
    .store(in: &cancellables)

额外注意事项 #

• 务必用[weak self]捕获避免循环引用，尤其是ViewModel中的Combine订阅。
• 所有订阅都需要存储到Set<AnyCancellable>中，否则订阅会被立即释放，业务逻辑无法执行。
• @Published属性本身是线程安全的（iOS 13+），但如果需要在非主线程修改后立即更新UI，还是建议通过.receive(on: DispatchQueue.main)确保UI更新在主线程。

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