Combine与RxSwift中throttle算子行为差异原因咨询

使用Combine的throttle算子时，发现它无法对异步序列的前两个元素进行节流，对比RxSwift的同逻辑实现，输出结果存在明显差异。

测试代码实现 #

封装类代码 #

class MyAnyPublisher<Element> {
    init() {
        setupOperator()
    }
    func setupOperator() {}
    func send(_ E: Element) {}
}

class MyCombinePublisher<Element>: MyAnyPublisher<Element> {
    let publisher = PassthroughSubject<Element, Never>()
    private var cancellables = Set<AnyCancellable>()
    override func setupOperator() {
        publisher
            .throttle(for: .seconds(1), scheduler: DispatchQueue.main, latest: false)
            .sink { value in
                print("Received value: \(value)")
            }.store(in: &cancellables)
    }
    override func send(_ e: Element) {
        publisher.send(e)
    }
}

class MyRxPublisher<Element>: MyAnyPublisher<Element> {
    let subject = PublishSubject<Element>()
    private let disposeBag = DisposeBag()
    override func setupOperator() {
        subject
            .throttle(.milliseconds(1000), latest: false, scheduler: MainScheduler.instance)
            .subscribe(onNext: { s in
                print("Received value \(s)")
            })
            .disposed(by: disposeBag)
    }
    override func send(_ e: Element) {
        subject.onNext(e)
    }
}

测试逻辑代码 #

func testPublisher(_ publisher: MyAnyPublisher<String>) {
    DispatchQueue.main.asyncAfter(wallDeadline: .now() + 0.1) {
        publisher.send("A")
        publisher.send("B")
        publisher.send("C")
    }
    DispatchQueue.main.asyncAfter(deadline: .now() + 0.2) {
        publisher.send("D")
    }
    DispatchQueue.main.asyncAfter(deadline: .now() + 0.4) {
        publisher.send("E")
    }
    DispatchQueue.main.asyncAfter(deadline: .now() + 0.6) {
        publisher.send("F")
    }
    DispatchQueue.main.asyncAfter(deadline: .now() + 1.2) {
        publisher.send("G")
    }
    DispatchQueue.main.asyncAfter(deadline: .now() + 1.4) {
        publisher.send("H")
    }
    DispatchQueue.main.asyncAfter(deadline: .now() + 1.6) {
        publisher.send("I")
    }
}

运行输出对比 #

RxSwift输出 #

MyRxPublisher
Received value A
Received value G

Combine输出 #

MyCombinePublisher
Received value: A
Received value: D
Received value: G

原因分析 #

造成这种差异的核心原因是Combine与RxSwift的throttle算子内部状态更新机制不同：

• RxSwift的throttle在输出第一个元素后，会同步标记进入节流状态，后续元素只要在冷却时间内到达就会被直接忽略，无需依赖定时器调度。
• Combine的throttle(for:scheduler:latest:)在latest: false模式下，输出第一个元素后，需要依赖指定调度器（此处为主线程）的定时器任务来更新节流状态。而主线程是串行RunLoop，当第一个元素发送后，定时器任务会被排入RunLoop队列等待执行，若第二个元素（如示例中的D）在定时器任务执行前到达，Combine的throttle此时还未标记为节流状态，会将该元素误判为新的窗口首元素并输出。

这种现象在Xcode14.3、iOS16.5版本中存在，本质是Combine throttle算子在状态更新时机上的设计差异，而非bug，但会导致快速连续发送元素时出现不符合预期的输出。

解决建议 #

如果需要完全匹配RxSwift的throttle冷却模式行为，可以通过以下方式处理：

1. 自定义Combine操作符，实现同步状态更新的节流逻辑；
2. 在throttle前添加receive(on:)将元素接收调度到后台队列，再通过receive(on: DispatchQueue.main)将输出切回主线程，避免主线程RunLoop的时序冲突。

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