Swift AsyncStream多观察者问题：如何实现类Combine多订阅者行为？

问题根源 #

你当前使用AsyncStream.makeStream()创建的是单播异步序列，同一个AsyncStream的迭代器具有排他性：当一个迭代器开始遍历元素时，元素会被该迭代器消费，其他迭代器无法获取已被消费的元素。这就是为什么多个同类型订阅者仅最后一个能收到更新、不同类型订阅者会互相抢占消费元素的核心原因。

在Swift Concurrency中，完全可以实现类似Combine PassthroughSubject的多播（多观察者）行为，以下是两种可行方案：

方案1：使用AsyncChannel（推荐） #

Swift官方的swift-async-algorithms库提供了AsyncChannel，它是原生支持多消费者的异步通道，每个订阅者都能收到所有发送的元素，行为和PassthroughSubject完全一致。

步骤1：添加依赖 #

通过Swift Package Manager引入库：

.package(url: "https://github.com/apple/swift-async-algorithms", from: "1.0.0")

步骤2：修改代码 #

import AsyncAlgorithms

final class MyComponent {
    private let receivedMessageChannel = AsyncChannel<Data>()
    private let decoder = JSONDecoder() // 确保你已初始化该实例

    func commandStream<C: Decodable>() -> AsyncStream<C> {
        AsyncStream<C> { continuation in
            Task {
                for await data in receivedMessageChannel {
                    guard let commandWrapper = try? decoder.decode(CommandWrapper.self, from: data),
                          commandWrapper.commandType == String(describing: C.self) else {
                        continue
                    }
                    
                    if let command = try? decoder.decode(C.self, from: commandWrapper.commandData) {
                        continuation.yield(command)
                    }
                }
                continuation.finish()
            }
        }
    }

    // 收到新数据时发送到通道
    func receiveNewData(_ data: Data) {
        Task {
            await receivedMessageChannel.send(data)
        }
    }
}

这样每个调用commandStream()的订阅者都会独立遍历通道，所有符合类型的元素都会被分发给对应订阅者，不会出现抢占消费的问题。

方案2：自定义多播容器（无第三方依赖） #

如果不想依赖外部库，可以自己实现一个简单的多播管理容器，维护所有订阅者的continuation，当新元素到达时分发给匹配类型的订阅者：

final class MyComponent {
    private typealias ContinuationMap = [String: [Any]]
    private var continuations: ContinuationMap = [:]
    private let decoder = JSONDecoder()
    private let queue = DispatchQueue(label: "com.yourcomponent.multiCastQueue")

    func commandStream<C: Decodable>() -> AsyncStream<C> {
        AsyncStream<C> { continuation in
            let typeKey = String(describing: C.self)
            
            // 线程安全地添加订阅者
            queue.sync {
                if continuations[typeKey] == nil {
                    continuations[typeKey] = []
                }
                continuations[typeKey]?.append(continuation)
            }

            // 订阅终止时移除continuation
            continuation.onTermination = { [weak self] _ in
                self?.queue.sync {
                    self?.continuations[typeKey]?.removeAll { $0 as? AsyncStream<C>.Continuation === continuation }
                    if self?.continuations[typeKey]?.isEmpty == true {
                        self?.continuations.removeValue(forKey: typeKey)
                    }
                }
            }
        }
    }

    func receiveNewData(_ data: Data) {
        queue.async { [weak self] in
            guard let self = self else { return }
            guard let commandWrapper = try? self.decoder.decode(CommandWrapper.self, from: data) else {
                return
            }
            
            let typeKey = commandWrapper.commandType
            guard let typeContinuations = self.continuations[typeKey],
                  let commandData = commandWrapper.commandData else {
                return
            }
            
            // 解码元素并分发给所有对应订阅者
            if let command = try? self.decoder.decode(AnyDecodable.self, from: commandData) {
                for continuation in typeContinuations {
                    (continuation as? AsyncStream<AnyDecodable>.Continuation)?.yield(command.value)
                }
            }
        }
    }
}

// 辅助类型：支持动态解码任意Decodable类型
struct AnyDecodable: Decodable {
    let value: Any

    init(from decoder: Decoder) throws {
        let container = try decoder.singleValueContainer()
        if let int = try? container.decode(Int.self) {
            value = int
        } else if let string = try? container.decode(String.self) {
            value = string
        } else if let bool = try? container.decode(Bool.self) {
            value = bool
        } else if let array = try? container.decode([AnyDecodable].self) {
            value = array.map { $0.value }
        } else if let dict = try? container.decode([String: AnyDecodable].self) {
            value = dict.mapValues { $0.value }
        } else {
            throw DecodingError.dataCorruptedError(in: container, debugDescription: "Unsupported data type")
        }
    }
}

自定义方案需要手动处理线程安全和类型转换，适合对依赖有严格限制的场景。

核心原理总结 #

• AsyncStream是单播设计，元素仅能被一个迭代器消费；
• AsyncChannel是多播设计，支持多个并发迭代器共享所有元素；
• 自定义多播容器的核心是维护订阅者列表，将新元素主动分发给所有匹配的订阅者。

内容的提问来源于stack exchange，提问作者José Roberto Abreu