从案例学 RxAndroid 开发（上）

原文链接： RxAndroid Basics: Part 1

如果你在阅读这篇文章，相信你一定很想了解 RxJava 以及如何在 Android 应用中使用它。可能你已经见过 RxJava 的代码了，但仍然有些疑惑，愿你能在这篇文章里找到答案。

当我第一次使用 RxJava 的时候我只是在照搬代码，这些代码能跑起来，但是我对 RxJava 的基础部分仍然存在误解，而且我找不到好的源码来学习。所以为了理解 RxJava，我不得不一点一点学习，踩了不少坑。

为了不让你把我踩过的坑再踩一遍，我会基于我的学习成果写一些例子出来，目的就是让你能够对 RxJava 有足够的了解，并能在你的 Android 应用中使用它。

源码可以在这里找到。在每个例子的开始，我会写清每个代码段是属于哪个Activity 的。我会将本文分为两个部分，在第一部分里，我会着重讲解如何用RxJava 异步加载数据；在第二部分里，我会探索一些更高级的用法。

几个概念

在开始说代码之前，先澄清几个概念。RxJava 最核心的东西就是Observable 和Observer。Observable 会发出数据，而与之相对的Observer 则会通过订阅Observable 来进行观察。

Observer 可以在 Observable 发出数据、报错或者声明没有数据可以发送时进行相应的操作。这三个操作被封装在 Observer 接口中，相应的方法为 onNext()，onError() 和 onCompleted()。

明确了这些概念以后，让我们来看一些例子。

案例 1：基础

现在要写一个用来展示一个颜色列表的 Activity。我们要写一个能发送一个字符串列表、然后结束的 Observeable。而后我们会通过这个字符串列表来填充颜色列表，这里要使用到 Observable.just() 方法。由这个方法创建的 Observable 对象的特点是：所有 Observer 一旦订阅这个 Observable 就会立即调用 onNext() 方法并传入 Observable.just() 的参数，而后因为 Observable 没有数据可以发送了，onComplete() 方法会被调用。

Observable<List<String>> listObservable = Observable.just(getColorList());注意这里的 getColorList() 是一个不耗时的方法。虽然现在看来这个方法无足轻重，但一会我们会回到这个方法。

下一步，我们写一个 Observer 来观察 Observable。

listObservable.subscribe(new Observer<List<String>>() { 
 
    @Override 
    public void onCompleted() { } 
 
    @Override 
    public void onError(Throwable e) { } 
 
    @Override
    public void onNext(List<String> colors) {
        mSimpleStringAdapter.setStrings(colors);
    }
});

而后神奇的事情就发生了。如我刚才所说，一旦通过 subscribe() 方法订阅 Observable，就会发生一系列事情：

1. onNext() 方法被调用，被发送的颜色列表会作为参数传入。
2. 既然不再有数据可以发送（我们在 Observable.just() 中只让 Observable 发送一个数据），onComplete() 方法会被调用。

请记住：通过 Observable 被订阅后的行为来区分它们。

在这个例子中我们不关心 Observable 何时完成数据的传输，所以我们不用在 onComplete() 方法里写代码。而且在这里不会有异常抛出，所以我们也不用管 onError() 方法。

写了这么多你可能觉得很多余，毕竟我们本可以在 adapter 中直接设置作为数据源的颜色列表。请带着这个疑问，和我看下面这个更有趣一些的例子。

案例 2：异步加载

在这里我们要写一个显示电视剧列表的 Activity 。在 Android 中 RxJava 的主要用途就在于异步数据加载。首先让我们写一个 Observable：

Observable<List<String>> tvShowObservable = Observable.fromCallable(new Callable<List<String>>() { 
 
    @Override 
    public List<String> call() { 
        return mRestClient.getFavoriteTvShows(); 
    }
});

在刚才的例子中，我们使用 Observable.just() 来创建 Observable，你可能认为在这里可以通过 Observable.just(mRestClient.getFavoriteTvShows()) 来创建 Observable。

但在这里我们不能这么做，因为 mRestClient.getFavoriteTvShows() 会发起网络请求。如果在这里我们使用 Observable.just()，mRestClient.getFavoriteTvShows() 会被立即执行并阻塞 UI 线程。

使用 Observable.fromCallable() 方法有两点好处：

1. 获取要发送的数据的代码只会在有 Observer 订阅之后执行。
2. 获取数据的代码可以在子线程中执行。

这两点好处有时可能非常重要。现在让我们订阅这个 Observable。

mTvShowSubscription = tvShowObservable
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer<List<String>>() {
 
        @Override 
        public void onCompleted() { }
 
        @Override 
        public void onError(Throwable e) { }
 
        @Override 
        public void onNext(List<String> tvShows){
            displayTvShows(tvShows);
        }
    });

让我们一个方法一个方法地来看这段代码。subscribeOn 会修改我们刚刚创建的 Observable。在默认情况下 Observable 的所有代码，包括刚才说到的只有在被订阅之后才会执行的代码，都会在执行 subscribe() 方法的线程中运行。而通过 subscribeOn() 方法，这些代码可以在其他线程中执行。但具体是哪个线程呢？

