Combine Subjects（CurrentValueSubject等）丢值问题及一致性实现咨询

问题分析 #

你遇到的问题本质是Combine 的 Subject（包括 CurrentValueSubject、PassThroughSubject 等）不具备线程安全性：当在异步上下文（比如 Task）中快速连续发送值时，Subject 内部状态更新会出现竞态条件，导致部分值被覆盖或跳过，最终订阅端无法收到完整的数据流 [0, 1, 2, 3, 4]。

核心原因 #

Combine 官方未给 Subject 提供线程安全保障，所有 send(_:)、value 赋值等操作都是非原子的。当多个线程（或同一线程内的快速异步操作）同时修改 Subject 状态时，会破坏其内部的事件队列逻辑，引发数据丢失或顺序错乱。

通用解决方案（无需逐个用例定制测试） #

要让 Subject 行为一致，核心是强制所有发送操作串行化执行，从根源上避免竞态条件。以下是两种可复用的实现方式：

1. 用 Actor 封装线程安全的 Subject #

利用 Swift Actor 的串行执行特性，将原生 Subject 包裹在 Actor 内部，确保所有发送操作按顺序执行：

import Combine

actor ThreadSafeSubject<Output, Failure: Error> {
    private let subject: CurrentValueSubject<Output, Failure>
    
    init(initialValue: Output) {
        self.subject = CurrentValueSubject(initialValue)
    }
    
    func send(_ value: Output) {
        subject.send(value)
    }
    
    func send(completion: Subscribers.Completion<Failure>) {
        subject.send(completion: completion)
    }
    
    var value: Output {
        get { subject.value }
        set { subject.value = newValue }
    }
    
    func eraseToAnyPublisher() -> AnyPublisher<Output, Failure> {
        subject.eraseToAnyPublisher()
    }
}

使用示例：

var cancellables = Set<AnyCancellable>()

func threadSafeSubject() -> ThreadSafeSubject<Int, Never> {
    let subject = ThreadSafeSubject(initialValue: 0)
    Task {
        await subject.send(1)
        await subject.send(2)
        await subject.send(3)
        await subject.send(4)
        await subject.send(completion: .finished)
    }
    return subject
}

threadSafeSubject()
    .eraseToAnyPublisher()
    .receive(on: DispatchQueue.main)
    .sink(
        receiveCompletion: { _ in print("Subscription completed.") },
        receiveValue: { value in print("💚 Received value: \(value)") }
    ).store(in: &cancellables)

2. 用串行队列统一调度发送操作 #

如果不想使用 Actor，也可以自定义一个串行队列，强制所有 Subject 的发送操作通过该队列执行：

import Combine

class ThreadSafeCurrentValueSubject<Output, Failure: Error>: Publisher {
    typealias Output = Output
    typealias Failure = Failure
    
    private let subject: CurrentValueSubject<Output, Failure>
    private let queue = DispatchQueue(label: "com.example.thread-safe-subject-queue")
    
    init(initialValue: Output) {
        self.subject = CurrentValueSubject(initialValue)
    }
    
    func send(_ value: Output) {
        queue.sync { subject.send(value) }
    }
    
    func send(completion: Subscribers.Completion<Failure>) {
        queue.sync { subject.send(completion: completion) }
    }
    
    var value: Output {
        get { queue.sync { subject.value } }
        set { queue.sync { subject.value = newValue } }
    }
    
    func receive<S>(subscriber: S) where S : Subscriber, Failure == S.Failure, Output == S.Input {
        subject.receive(subscriber: subscriber)
    }
}

使用示例：

var cancellables = Set<AnyCancellable>()

func threadSafeSubject() -> ThreadSafeCurrentValueSubject<Int, Never> {
    let subject = ThreadSafeCurrentValueSubject(initialValue: 0)
    Task {
        subject.send(1)
        subject.send(2)
        subject.send(3)
        subject.send(4)
        subject.send(completion: .finished)
    }
    return subject
}

threadSafeSubject()
    .receive(on: DispatchQueue.main)
    .sink(
        receiveCompletion: { _ in print("Subscription completed.") },
        receiveValue: { value in print("💚 Received value: \(value)") }
    ).store(in: &cancellables)

总结 #

不管用 Actor 还是串行队列，核心逻辑都是确保所有发送 Subject 的操作在同一个串行上下文执行，从根源上消除竞态条件。这种方案无需逐个用例测试，只要统一使用封装后的线程安全 Subject，就能保证数据流的一致性。

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