1．基本概念

Rx是RxJava针对Android的定制版本。这个版本中通过增加最少的类使在Android应用中编写响应式组件简单而且无障碍，特别之处在与它还提供了一个Scheduler，可以在主线程或任何给定的Handler上进行调度。Rx编程方式基于三个基本概念：观察者模式、迭代器模式、函数式编程。下面通过几个例子，描述在Android上面Rx方式进行编程的基本使用。

2．观察者模式

在Android Studio里面引入rxjava和rxandroid两个依赖包后，我们就可以进行rx开发了：

compile 'io.reactivex:rxjava:1.1.6'

compile 'io.reactivex:rxandroid:1.2.1'

Rx的两个核心类是Observable（被观察者）和Observer（观察者）。前者发送数据和消息，后者处理数据和消息。Observer收到数据时，它的onNext()方法会被调用，数据在方法的参数里面被传入。另外Observer还有onComplete()方法，来表示Observable的数据发送完毕，而onError()则表示数据发送过程中出错。

先看一个例子。假设要发送一系列的http请求，这些请求的相对路径我们已经有了：“/home”, “/productList”, “/productDetail”，但是发送的时候需要使用绝对路径，因此可以这样拼接一下：

String[] urls = {"/home", "/productList", "/productDetails"};
for (String url : urls) {sendHttpRequest(BASE_URL + url);
}

如果用RxAndroid写的话，可以这样做：

 Observable observable = Observable

            .just("/home", "/productList", "/productDetails");
Observer<String> observer = new Observer<String>() {public void onCompleted() {}public void onError(Throwable e) {}

public void onNext(String s) {sendHttpRequest(BASE_URL + url);}
};
observable.subscribe(observer);

单从代码量来讲，RxAndroid似乎是把事情搞复杂了。这个我们先不讨论。先看看RxAndroid代码的基本逻辑：

1）创建一个Observable，它发送了一系列的数据。

2）创建一个Observer，它的onNext()方法会接收到数据。另外两个方法暂时不感兴趣，不作任何处理。

3）Observer订阅Observable。Subscribe方法将两者关联起来。

实际上，上述代码还有一种简单的写法：

Observable.just("/home", "/productList", "/productDetails").subscribe(new Action1<String>() {@Overridepublic void call(String s) {sendHttpRequest(BASE_URL + url);}});

因为对事件结束和事件出错不感兴趣，所以我们可以用Action1类来接收和处理数据。但是过程是一样的：发送数据；订阅；接收数据。

3．异步编程

RxAndroid一个最重要的功能就是简化异步编程。e.g.假设你要读取手机里面的所有联系人，然后在UI上用一个ListView把联系人一个个显示出来，可以这样做：

