如何基于阻塞回调API实现无需手动调度的Combine Publisher

核心思路是通过Combine原生的延迟执行与调度管理能力，把阻塞的collecting调用从Publisher创建阶段移到订阅阶段，由Combine自行管理执行上下文，无需手动调用GCD的异步方法。

实现代码 #

import Combine

func publisher<T>(_ type: T.Type) -> AnyPublisher<T, Error> {
    return Deferred {
        let subject = PassthroughSubject<T, Error>()
        collecting(
            newValue: { value in
                subject.send(value)
            },
            completion: { error in
                if let error = error {
                    subject.send(completion: .failure(error))
                } else {
                    subject.send(completion: .finished)
                }
            }
        )
        return subject
    }
    // 可根据实际需求替换为对应调度器，调度逻辑完全由Combine管理
    .subscribe(on: DispatchQueue.global())
    .eraseToAnyPublisher()
}

原理解释 #

• 原代码的问题是调用publisher方法时会立刻执行阻塞的collecting，导致return语句被卡住无法返回。Deferred容器会将内部Publisher的创建逻辑延迟到第一个订阅者发起订阅时才执行，调用publisher方法时会立刻返回实例，不会被阻塞。
• subscribe(on:)是Combine原生调度API，会将Deferred闭包内的所有逻辑（包括阻塞的collecting调用）放到指定调度器上执行，整个调度过程不需要手动写DispatchQueue.async，完全符合需求。
• 下游可以通过receive(on:)自由指定接收事件的线程，和上游调度逻辑完全隔离，不会产生冲突。

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