Combine仅返回completion未输出output问题求助

核心原因 #

该异常是由于Combine操作符在全局并发队列上的调度竞态导致的：

• subscribe(on: DispatchQueue.global()) 将订阅建立流程（含Deferred闭包执行、Future实例化）放到全局并发队列，而全局队列的任务调度无严格顺序，易被系统其他任务抢占。
• delay(for:scheduler:) 复用全局并发队列，Future同步发送的output与completion事件的延迟任务会被加入队列。并发队列的无序调度可能打破output先于completion的传递顺序，甚至导致output事件被丢弃。
• 全局并发队列的无顺序特性，会让订阅建立与事件发送的任务交叉执行，干扰事件的正常传递链路。

修复方案 #

方案1：使用串行队列（最简有效） #

将upstreamQueue改为串行队列，从根源上保证事件流的顺序执行：

private func createPublisher() -> AnyPublisher<String, Error> {
    let upstreamQueue = DispatchQueue(label: "upstream-queue") // 替换为串行队列
    return Deferred {
        Future { promise in
            promise(.success("TestCompleted"))
        }
    }
    .subscribe(on: upstreamQueue)
    .delay(for: .milliseconds(100), tolerance: .milliseconds(10), scheduler: upstreamQueue)
    .receive(on: RunLoop.main)
    .eraseToAnyPublisher()
}

方案2：给delay单独使用串行队列 #

若需保留subscribe(on:)的并发队列，可让delay操作在串行队列上执行，确保延迟事件的顺序处理：

private func createPublisher() -> AnyPublisher<String, Error> {
    let upstreamQueue = DispatchQueue.global()
    let delaySerialQueue = DispatchQueue(label: "delay-serial-queue")
    return Deferred {
        Future { promise in
            promise(.success("TestCompleted"))
        }
    }
    .subscribe(on: upstreamQueue)
    .delay(for: .milliseconds(100), tolerance: .milliseconds(10), scheduler: delaySerialQueue)
    .receive(on: RunLoop.main)
    .eraseToAnyPublisher()
}

方案3：异步发送Future事件 #

将promise调用改为异步，避免订阅建立与事件发送在同一同步任务中执行：

private func createPublisher() -> AnyPublisher<String, Error> {
    let upstreamQueue = DispatchQueue.global()
    return Deferred {
        Future { promise in
            upstreamQueue.async { // 在订阅队列异步发送事件
                promise(.success("TestCompleted"))
            }
        }
    }
    .subscribe(on: upstreamQueue)
    .delay(for: .milliseconds(100), tolerance: .milliseconds(10), scheduler: upstreamQueue)
    .receive(on: RunLoop.main)
    .eraseToAnyPublisher()
}

验证逻辑 #

Combine的事件流依赖严格的顺序契约（先output后completion），全局并发队列的无序调度会破坏该契约。上述方案通过保证事件处理的顺序性，确保output与completion按预期传递到下游订阅者。

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