Observable.fromCallable(new Callable<Cursor>() {@Overridepublic Cursor call() throws Exception {return queryContacts();                    // ①}}).subscribeOn(Schedulers.io())                      // 设置I.observeOn(AndroidSchedulers.mainThread())       // 设置II.subscribe(new Action1<Cursor>() {@Overridepublic void call(Cursor contacts) {showContacts();                             // ②}});

这里面，queryContacts涉及到IO操作，因此,需要保证它在一个异步线程里面运行。因此，使用了一个Callable对象，来包裹这个IO操作。Callable的call()方法只有在Observable被订阅后才会执行，并且你可以指定call()方法在哪个线程里面执行。我们使用了subscribeOn(Schedulers.io()), 指定Callable.call()方法在一个单独的IO线程里面执行。但是一旦加上这一行，后面的Action1.call()方法也会在异步线程里面执行，所以还得加上一句observeOn(AndroidSchedulers.mainThread())
，它的作用是让Action1.call()在主线程里面执行。

总结一下，假设当前线程为主线程，以下表格列出来了不同情况下代码①和代码②的线程关系。

在Android里面，异步编程还有一个需要注意的地方，那就是Activity销毁之后，需要停止线程。上面的代码，在Activity.onDestroy()里面，应该停止对Observable的订阅，因为这个时候，线程可能还没执行完毕，这个时候，需要停止对数据的接收。只要停止订阅就可以了，如何停止订阅呢？实际上，上面的代码最终会返回一个Subscription对象：

Subscription subscription = Observable.fromCallable(..

        .subscribeOn(...

        .observeOn(...)

这个Subscription对象代表Observable和Observer的一个连接，我们可以这样来停止订阅：

subscription.unsubscribe();

4．操作符(Operator)

在Observer接收到数据前，可以用操作符可以对Observable发出的数据进行转换。还是上面的例子，假设我们在读取联系人后，要根据联系人的名字生成对应的图片，并且，只列出来姓“李”的联系人，我们可以这样做：

Observable.fromCallable(new Callable<List<File>>() {@Overridepublic Cursor call() throws Exception {return queryContacts(); // ①}})

        .map(new Func1<Cursor, List<Contact>() {@Overridepublic List<Contact> call(Cursor c) {return getContactsFromCursor(c);           // ④}}).flatMap(new Func1<List<Contact>, Observable<Contact>>() {@Overridepublic Observable<Contact> call(List<Contact> contacts) {return Observable.from(contacts);                   // ②}}).filter(new Func1<Contact, Boolean>() {@Overridepublic Boolean call(Contact c) {return c.getName().startWith("黄");        // ③}})

.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Action1<Contact>() {@Overridepublic void call(Contact c) {showContact(c);}});

这段代码里面展示了RxAndroid里面三个常用的操作符，flatMap，filter，map. 前面提到，操作符是在Observer接收到数据之前，对Observable发送的数据进行中间转换。以下是转换过程：

步骤①：此时所有数据都存在一个List里面，如果没有其它操作符的转换，我们最后的Observer也会接收到一个List，也就是说，数据是一次性接收到的。但是，需求是图片是分批、一个一个传过来，因此我们需要将list展开。

步骤② flatMap操作符：这一步实际上是将上一步的Observable转换成了一个新的Observable，转换过程中，输入是List<Contact>类型，输出是Contac类型。发送的数据并没有减少，只是发送方式变了，由一次性发送变成一个个发送。

步骤③ filter操作符：这一步对数据进行了过滤，只有姓李的联系人才会被发送到Observer

步骤④ map操作符：这一步对数据进行了类型转换。我们读Cursor，并生成联系人列表。输入的是Cursor，输出的是List<Contact>。

5．Subject

回顾我们第一个例子，我们的Observer，Observable都是临时对象，数据处理完毕之后，它们也完成了自己的生命周期。假设我们再来一批数据，并且数据处理流程一样，我们是不是需要再创建新的Observable和Observer？代码重新写一遍？

Observable.just("/home", "/productList", "/productDetails").subscribe(new Action1<String>() {@Overridepublic void call(String s) {}});

使用RxAndroid里面的Subject对象可以解决这个问题。Subject既是Observable，也是Observer，可以把它想象成一个管道：可以随时往一端输入数据，经过中间处理后，另一端处理输出。考虑这样一个场景，你在联系人列表里面进行搜索，你在搜索框里面每输入一个字符，联系人都会实时重新查询一遍。这里面就是一个标准的Rx流程：输入->处理（查询）->输出。只不过输入条件会不断的变，然后这个流程需要重复执行。

mSearchView.setOnQueryTextListener(new OnQueryTextListener() {@Overridepublic boolean onQueryTextSubmit(String query) {return false;}

@Overridepublic boolean onQueryTextChange(String query) {mSubject.onNext(query);return true;}
});

mSubject = PublishSubject.create();
mSubject.debounce(400, TimeUnit.MILLISECONDS).observeOn(Schedulers.io()).map(new Func1<String, List<Contact>>() {@Overridepublic List<Contact> call(String s) {return queryContacts(s);}}).observeOn(AndroidSchedulers.mainThread()).subscribe(new Action1<Contact>() {@Overridepublic void call(Contact contact) {addContactToList(contact);}});

这里面，searchView每次输入有变化时，都会重新查询一遍联系人数据库。因此我们用一个全局变量保存subject，这样可以调用subject.onNext()多次发送数据。这里有两个需要说明的地方：

①：我们用了一个debounce方法，它的作用是，当400ms内没有新的数据输入时，subject才会发送数据。因为用户每输入一个字符时，searchView的onQueryTextChange方法都会被触发。而我们不需要这么频繁的查询。

②：我们用了两个observeOn来指定中间操作和最后操作的执行线程。查询的时候，我们需要在异步线程执行，而最终把联系人加入到查询结果列表的时候，我们又需要切换到主线程。

参考文档：

① 从案例学RxAndroid开发：

http://www.infoq.com/cn/articles/RxAndroid-basics

② Rx官网Subject文档：

http://reactivex.io/documentation/subject.html