原文地址
在Room中使用RxJava进行查询

更少的模板代码，编译时SQL查询检查，除此之外，异步和响应式查询能力——听起来怎么样？所有的这些，使用Room数据库（Architecture Components中的持久化库
）都是可能的。异步查询可以返回 LiveData 或者RxJava中的 Maybe, Single和 Flowable。其中返回的LiveData和Flowable是响应式的查询。他们在数据变化时，能确保UI可以自动更新为数据库中的最新数值。如果你已经在APP中使用RxJava 2，那么在Room中使用 Maybe、Single 、Flowable将变得简单。

备注：从 2.0.0-beta01开始，Room也支持Observable

备注2：从2.1.0-alpha01开始，DAO中使用@Insert, @Delete或@Update 声明的方法，支持Rx的Completable, Single<T>和 Maybe<T>等返回类型

让我们考虑如下的界面：用户可以查看和编辑他们的名称。这里和用户有关的其他一些信息，已经保存在数据库中。这里展示如何插入、更新、删除和查询用户。

Insert

集成了RxJava的Room针对insert支持如下返回类型：

• Completable — 当insert完成时，马上调用onComplete
• Single<Long>/ Maybe<Long> — onSuccess发射的值，是插入项的rowID
• Single<List<Long>> /Maybe<List<Long>> —  onSuccess发射的值，是插入列表的rowID列表

@Insert
Completable insert(User user);
// or
@Insert
Maybe<Long> insert(User user);
// or
@Insert
Single<Long> insert(User[] user);
// or
@Insert
Maybe<List<Long>> insert(User[] user);
// or
@Insert
Single<List<Long>> insert(User[] user);

使用observeOn操作符来指定Scheduler在哪个线程Observer观察Observable,使用subscribeOn来指定Observable在哪个线程执行。

Update/Delete

集成了RxJava的Room针对 update/delete支持如下返回类型：

• Completable — 当update/delete完成时，马上调用onComplete
• Single<Long>/ Maybe<Long> — onSuccess发射的值，是update/delete影响的行数

@Update
Completable update(User user);
// or
@Update
Single<Integer> update(User user);
// or
@Update
Single<Integer> updateAll(User[] user);
// or
@Delete
Single<Integer> deleteAll(User[] user);
// or
@Delete
Single<Integer> deleteAll(User[] user);

使用observeOn操作符来指定Scheduler在哪个线程Observer观察Observable,使用subscribeOn来指定Observable在哪个线程执行。

Query

为了从数据库中获取用户，我们可以在数据访问对象类(UserDao)中写如下的查询:

@Query(“SELECT * FROM Users WHERE id = :userId”)
User getUserById(String userId);

这种方式有2大缺点：

1. 它是阻塞的、同步的访问
2. 当数据被修改时，我们每次都需要手动调用这个方法

借助RxJava中的 Maybe, Single 和Flowable 对象
，Room提供了观察数据库中数据和执行异步查询的方式。

如果你担心线程问题，Room能够轻松的保证观察查询在非主线程完成。通过在observeOn方式设置Scheduler，下游的线程发送事件在哪个线程完全由你决定。

查询操作返回Maybe或Single,确保调用了Scheduler的subscribeOn，并且不要用AndroidSchedulers.mainThread()。

为了开始在Room中使用RxJava 2，只需要在
build.gradle中添加如下依赖：

// RxJava support for Room
implementation “android.arch.persistence.room:rxjava2:1.0.0-alpha5”
// Testing support
androidTestImplementation “android.arch.core:core-testing:1.0.0-alpha5”

Maybe

@Query(“SELECT * FROM Users WHERE id = :userId”)
Maybe<User> getUserById(String userId);

这里将发生：

1. 当数据库中没有用户，查询将不返回行，Maybe会完成。
2. 当数据库中有一个用户，Maybe会触发onSuccess，然后将完成。
3. Maybe完成后，如果这个用户有更新，什么都不会发生。

Single

@Query(“SELECT * FROM Users WHERE id = :userId”)
Single<User> getUserById(String userId);

这里分几种情况：

1. 当数据库中没有用户，查询将不返回行，Single将触发onError(EmptyResultSetException.class)
2. 当数据库中有一个用户，Single会触发onSuccess。
3. Single完成后，如果这个用户有更新，什么都不会发生。

Flowable/Observable

@Query(“SELECT * FROM Users WHERE id = :userId”)
Flowable<User> getUserById(String userId);

Flowable/Observable有如下表现方式：

1. 当数据库中没有用户，查询将不返回行，Flowable将不会触发，不管是onNext还是onError。
2. 当数据库中有一个用户，Flowable会触发onNext。
3. 如果这个用户有更新，Flowable会自动触发，允许你来基于最新的数据更新UI。

Testing

测试返回Maybe/Single/Flowable的查询，和同步查询并无什么不同。UserDaoTest中，我们确保使用一个in-memory数据库，因为在其中保存的信息在进程被杀时，会自动清除。

@RunWith(AndroidJUnit4.class)
public class UserDaoTest {
…
private UsersDatabase mDatabase;
@Before
public void initDb() throws Exception {
    mDatabase = Room.inMemoryDatabaseBuilder(
                     InstrumentationRegistry.getContext(),
                     UsersDatabase.class)
            // allowing main thread queries, just for testing
            .allowMainThreadQueries()
            .build();
}

@After
public void closeDb() throws Exception {
    mDatabase.close();
}

在测试中添加InstantTaskExecutorRule，来使Room数据库立即执行所有数据库操作。

@Rule
public InstantTaskExecutorRule instantTaskExecutorRule = 
                                      new InstantTaskExecutorRule();

让我们实现getUserById的发射订阅，并且检查当用户真正插入时，数据是否正确的被Flowable发射。

@Test
public void insertAndGetUserById() {
    // Given that we have a user in the data source
    mDatabase.userDao().insertUser(USER);
    // When subscribing to the emissions of user
    mDatabase.userDao()
             .getUserById(USER.getId())
             .test()
             // assertValue asserts that there was only one emission
             .assertValue(new Predicate<User>() {
                @Override
                public boolean test(User user) throws Exception {
                    // The emitted user is the expected one
                    return user.getId().equals(USER.getId()) &&
                      user.getUserName().equals(USER.getUserName());
                }
            });
}

就这样！如果你在APP中使用RxJava 2，使你的数据库变成响应式来确保你的UI始终显示最新的数据。点击这里检出一个使用Room&RxJava的示例APP。

参考文章

1. RxJava & Room
2. Android Room 使用以及配合 RxJava
3. 在Room中使用RxJava