CurrentValueSubject是否线程安全？严格并发下解决方案咨询

现象是否正常？ #

这是正常现象。Combine的CurrentValueSubject（以及其他Subject类型）本身并非线程安全组件，它要求所有发送（send(_:)）操作必须在同一个串行执行上下文（比如同一个DispatchQueue或Actor）中完成。

在Swift Strict Concurrency（严格并发）模式下，编译器会更严格地校验代码的并发安全性，原本可能被隐藏的线程违规操作会触发内部的队列断言，也就是你遇到的dispatch_assert_queue_fail崩溃。而禁用Strict Concurrency时，编译器放松了这类检查，崩溃暂时消失，但这只是掩盖了线程不安全的问题，高并发场景下仍可能出现数据竞争或其他未知异常。

严格并发模式下的线程安全发布方案 #

方案1：用Actor封装Subject #

利用Swift Actor的串行执行特性，把Subject的发送操作限制在Actor上下文内，从根本上保证线程安全：

@preconcurrency import Combine
import XCTest

final class ThreadSafeValueActor: Actor {
    private let subject = CurrentValueSubject<Int, Never>(0)
    // 对外暴露只读的Publisher
    var publisher: AnyPublisher<Int, Never> {
        subject.eraseToAnyPublisher()
    }

    // 仅通过Actor方法发送值
    func send(_ value: Int) {
        subject.send(value)
    }
}

final class CombineTests: XCTestCase {
    func testExample() throws {
        let exp = expectation(description: "received 1")
        let safeActor = ThreadSafeValueActor()
        
        let cancellable = safeActor.publisher.sink { value in
            if value == 1 {
                exp.fulfill()
            }
        }

        // 在任意上下文通过await调用Actor方法发送值
        Task.detached {
            await safeActor.send(1)
        }

        wait(for: [exp], timeout: 1)
        _ = cancellable
    }
}

方案2：用专用串行队列统一发送操作 #

创建一个专用的串行队列，所有send操作都通过该队列执行，确保Subject的所有修改都在同一线程上下文：

@preconcurrency import Combine
import XCTest

final class CombineTests: XCTestCase {
    func testExample() throws {
        let exp = expectation(description: "received 1")
        let currentValue = CurrentValueSubject<Int, Never>(0)
        // 创建专用串行队列用于发送操作
        let sendQueue = DispatchQueue(label: "com.subject.send.queue")
        
        let cancellable = currentValue.eraseToAnyPublisher().sink { value in
            if value == 1 {
                exp.fulfill()
            }
        }

        DispatchQueue.global(qos: .background).async {
            // 所有send操作都通过专用队列执行
            sendQueue.async {
                currentValue.send(1)
            }
        }

        wait(for: [exp], timeout: 1)
        _ = cancellable
    }
}

补充说明 #

当前Combine框架尚未完全适配Swift的并发标注，所以需要添加@preconcurrency import Combine来避免编译器警告，后续苹果官方更新Combine的并发支持后可移除该标注。

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