在这个例子中我们让代码在"IO Scheduler"中执行（Schedulers.io()）。现在我们可以只把 Scheduler 当做一个可以工作的子线程，这个描述对于现在的我们已经足够了，不过这其中还有更深层次的内容。

不过我们的确遇到了一个小障碍。既然 Observable 会在 IO Scheduler 中运行，那么它与 Observer 的连接也会在 IO Scheduler 中完成。这就意味着 Observer 的 onNext() 方法也会在 IO Scheduler 中运行，而 onNext() 方法会操作 UI 中的 View，但 View 只能在 UI 主线程中操作。

事实上解决这个问题也很简单，我们可以告诉 RxJava 我们要在 UI 线程中观察这个 Observable，也就是，我们想让 onNext() 方法在 UI 线程中执行。这一点我们可以通过在 observeOn() 方法中指定另一个 Scheduler 来完成，在这里也就是 AndroidSchedules.mainThread() 所返回的 Scheduler(UI 线程的 Scheduler)。

而后我们调用 subscribe() 方法。这个方法最重要，因为 Callable 只会在有 Observer 订阅后运行。还记得刚才我说 Observable 通过其被订阅后的行为来区分吗？这就是一个很好的例子。

还有最后一件事。这个 mTvShowSubscription 到底是什么？每当 Observer 订阅 Observable 时就会生成一个 Subscription 对象。一个 Subscription 代表了一个 Observer 与 Observable 之间的连接。有时我们需要操作这个连接，这里拿在 Activity 的 onDestroy() 方法中的代码举个例子：

if (mTvShowSubscription != null && !mTvShowSubscription.isUnsubscribed()) {
    mTvShowSubscription.unsubscribe();
}

如果你与多线程打过交道，你肯定会意识到一个大坑：当 Activity 执行 onDestroy() 后线程才结束（甚至永不结束）的话，就有可能发生内存泄漏与 NullPointerException 空指针异常。

Subscription 就可以解决这个问题，我们可以通过调用 unsubscribe() 方法告诉 Observable 它所发送的数据不再被 Observer 所接收。在调用 unsubscribe() 方法后，我们创建的 Observer 就不再会收到数据了，同时也就解决了刚才说的问题。

说到这里难点已经过去，让我们来总结一下：

• Observable.fromCallable() 方法可以拖延 Observable 获取数据的操作，这一点在数据需要在其他线程获取时尤其重要。
• subscribeOn() 让我们在指定线程中运行获取数据的代码，只要不是 UI 线程就行。
• observeOn() 让我们在合适的线程中接收 Observable 发送的数据，在这里是 UI 主线程。
• 记住要让 Observer 取消订阅以免 Observable 异步加载数据时发生意外。

案例 3：使用 Single

这次我们还是写一个展示电视剧列表的 Activity ，但这次我们走一种更简单的风格。Observable 挺好用的，但在某些情况下过于重量级。比如说，你可能一经发现在过去的两个方法中我们只是让 Observable 发送一个数据，而且我们从来也没写过 onComplete() 回调方法。

其实呢，Observable 还有一个精简版，叫做 Single。Single 几乎和 Observable 一模一样，但其回调方法不是 onComplete()/onNext()/onError()，而是 onSuccess()/onError()。

我们现在把刚才写过的 Observable 用 Single 重写一遍。首先我们要创建一个 Single:

Single<List<String>> tvShowSingle = Single.fromCallable(new Callable<List<String>>() { 
    @Override
    public List<String> call() throws Exception {
        mRestClient.getFavoriteTvShows(); 
    }
});

然后订阅一下：

mTvShowSubscription = tvShowSingle
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new SingleSubscriber<List<String>>() {
 
        @Override 
        public void onSuccess(List<String> tvShows) {
            displayTvShows(tvShows); 
        }
 
        @Override 
        public void onError(Throwable error) {
            displayErrorMessage(); 
        } 
    });

这段代码和刚才很像，我们调用 subscribeOn() 方法以确保 getFavoriteTvShows() 在子线程中执行。而后我们调用 observeOn() 以确保 Single 的数据被发送到 UI 线程。

但这次我们不再使用 Observer，而是使用一个叫 SingleSubscriber 的类。这个类和 Observer 非常像，只不过它只有上述两个方法：onSuccess() 和 onError()。SingleSubscriber 之于 Single 就如 Observer 之于 Observable。

订阅一个 Single 的同时也会自动创建一个 Subscription 对象。这里的 Subscription 和案例 2 中没有区别，一定要在 onDestroy() 中解除订阅。

最后一点：在这里我们添加了处理异常的代码，所以如果 mRestClient 出了问题，onError() 就会被调用。建议你亲手写一个案例玩一玩，体验一下有异常时程序是怎么运行的。

结语

第一部分就说这么多了，希望这几个案例能对你有所帮助，不要忘了看看第二部分中更加高级的使用方法哦。