RxSwift判断Observable是否已被dispose及移除.debug后失效问题求解

你当前的实现逻辑本身存在错误，靠弱引用持有Observable来判断订阅是否销毁的思路从根上不成立：Observable只是数据流的定义模板，和订阅的生命周期没有绑定关系。
之前保留.debug操作符时能正常运行只是巧合：.debug作为中间操作符会强持有上游的Observable实例，只要订阅还存在，你的弱引用就不会被置空；移除.debug后，你生成的Observable没有任何强引用持有，函数返回后就会立刻释放，所以弱引用直接变成nil。

RxSwift本身提供了监听订阅销毁事件的能力，你只需要在返回的Observable上追加do(onDispose:)操作符，就可以在消费者dispose订阅时触发回调，完成订阅者列表的清理，不需要轮询判断弱引用状态。

修改后的代码如下：

private class Subscriber {
    let ids: [Int]
    // 用于唯一标识订阅者，方便后续查找移除
    let identifier = UUID()
}

class Publisher {
    private let relay: BehaviorRelay<[Int: Updates]> = .init(value: [:])
    private var subscribers: [Subscriber] = []
    // 增加串行队列保证数组操作线程安全，避免多线程操作崩溃
    private let queue = DispatchQueue(label: "com.publisher.queue")

    func updatesStream(for ids: [Int]) -> Observable<[Updates]> {
        let observable = relay
            .map { map in
                return map
                    .filter { ids.contains($0.key) }
                    .map { $0.value }
            }
            .filter { !$0.isEmpty }
            .asObservable()

        let subscriber = Subscriber(ids: ids)
        // 队列中执行插入操作
        queue.sync {
            subscribers.append(subscriber)
        }
        
        // 追加dispose监听
        return observable
            .do(onDispose: { [weak self] in
                guard let self = self else { return }
                self.queue.sync {
                    self.subscribers.removeAll { $0.identifier == subscriber.identifier }
                }
            })
    }

    private func repeatTimer() {
        let updates: [Updates] = [] // 你的业务逻辑生成updates
        
        // 原有清理逻辑可以完全删除，订阅销毁时会自动清理列表
        relay.accept(updates)
    }
}

• 所有对subscribers数组的增删操作都放在同一个串行队列中执行，避免多线程访问导致的数据竞争崩溃。
• 如果你的业务场景中需要获取当前有效订阅的ids列表，直接从队列中读取subscribers的ids即可。

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