Combine异步收集Publisher值失败问题排查

问题场景 #

在为Combine代码编写单元测试时，使用async let结合PassthroughSubject的values.reduce方法收集值，断言失败，实际收到空数组，而非预期的[10,20]。

测试代码 #

func test_collectingPassthroughValues() async throws {
    // 实际测试中该Subject会注入到被测单元中
    let subject = PassthroughSubject<Int, Never>()

    // 预期Subject完成后reduce才会结束，用async let先启动收集任务，后续发送数据后再await结果
    // 实际测试中这是被测单元的属性，会对Subject执行各种操作符
    async let collectValues = await subject.values.reduce(into: []) { $0.append($1) }

    // 发送数据并结束Subject
    subject.send(10)
    subject.send(20)
    subject.send(completion: .finished)

    // 等待收集结果并断言
    let values = await collectValues
    XCTAssertEqual(values, [10, 20])
}

报错信息 #

test_collectingPassthroughValues(): XCTAssertTrue failed - Found difference for 
Different count:
 |  Received: (0) []
 |  Expected: (2) [10, 20]

问题原因 #

核心问题在于异步任务调度的时序：

• async let声明后，Swift会立即将右侧的subject.values.reduce异步任务调度到后台执行，但当前测试线程会同步继续执行后续的subject.send代码。
• PassthroughSubject是热发布者，不会缓存未订阅前发送的值。此时reduce任务的订阅操作尚未完成，后续发送的10、20直接被丢弃，最终收集到空数组。

解决方法 #

在发送数据前，让当前任务让出线程，确保subject.values的订阅操作完全完成。可以使用await Task.yield()实现：

func test_collectingPassthroughValues() async throws {
    let subject = PassthroughSubject<Int, Never>()

    async let collectValues = await subject.values.reduce(into: []) { $0.append($1) }

    // 让出线程，确保订阅建立完成
    await Task.yield()

    subject.send(10)
    subject.send(20)
    subject.send(completion: .finished)

    let values = await collectValues
    XCTAssertEqual(values, [10, 20])
}

Task.yield()会让当前任务暂停，给系统时间调度并完成reduce任务的订阅操作，后续发送的值就能被正常收集到。

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