RxJava2 & Room：如何避免deleteAll()触发get()事件发射

这个问题我之前做Room开发的时候也碰到过——Room的@Query返回Flowable时，只要表数据有任何变更（包括清空操作），都会自动触发回调。要实现调用deleteAll()时阻止get()发射事件，有几个实用的思路：

方案1：RxJava流层面过滤（最简单高效） #

核心思路是用一个线程安全的标记，在执行deleteAll()时标记状态，然后在Flowable流中过滤掉这个状态下触发的空列表事件。

步骤1：添加线程安全标记 #

在你的Repository类中定义一个原子布尔变量：

private val isDeletingAll = AtomicBoolean(false)

步骤2：封装deleteAll()方法 #

不要直接调用Dao的deleteAll()，而是封装一个带标记的方法：

fun deleteAllWithoutTrigger() {
    isDeletingAll.set(true)
    try {
        db.userDao().deleteAll()
    } finally {
        // 无论操作成功失败，都要重置标记，避免影响后续事件
        isDeletingAll.set(false)
    }
}

步骤3：修改getUsers()方法过滤事件 #

在流中加入过滤逻辑，跳过deleteAll()触发的空列表：

override fun getUsers(): Flowable<List<User>> {
    // 如果你确实需要返回单个User的Flowable，后面加.flatMapIterable { it }即可
    return db.userDao().get()
        .filter { users ->
            // 只有当不是deleteAll状态，或者列表非空时，才发射事件
            !isDeletingAll.get() || users.isNotEmpty()
        }
        .distinctUntilChanged() // 避免重复发送相同的列表，优化性能
}

如果你的业务确实需要返回Flowable<User>（而不是List<User>），可以调整成这样：

override fun getUsers(): Flowable<User> {
    return db.userDao().get()
        .filter { users ->
            !isDeletingAll.get() || users.isNotEmpty()
        }
        .distinctUntilChanged()
        .flatMapIterable { it } // 把列表拆成单个User的流
}

方案2：自定义Room变更监听（精细控制） #

如果需要更底层的控制，可以自己实现Room的InvalidationTracker.Observer，手动决定何时发射事件。这个方法适合复杂场景，但代码量稍大：

private var userFlowable: Flowable<List<User>>? = null
private val isDeletingAll = AtomicBoolean(false)

override fun getUsers(): Flowable<List<User>> {
    if (userFlowable == null) {
        userFlowable = Flowable.create({ emitter ->
            // 创建自定义观察者，监听user表的变更
            val observer = object : InvalidationTracker.Observer("user") {
                override fun onInvalidated(tables: MutableSet<String>) {
                    // 只有当不是deleteAll状态时，才查询并发射数据
                    if (!isDeletingAll.get()) {
                        val users = db.userDao().get().blockingFirst()
                        emitter.onNext(users)
                    }
                }
            }
            // 添加观察者到数据库的失效跟踪器
            db.invalidationTracker.addObserver(observer)
            // 订阅取消时移除观察者，避免内存泄漏
            emitter.setCancellable {
                db.invalidationTracker.removeObserver(observer)
            }
            // 初始发射一次当前数据
            emitter.onNext(db.userDao().get().blockingFirst())
        }, BackpressureStrategy.LATEST)
            .share() // 共享流，避免重复创建观察者
    }
    return userFlowable!!
}

// 同样封装deleteAll方法
fun deleteAllWithoutTrigger() {
    isDeletingAll.set(true)
    try {
        db.userDao().deleteAll()
    } finally {
        isDeletingAll.set(false)
    }
}

注意事项 #

• 一定要用AtomicBoolean作为标记，因为Room的操作可能在不同线程执行，确保线程安全。
• try-finally块必须加，防止异常导致标记一直处于true状态，后续所有空列表事件都被过滤。
• distinctUntilChanged()是可选的，但建议加上，避免重复发送相同的列表，提升性能。

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