10、RxJava与Room：RxJava基础回顾、在DAO中返回Flowable/Single、线程调度、背压处理

说实话，很多Android开发者刚接触Room时，都习惯用LiveData做数据观察。LiveData确实简单，但一旦涉及到复杂的线程切换、数据流变换、或者需要处理背压的场景，它就有点力不从心了。这时候，RxJava就派上了用场。

我个人习惯把Room + RxJava称为“黄金搭档”。Room天然支持RxJava的返回类型，你可以在DAO方法里直接返回Flowable、Single、Maybe、Completable等。配合RxJava的线程调度和操作符，数据流的处理会变得非常优雅。

这一章，我们就来聊聊怎么把RxJava和Room结合起来用。我会从基础回顾开始，然后深入到DAO中的具体用法，最后聊聊背压这个容易踩坑的点。

10.1 RxJava基础回顾

RxJava的核心思想，说白了就是“观察者模式 + 异步 + 操作符”。你有一个数据源（Observable），一个观察者（Observer），中间可以加各种操作符来变换数据、切换线程。

我记得刚开始学RxJava时，最头疼的就是各种类型的区别：Observable、Flowable、Single、Completable、Maybe。其实你只要记住一个原则：根据数据量选类型。

嗯，这里要注意：Room的DAO中，最常用的是Flowable和Single。Flowable用于监听数据变化（类似LiveData），Single用于一次性查询。

10.2 在DAO中返回Flowable/Single

我们先看一个简单的例子。假设你有一个用户表，你想监听所有用户的变化。用LiveData是这样写的：

@Dao
public interface UserDao {
    @Query("SELECT * FROM users")
    LiveData<List<User>> getAllUsers();
}

换成RxJava的Flowable，只需要改一下返回类型：

@Dao
public interface UserDao {
    @Query("SELECT * FROM users")
    Flowable<List<User>> getAllUsers();
}

就这么简单。Room会自动帮你把数据库的变更事件转换成Flowable的发射事件。每次表数据有变化，Flowable就会重新查询并发射新的结果。

如果你只需要查一次，不想监听后续变化，那就用Single：

@Dao
public interface UserDao {
    @Query("SELECT * FROM users WHERE id = :userId")
    Single<User> getUserById(int userId);
}

Single只会发射一次数据，然后结束。我在项目中经常用它来做“详情页”的数据加载——用户进入页面时查一次，离开页面时自动取消订阅，非常干净。

10.3 线程调度

RxJava最强大的地方之一就是线程调度。Room默认是在后台线程执行数据库操作的，但如果你不指定调度器，订阅者（观察者）会运行在哪个线程呢？答案是：跟发射者同一个线程。

这就有问题了。如果你在DAO中返回Flowable，然后直接在UI线程订阅，Room会抛异常——因为数据库操作不能在主线程执行。但Flowable的订阅回调又默认运行在发射线程。嗯，这里有点绕，我解释一下。

实际上，Room的Flowable是在后台线程发射数据的。如果你不切换线程，订阅者的onNext也会在后台线程执行。这时候你如果在onNext里更新UI，就会报错：

// 错误示例：直接在后台线程更新UI
userDao.getAllUsers()
    .subscribe(users -> {
        // 这里运行在Room的后台线程
        textView.setText(users.size()); // 崩溃！不能在子线程更新UI
    });

正确的做法是使用observeOn切换到主线程：

userDao.getAllUsers()
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(users -> {
        // 现在运行在主线程，可以更新UI
        textView.setText(users.size());
    });

我个人习惯在ViewModel中做线程调度，这样Activity/Fragment只关心UI更新：

public class UserViewModel extends ViewModel {
    private final UserDao userDao;
    private final CompositeDisposable disposables = new CompositeDisposable();

    public UserViewModel(UserDao userDao) {
        this.userDao = userDao;
    }

    public Flowable<List<User>> getUsers() {
        return userDao.getAllUsers()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread());
    }

    @Override
    protected void onCleared() {
        super.onCleared();
        disposables.clear(); // 防止内存泄漏
    }
}

10.4 背压处理

背压（Backpressure）是RxJava中一个比较高级的概念。简单来说，就是生产者发射数据的速度 > 消费者处理数据的速度时，数据会堆积，最终导致内存溢出。

为什么会这样？你想想看，如果你的数据库表频繁变化（比如每秒插入100条记录），Flowable就会每秒发射100次新列表。如果你的UI处理速度跟不上（比如列表渲染需要200ms），数据就会在内存中堆积。

Flowable天生支持背压，而Observable不支持。这也是为什么我建议在Room的DAO中优先使用Flowable而不是Observable。

Flowable提供了多种背压策略，你可以通过onBackpressureBuffer()、onBackpressureDrop()、onBackpressureLatest()来控制：

我曾经在一个聊天应用里遇到过背压问题。用户发消息非常频繁，每次消息入库都会触发Flowable重新查询整个聊天记录列表。列表有几千条数据，渲染一次要几百毫秒。结果就是：用户发了10条消息，列表只渲染了3次，剩下的7次数据被积压在内存里。

解决方案很简单，用onBackpressureLatest()：

userDao.getMessages(chatId)
    .onBackpressureLatest()
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(messages -> {
        // 只处理最新的数据，中间的数据被丢弃
        adapter.submitList(messages);
    });

这样，即使数据库变化再频繁，UI也只会处理最新的那一次数据。中间的变化被自动丢弃，内存压力大大降低。

10.5 知识体系总览

下面这张图总结了本章的核心逻辑，你可以把它当作一个速查表：

这张图把本章的核心知识点串起来了。你从DAO返回类型出发，根据需求选择Flowable或Single，然后通过线程调度确保数据在正确的线程上流动，最后用背压策略应对高频数据变更。每一步都有对应的RxJava操作符，组合起来就是一个完整的数据管道。

公众号:蓝海资料掘金营，微信deep3321