本文出自 “阿敏其人” 简书博客，转载或引用请注明出处。
新的技术点的普及总是需要一些现在前面苦心探索，感谢
大头鬼Bruce、
扔物线 给 Android 开发者的 RxJava 详解。
等的开源奉献。
一、RxJava初认识
RxJava用来做什么？
异步，就像AsyncTask 和Handler等，但是更加强大。随着程序逻辑变得越来越复杂，它依然能够保持简洁。
RxJava最大的特点
观察者模式
后台处理，前台回调 的异步机制
观察者模式是一种经典的设计模式，网上有很多相关文章，经典的Java观察者的学习可以参阅此文——23种设计模式（8）：观察者模式
我要观察你，你变动就要告诉我。程序你要听话。
安卓的中广播就是一种特别明显的观察者模式，我们注册了Home键监听，home被按下，你就要告诉我。
RxJava 的观察者模式
RxJava 有四个基本概念：
- Observable (可观察者，即被观察者)
- Observer (观察者)
- subscribe (订阅)
- 事件（Observer有默认三个事件）
Observable 和 Observer 通过 subscribe() 方法实现订阅关系，从而 Observable 可以在需要的时候发出事件来通知 Observer。
少年我们看图吧
我知道你不爱美人不爱江山，只爱代码，好吧
（按照常理来说应该是 观察者 订阅 被观察者 但是为了api设计需要，Rx这里是 被观察者 订阅 观察者 ，这个需要明确一下）
四个基本概念中，被观察者和观察者通过订阅产生关系，这个我们好理解，但是关于事件我们需要掰扯一下。
RxJava的观察者有2个，一个是Observer，另外一个是Subscriber。
Subscriber是实现了接口的Observer 的抽象类，他们的基本使用方式一样，但是拥有的事件确不同。
Observer，除了普通事件 onNext() （相当于 onClick() / onEvent()）之外，定义了两个特殊的事件：onCompleted() 和 onError()。
Subscriber 在Observer的基础之上，还有拥有两个特殊的方法。
观察者Observer的三个事件
onNext(T t) 观察到变化之后要做的逻辑 ，关键工作都在这里
onCompleted(): 事件队列完结。RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。RxJava 规定，当不会再有新的 onNext() 发出时，需要触发 onCompleted() 方法作为标志。
onError(): 事件队列异常。在事件处理过程中出异常时，onError() 会被触发，同时队列自动终止，不允许再有事件发出。
在一个正确运行的事件序列中, onCompleted() 和 onError() 有且只有一个，并且是事件序列中的最后一个。需要注意的是，onCompleted() 和 onError() 二者也是互斥的，即在队列中调用了其中一个，就不应该再调用另一个。
Subscriber 特有的两个方法：
onStart(): 这是 Subscriber 增加的方法。它会在 subscribe 刚开始，而事件还未发送之前被调用，可以用于做一些准备工作，例如数据的清零或重置。这是一个可选方法，默认情况下它的实现为空。
需要注意的是，如果对准备工作的线程有要求（例如弹出一个显示进度的对话框，这必须在主线程执行）， onStart() 就不适用了，因为它总是在 subscribe 所发生的线程被调用，而不能指定线程。要在指定的线程来做准备工作，可以使用 doOnSubscribe() 方法，具体可以在后面的文中看到。unsubscribe(): 这是 Subscriber 所实现的另一个接口 Subscription 的方法，用于取消订阅。
在这个方法被调用后，Subscriber 将不再接收事件。一般在这个方法调用前，可以使用 isUnsubscribed() 先判断一下状态。 unsubscribe() 这个方法很重要，因为在 subscribe() 之后， Observable 会持有 Subscriber 的引用，这个引用如果不能及时被释放，将有内存泄露的风险。
所以最好保持一个原则：要在不再使用的时候尽快在合适的地方（例如 onPause() onStop() 等方法中）调用 unsubscribe() 来解除引用关系，以避免内存泄露的发生。
我的是我的，你的也是我的， Observer都是Subscriber的
在 RxJava 的 subscribe 过程中，Observer 也总是会先被转换成一个 Subscriber 再使用。所以如果你只想使用基本功能，选择 Observer 和 Subscriber 是完全一样的，但是Observer在 subscribe()过程中最终会被转换成Subscriber对象。
所以，你用了Observer和在订阅后其实就相当于你用了Subscriber。
二、RxJava代码示例
as biuld文件引入
compile 'io.reactivex:rxjava:1.1.6'
compile 'io.reactivex:rxandroid:1.2.1'
引入io.reactivex:rxjava自不必说，引入io.reactivex:rxandroid是为了切换到安卓主线程等。
成功引入RxJava
二、1 最简单的示例
第一步 创建 观察者 Observer
Observer<String> observer = new Observer<String>() {
// 观察者的三个事件 onNext onCompleted onError
@Override
public void onNext(String s) {
System.out.println("=======我是观察者 Observer 的 onNext 事件，打印参数s："+s);
}
@Override
public void onCompleted() {
System.out.println("=======我是观察者 Observer 的 onCompleted 事件，无参数");
}
@Override
public void onError(Throwable e) {
System.out.println("=======我是观察者 Observer 的 onError 事件，参数 e："+e);
}
};
观察者的是 new 出来的，然后有三个事件需要复写。
第二步 创建 被观察者 Observable
Observable 即被观察者，它决定什么时候触发事件以及触发怎样的事件。
Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("onNext Fist");
subscriber.onNext("onNext Secind");
subscriber.onNext("onNext Third");
subscriber.onStart();
subscriber.onCompleted();
}
});
第一行代码中我们用 create()方法来创建一个 Observable ，并为它定义事件触发规则。
第三行代码开始我们看到了call方法，当 Observable 被订阅的时候，OnSubscribe 的 call() 方法会自动被调用，事件序列就会依照设定依次触发（对于上面的代码，就是观察者Subscriber 将会被调用三次 onNext() 和一次 onCompleted()）
由被观察者调用了观察者的回调方法，就实现了由被观察者向观察者的事件传递，即观察者模式。（金句也，被观察者 调用了 观察者的 回调，数据从 被观察者 手里交给 观察者，此乃观察者模式，其实真正干活的还是观察者本身，其实这就是回调的工作原理，被观察者是一个被监控的闲情大爷，关于回调不太熟悉的同学可以看下此 野文）
第三步，订阅
// 被观察者 订阅 观察者 是的，好像是有点别扭，但是还是需要接受
observable.subscribe(observer);
完整的代码运行和分析
完整代码
public class MainActivity extends Activity {
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
Observer<String> observer = new Observer<String>() {
// 观察者的三个事件 onNext onCompleted onError
@Override
public void onNext(String s) {
System.out.println("=======我是观察者 Observer 的 onNext 事件，打印参数s：" + s);
}
@Override
public void onCompleted() {
System.out.println("=======我是观察者 Observer 的 onCompleted 事件，无参数");
}
@Override
public void onError(Throwable e) {
System.out.println("=======我是观察者 Observer 的 onError 事件，参数 e：" + e);
}
};
Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("=======onNext Fist");
subscriber.onNext("=======onNext Secind");
subscriber.onNext("=======onNext Third");
subscriber.onStart();
subscriber.onCompleted();
}
});
// 被观察者 订阅 观察者 是的，好像是有点别扭，但是还是需要接受
observable.subscribe(observer);
}
}
打印：
=======我是观察者 Observer 的 onNext 事件，打印参数s：=======onNext Fist
=======我是观察者 Observer 的 onNext 事件，打印参数s：=======onNext Secind
=======我是观察者 Observer 的 onNext 事件，打印参数s：=======onNext Third
=======我是观察者 Observer 的 onCompleted 事件，无参数
我们在 被观察者 的 call 方法里面调用了三次我们可以和一次subscriber.onStart()以及一次subscriber.onCompleted();
订阅关系产生之后，被观察者 会三次onNext会将数据（此处为String）传输给 观察者。观察者在事件中接受到传输过来的数据，然后可以执行其他操作。
最后执行subscriber.onCompleted()回调，打印最后一句语句。
需要注意的是，call里面的代码是排放是需要区分顺序的的，不是说你回调了三次 subscriber.onNext就一定得等你执行完成之后再来执行 subscriber.onCompleted()。
假设我们把call 里面的代码改成如下：
Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("=======onNext Fist");
subscriber.onCompleted();
subscriber.onNext("=======onNext Secind");
subscriber.onNext("=======onNext Third");
subscriber.onStart();
}
});
那么打印结果就是：
=======我是观察者 Observer 的 onNext 事件，打印参数s：=======onNext Fist
=======我是观察者 Observer 的 onCompleted 事件，无参数
如上，起始也就是，当你执行了观察者回调了subscriber.onCompleted();，那么这个订阅关系就会解绑，解绑了你的onNext关我毛事。
Observer + Observable 的简单代码到此为止
Subscriber + Observable
之前说过，对于观察而言。Observer的在产生订阅关系之后，就自动转成c。
我知道你是个长情的汉子（男/女），怎会抛下 Subscriber 于不顾。
好，来，把 Observer 换成 Subscriber。
public class MainActivity extends Activity {
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
Subscriber<String> subscriber = new Subscriber<String>() {
// 观察者的三个事件 onNext onCompleted onError
// 因为是Subscriber,所以有特有的onStart方法
@Override
public void onNext(String s) {
System.out.println("=======我是观察者 Subscriber 的 onNext 事件，打印参数s：" + s);
}
@Override
public void onCompleted() {
System.out.println("=======我是观察者 Subscriber 的 onCompleted 事件，无参数");
}
@Override
public void onError(Throwable e) {
System.out.println("=======我是观察者 Subscriber 的 onError 事件，参数 e：" + e);
}
@Override
public void onStart() {
super.onStart();
System.out.println("=======我是观察者 Subscriber 特有的 onStart ，无参数");
}
};
Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("=======onNext Fist");
subscriber.onStart(); // subscriber.onStart()的执行要先于subscriber.onNext
subscriber.onNext("=======onNext Secind");
subscriber.onNext("=======onNext Third");
subscriber.onCompleted();
}
});
// 被观察者 订阅 观察者 是的，好像是有点别扭，但是还是需要接受
observable.subscribe(subscriber);
}
}
new 出来一个Subscriber，和Observer区别不大。
关键是人家一出来就是Subsrciber，（Observer需要订阅之后才转成Subsrciber），所以人家具有特有的onStar方法，这个时候我们的小小伏笔subscriber.onStart()也可以登场。
打印结果
=======我是观察者 Subscriber 特有的 onStart ，无参数
=======我是观察者 Subscriber 的 onNext 事件，打印参数s：=======onNext Fist
=======我是观察者 Subscriber 的 onNext 事件，打印参数s：=======onNext Secind
=======我是观察者 Subscriber 的 onNext 事件，打印参数s：=======onNext Third
=======我是观察者 Subscriber 的 onCompleted 事件，无参数
从上面的打印我们可以看出来：
1、被观察者的 call 方法回调的subscriber.onStart()打印生效了，因为在观察者Subscriber里面具有onStart所以执行了打印。
2、onStart可以做准备动作，因为onStart是先于onNext执行的。
Subscriber做了这么3件事：
1、调用 Subscriber.onStart()这个方法在前面已经介绍过，是一个可选的准备方法。
2、调用 Observable中的 OnSubscribe.call(Subscriber) 。在这里，事件发送的逻辑开始运行。从这也可以看出：
** 在 RxJava 中，Observable并不是在创建的时候就立即开始发送事件，而是在它被订阅的时候，即当 subscribe()方法执行的时候。**
3、将传入的 Subscriber
作为 Subscription
返回。这是为了方便 unsubscribe()
.
对了，你是不是想问我为什么不用Log非要sout？其实是因为我这个电脑得了一种Log坏了而且一直医不好的病。所以我才一直sout tab。。
三、女娲造人，我们造 被观察者
从上面的两个小例子中，我们的 被观察者 都是 Observable.create(new Observable.OnSubscribe<String>()) 造出来的
造 被观察者，除了create，RxJava 还提供了一些方法用来快捷创建事件队列，比如 just 和 from
Observable.just(T...) 造 被观察者
手动传入n个类型的参数
代码：
核心
Observable<String> observable = Observable.just("Hello","R","x","Java");
完整
public class MainActivity extends Activity {
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
Subscriber<String> subscriber = new Subscriber<String>() {
// 观察者的三个事件 onNext onCompleted onError
// 因为是Subscriber,所以有特有的onStart方法
@Override
public void onNext(String s) {
System.out.println("=======我是观察者 Subscriber 的 onNext 事件，打印参数s：" + s);
}
@Override
public void onCompleted() {
System.out.println("=======我是观察者 Subscriber 的 onCompleted 事件，无参数");
}
@Override
public void onError(Throwable e) {
System.out.println("=======我是观察者 Subscriber 的 onError 事件，参数 e：" + e);
}
@Override
public void onStart() {
super.onStart();
System.out.println("=======我是观察者 Subscriber 特有的 onStart ，无参数");
}
};
Observable<String> observable = Observable.just("Hello","R","x","Java");
// 被观察者 订阅 观察者 是的，好像是有点别扭，但是还是需要接受
observable.subscribe(subscriber);
}
}
打印：
=======我是观察者 Subscriber 特有的 onStart ，无参数
=======我是观察者 Subscriber 的 onNext 事件，打印参数s：Hello
=======我是观察者 Subscriber 的 onNext 事件，打印参数s：R
=======我是观察者 Subscriber 的 onNext 事件，打印参数s：x
=======我是观察者 Subscriber 的 onNext 事件，打印参数s：Java
=======我是观察者 Subscriber 的 onCompleted 事件，无参数
发现：
1、相当于回调传值了n次onNext
2、onStart和onCompleted也被回调了
from(T[ ])/ from(Iterable<? extends T>) 造 被观察者
from(T[ ]) / from(Iterable<? extends T>) : 将传入的数组或 Iterable 拆分成具体对象后，依次回调onNext
代码：
核心
String[] names = {"Hello","R","x","Java"};
Observable<String> observable = Observable.from(names);
完整
public class MainActivity extends Activity {
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
Subscriber<String> subscriber = new Subscriber<String>() {
// 观察者的三个事件 onNext onCompleted onError
// 因为是Subscriber,所以有特有的onStart方法
@Override
public void onNext(String s) {
System.out.println("=======我是观察者 Subscriber 的 onNext 事件，打印参数s：" + s);
}
@Override
public void onCompleted() {
System.out.println("=======我是观察者 Subscriber 的 onCompleted 事件，无参数");
}
@Override
public void onError(Throwable e) {
System.out.println("=======我是观察者 Subscriber 的 onError 事件，参数 e：" + e);
}
@Override
public void onStart() {
super.onStart();
System.out.println("=======我是观察者 Subscriber 特有的 onStart ，无参数");
}
};
String[] names = {"Hello","R","x","Java"};
Observable<String> observable = Observable.from(names);
// 被观察者 订阅 观察者 是的，好像是有点别扭，但是还是需要接受
observable.subscribe(subscriber);
}
}
打印结果：
=======我是观察者 Subscriber 特有的 onStart ，无参数
=======我是观察者 Subscriber 的 onNext 事件，打印参数s：Hello
=======我是观察者 Subscriber 的 onNext 事件，打印参数s：R
=======我是观察者 Subscriber 的 onNext 事件，打印参数s：x
=======我是观察者 Subscriber 的 onNext 事件，打印参数s：Java
=======我是观察者 Subscriber 的 onCompleted 事件，无参数
结论就不需要了。
四、不完整的旋律，不完整的回调
上面看到了 被观察者 可以任性创建，观察者 表示不服，所以天赋神权，让 观察者 可以 不完整回调。
这个神权叫做 ActionX
先看代码再说道道
public class MainActivity extends Activity {
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
// 不完整回调 的 观察者
Action1<String> onNextAction = new Action1<String>() {
// onNext()
@Override
public void call(String s) {
System.out.println("=======我是 Action1 onNextAction 不完整观察者 的 call: "+s);
}
};
// 任性创建的 被观察者
String[] names = {"任","性","被观察者","传入"};
Observable<String> observable = Observable.from(names);
observable.subscribe(onNextAction);
}
}
打印结果：
=======我是 Action1 onNextAction 不完整观察者 的 call: 任
=======我是 Action1 onNextAction 不完整观察者 的 call: 性
=======我是 Action1 onNextAction 不完整观察者 的 call: 被观察者
=======我是 Action1 onNextAction 不完整观察者 的 call: 传入
完整起来也可以善良很可爱
虽然有不完整的神权，但是也完整起来也是可以相当完善的。
拼凑完整：
public class MainActivity extends Activity {
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
// 不完整回调 的 观察者
Action1<String> onNextAction = new Action1<String>() {
// onNext()
@Override
public void call(String s) {
System.out.println("=======我是 Action1 onNextAction 不完整观察者 的 call: "+s);
}
};
Action1<Throwable> onErrorAction = new Action1<Throwable>() {
// onError()
@Override
public void call(Throwable throwable) {
System.out.println("=======我是 Action1 onErrorAction 不完整观察者 的 call: "+throwable.toString());
}
};
Action0 onCompletedAction = new Action0() {
// onCompleted()
@Override
public void call() {
System.out.println("=======我是 Action0 onCompletedAction 不完整观察者 的 call");
}
};
// 任性创建的 被观察者
String[] names = {"任","性","被观察者","传入"};
Observable<String> observable = Observable.from(names);
// 完整起来也可以善良很可爱,但是只给传入3个ActionX
observable.subscribe(onNextAction,onErrorAction,onCompletedAction);
}
}
打印输出：
=======我是 Action1 onNextAction 不完整观察者 的 call: 任
=======我是 Action1 onNextAction 不完整观察者 的 call: 性
=======我是 Action1 onNextAction 不完整观察者 的 call: 被观察者
=======我是 Action1 onNextAction 不完整观察者 的 call: 传入
=======我是 Action0 onCompletedAction 不完整观察者 的 call
Action0 是 RxJava 的一个接口，它只有一个方法 call()，这个方法是无参无返回值的；
由于 onCompleted() 方法也是无参无返回值的，因此 Action0 可以被当成一个包装对象，将 onCompleted() 的内容打包起来将自己作为一个参数传入 subscribe() 以实现不完整定义的回调。这样其实也可以看做将 onCompleted() 方法作为参数传进了 subscribe()，相当于其他某些语言中的闭包。
Action1 也是一个接口，它同样只有一个方法 call(T param)，这个方法也无返回值，但有一个参数；与 Action0 同理，由于 onNext(T obj) 和 onError(Throwable error) 也是单参数无返回值的，因此 Action1 可以将 onNext(obj) 和 onError(error) 打包起来传入 subscribe() 以实现不完整定义的回调。
事实上，虽然 Action0 和 Action1 在 API 中使用最广泛，但 RxJava 是提供了多个 ActionX 形式的接口 (例如 Action2, Action3) 的，它们可以被用以包装不同的无返回值的方法。
ActionX的一个重要特征就是：无返回值（可能有参）
五、放开你的单线程，让Scheduler来
引用一下 扔物线同学 的归纳
在 RxJava 的默认规则中，事件的发出和消费都是在同一个线程的。也就是说，如果只用上面的方法，实现出来的只是一个同步的观察者模式。观察者模式本身的目的就是 后台处理，前台回调 的异步机制，因此异步对于 RxJava 是至关重要的。而要实现异步，则需要用到 RxJava 的另一个概念： Scheduler 。
在不指定线程的情况下， RxJava 遵循的是线程不变的原则，即：在哪个线程调用 subscribe()，就在哪个线程生产事件；在哪个线程生产事件，就在哪个线程消费事件。如果需要切换线程，就需要用到 Scheduler （调度器）。
Rxjava常用调度器
在RxJava 中，Scheduler ——调度器，相当于线程控制器，RxJava 通过它来指定每一段代码应该运行在什么样的线程。RxJava 已经内置了几个 Scheduler ，它们已经适合大多数的使用场景：
- 1、Schedulers.immediate(): 直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler。
- 2、Schedulers.newThread(): 总是启用新线程，并在新线程执行操作。
- 3、Schedulers.io(): I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和 newThread() 差不多，区别在于 io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中，可以避免创建不必要的线程。
- 4、Schedulers.computation(): 计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU。
- 5、AndroidSchedulers.mainThread()， Android 专用AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。
切换线程/切换调度器
常用这么几个调度器，那么现在我们只要知道如何切换线程了。
- subscribeOn(): 指定 subscribe() 所发生的线程，即 Observable.OnSubscribe 被激活时所处的线程。或者叫做事件产生的线程。
- observeOn(): 指定 Subscriber 所运行在的线程。或者叫做事件消费的线程。
看代码：
public class MainActivity extends Activity {
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
Observable.just("你", "好", "线", "程")
.subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
.subscribe(new Action1<String>() {
@Override
public void call(String str) {
System.out.println("====不完整的观察者 得到的数据："+str);
}
});
}
}
打印输出：
====不完整的观察者 得到的数据：你
====不完整的观察者 得到的数据：好
====不完整的观察者 得到的数据：线
====不完整的观察者 得到的数据：程
观察者和被观察者与线程之间的关系大概如下：
.subscribe...这一行代码被执行的时候，订阅关系正式产生。
被观察者要发送几个事件观察者，但是因为已经有指定线程的两句代码、
.subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
- 开始的线程：所以 被观察者 和 观察者 一开始是会进入io线程，从io线程发出处理，然后处理完毕之后（假设很耗时）
- 结束的线程：（io线程处理完毕之后）切换进入进入Android主线程。
（RxJava 中，Observable并不是在创建的时候就立即开始发送事件，而是在它被订阅的时候，即当 subscribe()方法执行的时候）
这种在 subscribe() 之前写上两句 subscribeOn(Scheduler.io()) 和 observeOn(AndroidSchedulers.mainThread()) 的使用方式非常常见，它适用于多数的 **后台线程取数据，主线程显示 ** 的程序策略。这就意味着，即使加载图片或数据库等操作操作花费了几十甚至几百毫秒的时间，也不会造成丝毫界面的卡顿。
问：被消费者和消费者谁在前台？谁在后台？
不是这么算的，他们是在一起，只是在哪一个阶段被切换到了哪一个线程/调度器。
Scheduler自由控制线程
- observeOn() 消费的线程 可以多次切换
- 当使用了多个 subscribeOn() 的时候，只有第一个 subscribeOn() 起作用。
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() 的位置放在哪里都可以，但它是只能调用一次的。
不同的是， subscribeOn() 的线程切换发生在 OnSubscribe 中，即在它通知上一级 OnSubscribe 时，这时事件还没有开始发送，因此 subscribeOn() 的线程控制可以从事件发出的开端就造成影响；而 observeOn() 的线程切换则发生在它内建的 Subscriber 中，即发生在它即将给下一级 Subscriber 发送事件时，因此 observeOn() 控制的是它后面的线程。
讲 Subscriber 的时候，提到过 Subscriber 的 onStart() 可以用作流程开始前的初始化。然而 onStart() 由于在 subscribe() 发生时就被调用了，因此不能指定线程，而是只能执行在 subscribe() 被调用时的线程。这就导致如果 onStart() 中含有对线程有要求的代码（例如在界面上显示一个 ProgressBar，这必须在主线程执行），将会有线程非法的风险，因为有时你无法预测 subscribe() 将会在什么线程执行。
与 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);
如上，在 doOnSubscribe()的后面跟一个 subscribeOn() ，就能指定准备工作的线程了。
六、变换
所谓变换，就是将事件序列中的对象或整个序列进行加工处理，转换成不同的事件或事件序列
有点晦涩，那我们先来看一个简单的代码，先不进行变换，像之前那样，被观察者和观察者 使用的是相同的数据类型。看代码：
就是一个代码加载图片的小demo
public class MainActivity extends Activity {
private ImageView mIv;
private int resId = R.mipmap.ic_launcher;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
mIv = (ImageView) findViewById(R.id.mIv);
Observable.create(new Observable.OnSubscribe<Drawable>() {
@Override
public void call(Subscriber<? super Drawable> subscriber) {
Drawable drawable = ContextCompat.getDrawable(MainActivity.this,resId);
subscriber.onNext(drawable);
subscriber.onCompleted();
}
}).subscribe(new Observer<Drawable>() {
@Override
public void onNext(Drawable drawable) {
mIv.setImageDrawable(drawable);
}
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
Toast.makeText(MainActivity.this, "Error!", Toast.LENGTH_SHORT).show();
}
});
}
}
画一画吧
这个我们熟悉。
接下来，我们来让 被观察者 传入的是int，但是最后 观察者 使用的数据类型是 Drawable
变换一下。
public class MainActivity extends Activity {
private ImageView mIv;
private int resId = R.mipmap.ic_launcher;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
mIv = (ImageView) findViewById(R.id.mIv);
Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
subscriber.onNext(R.mipmap.ic_launcher); //R.mipmap.ic_launcher在R里面对应一个int值
}
}).map(new Func1<Integer, Drawable>() {
@Override
public Drawable call(Integer integer) {
Drawable drawable = ContextCompat.getDrawable(MainActivity.this,integer);
return drawable;
}
}).subscribe(new Subscriber<Drawable>() {
@Override
public void onCompleted() {
System.out.println("======= onCompleted");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Drawable drawable) {
mIv.setBackgroundDrawable(drawable);
}
});
}
}
没错，上面出现了一个我们从没用过的map和Func1。这个待会再讲，我知道少年你觉得不就代码设置一个图片吗，搞这么一大长串代码，真是毁三观。但是这是为了说明作用必须的呀，比如我们后面结合Retrofit就可以见其效力，其实把这个打通了，后面各种变换还不是可以让我们耍得溜溜的。
不说了，先把上面的代码弄短一些吧，再说map和Func1
public class MainActivity extends Activity {
private ImageView mIv;
private int resId = R.mipmap.ic_launcher;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
mIv = (ImageView) findViewById(R.id.mIv);
Observable.just(R.mipmap.ic_launcher)
.map(new Func1<Integer, Drawable>() {
@Override
public Drawable call(Integer integer) {
return ContextCompat.getDrawable(MainActivity.this,integer);
}
}).subscribe(new Action1<Drawable>() {
@Override
public void call(Drawable drawable) {
mIv.setBackgroundDrawable(drawable);
}
});
}
}
瞬间清爽很多。
现在来说说map和Func1了
FuncX
Action1 非常相似，也是 RxJava 的一个接口，用于包装含有一个参数的方法。 Func1 和 Action 的区别在于， Func1 包装的是有返回值的方法。另外，和 ActionX 一样， FuncX 也有多个，用于不同参数个数的方法。FuncX 和 ActionX 的区别在 FuncX 包装的是有返回值的方法。
FuncX的重要特征： 有返回值
map
先来看一下map的这个在源码中的样子：
public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
return lift(new OperatorMap<T, R>(func));
}
这里传入两个参数，一个T，一个R，map就是传入T类型的输出，但是需要返回R类型的数据，你也可以认为第一个参数需要在map方法return之后转为第二个参数的类型。
map() 方法将参数中的 Integer 对象转换成一个 Drawable 对象后返回，而在经过 map() 方法后，事件的参数类型也由 Integer 转为了 Drawable。这种直接变换对象并返回的，是最常见的也最容易理解的变换。
注意：事件的参数类型也由 Integer 转为了 Drawable
不过 RxJava 的变换远不止这样，它不仅可以针对事件对象，还可以针对整个事件队列，这使得 RxJava 变得非常灵活。
map变换，在RxJava里面简单又常见，借用一下 扔物线同学 的好图
形象生动。
flatMap
现在有这么一个数据，有一个程序员数组，每个程序员掌握的编程语言是不同的。
实体如下：
程序员
public class ProgrammerBean {
public String name;
public List<LanguageDevBean> proLanList; // 掌握的语言
}
语言
public class LanguageDevBean {
public String proLanName;
public LanguageDevBean(String proLanName) {
this.proLanName = proLanName;
}
}
我们现在要求是这样子的，打印出每一程序员的掌握的语言。不用for、while等循环。
我们想，这个时候我们需要变换，这个不是说吧程序员ProgrammerBean这个类型转为LanguageDevBean就行，如果这样就是一对一了。我们需要做的是转化为多个LanguageDevBean。就是一个ProgrammerBean对应多个LanguageDevBean。
（其实你可以一个ProgrammerBean对应成为一个List<LanguageDevBean>,但是我们这里不这么弄，为了演示flatMap）
map()的变换是一对一的转化
flatMap的变换一对多的转化
flatMap() 和 map() 有一个相同点：
它也是把传入的参数转化之后返回另一个对象。和 map() 不同的是， flatMap() 中返回的是个 Observable 对象，并且这个 Observable 对象并不是被直接发送到了 Subscriber 的回调方法中。
原理
flatMap() 的原理是这样的：
- 使用传入的事件对象创建一个 Observable 对象；
- 并不发送这个 Observable, 而是将它激活，于是它开始发送事件（汇集齐统一发送）；
- 每一个创建出来的 Observable 发送的事件，都被汇入同一个 Observable ，而这个 Observable 负责将这些事件统一交给 Subscriber 的回调方法。
（注意：flatMap() 中返回的是个 Observable 对象）
这三个步骤，把事件拆成了两级，通过一组新创建的 Observable 将初始的对象铺平之后通过统一路径分发了下去。而这个铺平就是 flatMap() 所谓的 flat。
上代码：
public class MainActivity extends Activity {
private List<ProgrammerBean> mProgrammerBeanList;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
mProgrammerBeanList = new ArrayList<>();
ProgrammerBean pg1 = new ProgrammerBean();
pg1.name = "风清扬";
ArrayList<LanguageDevBean> programmeLanBeenList1 = new ArrayList<>();
programmeLanBeenList1.add(new LanguageDevBean("C++"));
programmeLanBeenList1.add(new LanguageDevBean("Java"));
programmeLanBeenList1.add(new LanguageDevBean("Js"));
pg1.proLanList = programmeLanBeenList1;
mProgrammerBeanList.add(pg1);
ProgrammerBean pg2 = new ProgrammerBean();
pg2.name = "孙悟空";
ArrayList<LanguageDevBean> programmeLanBeenList2 = new ArrayList<>();
programmeLanBeenList2.add(new LanguageDevBean("Php"));
pg2.proLanList = programmeLanBeenList2;
mProgrammerBeanList.add(pg2);
Subscriber<LanguageDevBean> subscriber = new Subscriber<LanguageDevBean>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(LanguageDevBean programmeLanBean) {
System.out.println("==========:"+programmeLanBean.proLanName);
}
};
Observable.from(mProgrammerBeanList)
.flatMap(new Func1<ProgrammerBean, Observable<LanguageDevBean>>() {
@Override
public Observable<LanguageDevBean> call(ProgrammerBean programmerBean) {
return Observable.from(programmerBean.proLanList);
}
}).subscribe(subscriber);
}
}
打印输出：
==========:C++
==========:Java
==========:Js
==========:Php
“风清扬”和“孙悟空”的编程语言（在当前代码下加起来就是4个）汇集齐后统一发给观察者。
再借 扔物线同学 图片一用
扩展：由于可以在嵌套的 Observable 中添加异步代码， flatMap() 也常用于嵌套的异步操作，例如嵌套的网络请求。（Retrofit + RxJava）
传统的嵌套请求需要使用嵌套的 Callback 来实现。而通过 flatMap() ，可以把嵌套的请求写在一条链中，从而保持程序逻辑的清晰。
本文大致如上，关于lift等就不谈了，大家从扔物线 给 Android 开发者的 RxJava 详解等文章可以得到更多的体验。扔物线和大头鬼等是目前为止的RxJava正宗，本文只是旁门左道，仅以闲记。
本篇完。
