参考：给 Android 开发者的 RxJava 详解
GitHub：RxJava、RxAndroid
// Build.gradle配置文件引入
compile 'io.reactivex:rxandroid:1.1.0'
compile 'io.reactivex:rxjava:1.1.0'
1、RxJava可以做什么
RxJava是一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库。可以轻松实现线程切换的方式。平时我们做线程切换的几种方式：
- Handler
这个应该是Android常用的一种方式，利用线程的Handler发送消息到相应的线程Looper里面，那么这个任务也将会在该线程执行。
handler.sendEmptyMessage(EVENT_DOWNLOAD_DATA);
Handler另外的一种发送消息的方式：
getUiHandler().post(new Runnable() {
@Override
public void run() {
// do something
}
});
- new Thread()
新建一个子线程执行耗时操作，需要切换到主线更新UI时
getActivity().runOnUiThread(new Runnable() {
@Override
public void run() {
// do something
}
});
- AsyncTask
内部有个线程池执行异常任务doInBackground()，在UI线程中执行任务前onPreExecute()、更新进度onProgressUpdate()、完成任务onPostExecute()的方法。 - EventBus、otto
比较通用的事件消息总线开源框架，对于普通的线程切换来说，也是比较重的一种方式。 - 另外还有广播、Service等不大灵活的线程切换方式。
相必，以上的几种方式使用起来都不是那么的简洁，业务复杂起来，还会导致逻辑上的不清晰，那么RxJava就可以做到逻辑的简洁，执行异步任务、线程切换都可以在一条链上完成。
2、观察者模式
观察者模式的例子：
- Button点击事件
Button被观察者，持有引用，OnClickListener观察者，通过setOnClickListener()注册/订阅点击事件，发生onClick()事件。 - 设置监听回调函数
比如执行选择会员弹窗Dialog。Dialog被观察者，持有引用，OnMemberSelectListener观察者，通过setOnMemberSelectListener()注册/订阅选择会员事件，发生onMemberSelect(Member member)事件。
3、RxJava观察者模式实现步骤
3.1、创建观察者 Observer
Observable被观察者 和 Observer观察者 通过 subscribe() 方法实现订阅关系
事件。
onNext()
onCompleted()
onError()
onStart() // 这是 Subscriber 增加的方法。它总是在 subscribe 所发生的线程被调用。不太适用于执行任务前的弹窗等UI操作。
unsubscribe() // 这是 Subscriber 所实现的另一个接口 Subscription 的方法，用于取消订阅。避免内存泄漏。
Observer 接口，Observer 的抽象类：Subscriber
Subscriber<String> subscriber = new Subscriber<String>() {
@Override
public void onNext(String s) {
Log.d(tag, "Item: " + s);
}
@Override
public void onCompleted() {
Log.d(tag, "Completed!");
}
@Override
public void onError(Throwable e) {
Log.d(tag, "Error!");
}
};
3.2、创建被观察者 Observable
create() 方法是 RxJava 最基本的创造事件序列的方法
Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("Hello");
subscriber.onNext("Hi");
subscriber.onNext("Aloha");
subscriber.onCompleted();
}
});
另外还提供了一些方法用来快捷创建事件队列
// just()
Observable observable = Observable.just("Hello", "Hi", "Aloha");
// from()
String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
3.3、Subscribe (订阅)
observable.subscribe(observer);
// 或者：
observable.subscribe(subscriber);
观察者被观察者换位置：对流式 API 的设计会造成影响。
在 RxJava 中， Observable 并不是在创建的时候就立即开始发送事件，而是在它被订阅的时候。
除了 subscribe(Observer) 和 subscribe(Subscriber) ，subscribe() 还支持不完整定义的回调
// 自动创建 Subscriber ，并使用 onNextAction 来定义 onNext()
observable.subscribe(onNextAction);
// 自动创建 Subscriber ，并使用 onNextAction 和 onErrorAction 来定义 onNext() 和 onError()
observable.subscribe(onNextAction, onErrorAction);
// 自动创建 Subscriber ，并使用 onNextAction、 onErrorAction 和 onCompletedAction 来定义 onNext()、 onError() 和 onCompleted()
observable.subscribe(onNextAction, onErrorAction, onCompletedAction);
Action0 是 RxJava 的一个接口，它只有一个方法 call()，这个方法是无参无返回值的；由于 onCompleted() 方法也是无参无返回值的，因此 Action0 可以被当成一个包装对象，将 onCompleted() 的内容打包起来将自己作为一个参数传入 subscribe() 以实现不完整定义的回调。这样其实也可以看做将 onCompleted() 方法作为参数传进了 subscribe()，相当于其他某些语言中的『闭包』。
4、RxJava操作符
变换：就是将事件序列中的对象或整个序列进行加工处理，转换成不同的事件或事件序列
FuncX 和 ActionX 的区别在 FuncX 包装的是有返回值的方法。
4.1、map()
事件对象的直接变换
Observable.from(students)
.map(new Func1<Student, String>() {
@Override
public String call(Student student) {
return student.getName();
}
})
.subscribe(subscriber);
4.2、flatMap()
Observable.from(students)
.flatMap(new Func1<Student, Observable<Course>>() {
@Override
public Observable<Course> call(Student student) {
return Observable.from(student.getCourses());
}
})
.subscribe(subscriber);
flatMap() 中返回的是个 Observable 对象，并且这个 Observable 对象并不是被直接发送到了 Subscriber 的回调方法中。 flatMap() 的原理是这样的：
- 使用传入的事件对象创建一个 Observable 对象；
- 并不发送这个 Observable, 而是将它激活，于是它开始发送事件；
- 每一个创建出来的 Observable 发送的事件，都被汇入同一个 Observable ，而这个 Observable 负责将这些事件统一交给 Subscriber 的回调方法。
由于可以在嵌套的 Observable 中添加异步代码， flatMap() 也常用于嵌套的异步操作，例如嵌套的网络请求。
5、Scheduler 线程切换
- Schedulers.immediate(): 直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler。
- Schedulers.newThread(): 总是启用新线程，并在新线程执行操作。
- Schedulers.io(): I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和 newThread() 差不多，区别在于 io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中，可以避免创建不必要的线程。
- Schedulers.computation(): 计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU。
- AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。
- subscribeOn()：subscribe() 所发生的线程，即Observable.OnSubscribe 被激活时所处的线程。或者叫做事件产生的线程。
- observeOn()：Subscriber 所运行在的线程。或者叫做事件消费的线程。
Observable.just(1, 2, 3, 4) // IO 线程，由 subscribeOn() 指定
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.newThread())
.map(mapOperator) // 新线程，由 observeOn() 指定
.observeOn(Schedulers.io())
.map(mapOperator2) // IO 线程，由 observeOn() 指定
.observeOn(AndroidSchedulers.mainThread)
.subscribe(subscriber); // Android 主线程，由 observeOn() 指定
通过 observeOn() 的多次调用，程序实现了线程的多次切换。
不过，不同于 observeOn() ， subscribeOn() 的位置放在哪里都可以，但它是只能调用一次的。
- doOnSubscribe()
与 Subscriber.onStart() 相对应的，有一个方法 Observable.doOnSubscribe() 。它和 Subscriber.onStart() 同样是在 subscribe() 调用后而且在事件发送前执行，但区别在于它可以指定线程。默认情况下， doOnSubscribe() 执行在 subscribe() 发生的线程；而如果在 doOnSubscribe() 之后有 subscribeOn() 的话，它将执行在离它最近的 subscribeOn() 所指定的线程。
Observable.create(onSubscribe)
.subscribeOn(Schedulers.io())
.doOnSubscribe(new Action0() {
@Override
public void call() {
progressBar.setVisibility(View.VISIBLE); // 需要在主线程执行
}
})
.subscribeOn(AndroidSchedulers.mainThread()) // 指定主线程
.observeOn(AndroidSchedulers.mainThread())
.subscribe(subscriber);
