iOS自定义Combine Publisher实现单数据库监听优化需求

要实现单监听器共享+自动生命周期管理（有订阅则创建监听器，无订阅则销毁），同时让新订阅能立即获取最新数据，可以通过「单例数据源 + CurrentValueSubject + 订阅计数」的组合来实现，具体代码和说明如下：

核心实现代码 #

import Combine

class ItemsDataSource {
    // 单例实例，保证全局唯一
    static let shared = ItemsDataSource()
    
    // 对外暴露的只读Publisher，外部仅能订阅
    var itemsPublisher: AnyPublisher<[Item], Never> {
        subject.eraseToAnyPublisher()
    }
    
    // 保存最新数据的Subject，初始值从UserDefaults缓存读取
    private let subject = CurrentValueSubject<[Item], Never>(UserDefaults.standard.cacheItems)
    // 跟踪订阅数量，用于控制监听器生命周期
    private var subscriberCount = 0
    // 数据库监听器引用
    private var listener: DatabaseListener?
    
    // 私有初始化，禁止外部创建实例
    private init() {}
    
    // 处理订阅开始逻辑
    func trackSubscription() {
        subscriberCount += 1
        // 首次有订阅时创建数据库监听器
        if subscriberCount == 1 {
            listener = db.items.addListener { [weak self] newItems in
                guard let self = self else { return }
                // 更新本地缓存
                UserDefaults.standard.cacheItems = newItems
                // 推送新数据给所有订阅者
                self.subject.send(newItems)
            }
        }
    }
    
    // 处理订阅取消逻辑
    func trackCancellation() {
        subscriberCount -= 1
        // 无任何订阅时销毁监听器
        if subscriberCount <= 0 {
            listener?.cancel()
            listener = nil
            subscriberCount = 0 // 防止计数出现负数
        }
    }
}

// 自定义Subscription，负责关联订阅者与数据源
class ItemsSharedSubscription<S: Subscriber>: Subscription where S.Input == [Item], S.Failure == Never {
    private let subscriber: S
    private let dataSource = ItemsDataSource.shared
    
    init(subscriber: S) {
        self.subscriber = subscriber
        // 通知数据源新增订阅
        dataSource.trackSubscription()
    }
    
    func request(_ demand: Subscribers.Demand) {
        // 立即推送当前最新数据给订阅者
        if demand > .none {
            let _ = subscriber.receive(dataSource.subject.value)
        }
    }
    
    func cancel() {
        // 通知数据源取消订阅
        dataSource.trackCancellation()
    }
}

// 对外暴露的Publisher，返回共享的Subscription
struct ItemsSharedPublisher: Publisher {
    typealias Output = [Item]
    typealias Failure = Never
    
    func receive<S>(subscriber: S) where S: Subscriber, S.Input == [Item], S.Failure == Never {
        let subscription = ItemsSharedSubscription(subscriber: subscriber)
        subscriber.receive(subscription: subscription)
    }
}

使用方式 #

private var cancellables: Set<AnyCancellable> = []

// 所有订阅共用同一个数据库监听器
ItemsSharedPublisher()
    .receive(on: DispatchQueue.main) // 若数据库回调在后台，切换至主线程更新UI
    .sink { items in
        // 处理最新数据
    }
    .store(&cancellables)

关键细节说明 #

1. 单例数据源：ItemsDataSource全局唯一，确保所有订阅共享同一状态与监听器，避免重复创建高开销资源。
2. CurrentValueSubject：
   • 自动缓存最新数据，新订阅会立即收到当前缓存值
   • 数据库更新时，通过send(_:)统一推送数据给所有订阅者
3. 订阅计数管理：
   • 新增订阅时计数+1，计数从0→1时创建监听器
   • 取消订阅时计数-1，计数≤0时销毁监听器，释放资源
4. 线程安全：若数据库回调在后台线程，需通过receive(on:)切换至主线程，避免UI操作异常。

原方案问题分析 #

• 原ItemsPublisher每次实例化都会创建独立的ItemsSubscription，每个订阅对应一个监听器，导致资源浪费
• 全局ItemsPublisher未处理状态共享，新订阅无法获取已有最新数据，也无法共享监听器
• CurrentValueSubject本身可解决数据共享，但需额外的订阅计数来控制监听器的创建/销毁时机，这正是ItemsDataSource的核心作用

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