Swift中Actor实现Combine Publisher的并发安全问题及async/await替代方案

一、让Actor保持Publisher特性的安全实现 #

Actor的隔离特性要求其方法默认在隔离上下文执行，但Publisher协议的receive(subscriber:)方法必须在非隔离环境调用，可通过委托给线程安全的Combine Subject规避并发风险：

import Combine

enum State {
    case initial, loading, loaded
}

actor StateMachine: Publisher {
    typealias Output = State
    typealias Failure = Never
    
    // 用nonisolated持有线程安全的CurrentValueSubject，Combine的Subject内部自带线程安全机制
    nonisolated private let stateSubject = CurrentValueSubject<State, Never>(.initial)
    // Actor内部状态，受隔离保护
    private var currentState: State = .initial
    
    // 标记为nonisolated以符合Publisher协议，直接转发订阅请求给Subject
    nonisolated func receive<S>(subscriber: S) where S: Subscriber, Never == S.Failure, State == S.Input {
        stateSubject.receive(subscriber: subscriber)
    }
    
    // 状态转换方法在Actor隔离上下文执行，保证并发安全
    func transition(to newState: State) {
        currentState = newState
        stateSubject.send(newState)
    }
}

核心逻辑： #

• stateSubject是nonisolated属性，Combine的CurrentValueSubject本身线程安全，因此receive(subscriber:)的转发操作不会触发Actor隔离冲突。
• Actor内部的状态修改和发送都在隔离上下文完成，外部只能通过Actor的方法触发状态变化，确保并发安全。

二、Async/Await风格的替代方案：使用AsyncSequence #

如果想完全脱离Combine，改用原生async/await范式，可让Actor实现AsyncSequence，通过AsyncStream分发状态更新：

enum State {
    case initial, loading, loaded
}

actor StateMachine: AsyncSequence {
    typealias Element = State
    
    private var currentState: State = .initial
    private var continuation: AsyncStream<State>.Continuation?
    
    // 创建AsyncStream，保存continuation用于后续发送状态
    private lazy var stateStream: AsyncStream<State> = {
        AsyncStream { [weak self] continuation in
            self?.continuation = continuation
            // 订阅时先发送当前状态
            continuation.yield(self?.currentState ?? .initial)
        }
    }()
    
    func transition(to newState: State) {
        currentState = newState
        continuation?.yield(newState)
    }
    
    // 实现AsyncSequence核心方法，返回迭代器
    func makeAsyncIterator() -> AsyncStream<State>.Iterator {
        stateStream.makeAsyncIterator()
    }
}

使用方式： #

let machine = StateMachine()

// 启动异步任务订阅状态变化
Task {
    for await state in machine {
        print("当前状态：\(state)")
    }
}

// 触发状态转换
await machine.transition(to: .loading)
await machine.transition(to: .loaded)

优势： #

• 完全基于Swift原生并发模型，无需依赖Combine框架。
• Actor的隔离特性天然保证状态修改的并发安全，AsyncStream的分发逻辑符合async/await异步语义。

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