使用Observable.create()发射监听器对象时如何避免内存泄漏

问题分析 #

你的代码出现内存泄漏的核心原因有两个：

1. 全局持有ObservableEmitter和item实例，当DocumentRepository不再被使用时，这些引用会被Firestore的快照监听器间接持有，无法被GC回收
2. 没有在Observable被取消订阅（dispose）时移除Firestore的快照监听器，导致Firestore一直保持对监听器的引用，进而牵连整个DocumentRepository实例

修复后的代码实现 #

下面是重写后的类，解决了内存泄漏问题，同时优化了代码结构：

import com.google.firebase.firestore.DocumentReference
import com.google.firebase.firestore.DocumentSnapshot
import com.google.firebase.firestore.FirebaseFirestore
import com.google.firebase.firestore.ListenerRegistration
import io.reactivex.rxjava3.core.Observable
import io.reactivex.rxjava3.core.ObservableEmitter

class DocumentRepository<T : ModelWithMetadata>(
    path: List<String>, 
    private val model: Class<T>
) {
    private val documentReference: DocumentReference
    val observable: Observable<T>

    init {
        // 用fold简化文档引用的构建，比循环更简洁
        documentReference = path.chunked(2)
            .fold(FirebaseFirestore.getInstance() as DocumentReference?) { ref, pair ->
                when {
                    ref == null -> FirebaseFirestore.getInstance().collection(pair[0]).document(pair[1])
                    else -> ref.collection(pair[0]).document(pair[1])
                }
            } ?: throw IllegalArgumentException("Invalid document path")

        observable = Observable.create { emitter ->
            // 注册快照监听器并保存注册实例
            val listenerRegistration = documentReference.addSnapshotListener { snapshot, error ->
                error?.let {
                    emitter.onError(it)
                    return@addSnapshotListener
                }
                // 处理快照为空或者转换失败的情况，避免NPE
                snapshot?.toObject(model)?.let { item ->
                    emitter.onNext(item)
                } ?: run {
                    emitter.onError(NullPointerException("Document snapshot or converted model is null"))
                }
            }

            // 当Observable被dispose时，自动移除Firestore监听器，释放资源
            emitter.setCancellable {
                listenerRegistration.remove()
            }
        }
        // 可选：如果需要在主线程接收数据，可以添加observeOn
        // .observeOn(AndroidSchedulers.mainThread())
    }

    fun put(item: T) {
        item.updatedAt = TimeExtension.now()
        documentReference.set(item)
    }

    fun delete() {
        documentReference.delete()
    }

    // 如果需要主动获取当前最新数据，可以用这个方法返回Single
    fun getCurrent(): io.reactivex.rxjava3.core.Single<T> {
        return documentReference.get()
            .map { snapshot ->
                snapshot.toObject(model) ?: throw NullPointerException("Document not found or conversion failed")
            }
    }
}

关键优化点说明 #

• 移除全局引用：不再持有emitter和item的全局实例，所有状态都在Observable的create闭包内管理，避免不必要的引用持有
• 监听器生命周期绑定：通过emitter.setCancellable()，在Observable被dispose时自动调用listenerRegistration.remove()，彻底移除Firestore的快照监听器，切断引用链
• 空安全处理：增加了快照和模型转换的空判断，避免空指针异常
• 代码简化：用chunked(2)和fold替代原有的循环构建文档引用，逻辑更清晰
• 优化get方法：原get()方法依赖全局item，现在改为返回Single<T>，主动获取最新数据，更符合RxJava的设计理念

相关学习资源推荐 #

• RxJava官方指南：重点关注Observable.create()的最佳实践，尤其是资源清理（Cancellable的使用），这是避免RxJava相关内存泄漏的核心
• Firebase Firestore文档：深入理解快照监听器的生命周期，明确知道必须在组件（如Activity/Fragment）销毁时移除监听器
• RxFirebase社区实践：很多开源项目的实现思路都是将Firebase的回调API转换为RxJava流，核心都是在流的生命周期内绑定监听器的注册与销毁，你可以参考这类实现的设计逻辑

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