环境搭建
在主项目APP gradle的dependencies {}里添加以下依赖，最新的引用请参考官网RxAndroid
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile 'io.reactivex.rxjava2:rxjava:2.1.3'
趣解rxjava
RxJava也是基于观察者模式来组建自己的程序逻辑的，就是构建被观察者（Observable），观察者（Observer/Subscriber），然后建立二者的订阅关系（就像那根电线，连接起台灯和开关）实现观察，在事件传递过程中还可以对事件做各种处理。
able英文解释为可以被，可以如何的
Observable： 被观察者（不支持背压）
Flowable： Rxjava2才有的（支持背压被观察者）
er表示的人，观察者
Observer：观察者的接口
Subscriber：实现观察者接口的抽象类，并封装了一些方法，更方便使用，如onNext可以执行多个，onCompleted之后会解除监听（发送者的onComplete之后的事件依旧会继续发送,但是接收者接收到onComplete之后就停止接收事件了.）onerro（发送者onError之后的事件依旧会继续发送,但是接收者当接收到onError之后就会停止接收事件了.）
实质上，在 RxJava 的 subscribe 过程中，Observer 也总是会先被转换成一个 Subscriber 再使用。所以如果你只想使用基本功能，选择 Observer 和 Subscriber 是完全一样的。它们的区别对于使用者来说主要有两点：
onStart(): 这是 Subscriber 增加的方法。它会在 subscribe 刚开始，而事件还未发送之前被调用，可以用于做一些准备工作，例如数据的清零或重置。这是一个可选方法，默认情况下它的实现为空。
需要注意的是，如果对准备工作的线程有要求（例如弹出一个显示进度的对话框，这必须在主线程执行）， onStart() 就不适用了，因为它总是在 subscribe 所发生的线程被调用，而不能指定线程。
要在指定的线程来做准备工作，可以使用 doOnSubscribe() 方法。
Subject
unsubscribe(): 这是 Subscriber 所实现的另一个接口 Subscription 的方法，用于取消订阅。在这个方法被调用后，Subscriber 将不再接收事件。一般在这个方法调用前，可以使用 isUnsubscribed() 先判断一下状态。 unsubscribe() 这个方法很重要，因为在 subscribe() 之后， Observable 会持有 Subscriber 的引用，这个引用如果不能及时被释放，将有内存泄露的风险。所以最好保持一个原则：要在不再使用的时候尽快在合适的地方（例如 onPause() onStop() 等方法中）调用 unsubscribe() 来解除引用关系，以避免内存泄露的发生。
onError事件和onComplete事件是互斥的,但是这并不代表你配置了多个onError和onComplete一定会崩溃,多个onComplete是可以正常运行的,但是只会接收到第一个,之后的就不会再接收到了,多个onError时,只会接收到第一个,第二个会直接造成程序崩溃.
观察者和被观察者怎么建立关系呢？
subscribe： 订阅关系开始，实际就是观察者(subscriber)对象把自己传递给被观察者(observable)内部的onSubscribe。
我们知道，点击事件的监听函数都是On开头的，同样的rxjava的观察监听叫OnSubscribe：被观察者执行方法回调，调用call(subscriber)来通知被观察者发送消息给这个subscriber。
创建被观察者
RxJava 使用 create() 方法来创建一个 Observable。 这里传入了一个 OnSubscribe 对象作为参数。OnSubscribe 会被存储在返回的 Observable 对象中，它的作用相当于一个计划表，当 Observable 被订阅的时候，OnSubscribe 的 call() 方法会自动被调用，事件序列就会依照设定依次触发
//创建一个被观察者（开关）
Observable switcher=Observable.create(new Observable.OnSubscribe<String>(){
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("On");
subscriber.onNext("Off");
subscriber.onNext("On");
subscriber.onNext("On");
subscriber.onCompleted();
}
});
创建观察者
观察者是观察和处理事件传递的终点，有onNext onCompleted onError三个回调方法。
Subscriber light = new Subscriber<String>() {
@Override
public void onCompleted() {
Log.d("DDDDDD","结束观察...\n");
}
@Override
public void onError(Throwable e) {
//出现错误会调用这个方法
}
@Override
public void onNext(String s) {
//处理事件
Log.d("DDDDD","handle this---"+s)；
}
}
订阅
switcher.subscribe(light);
一旦观察关系建立，被观察者的事件就会发射出来，观察者最终收到事件，最后解除监听。
来自网络的图（侵删）
RxJava1 和 RxJava2
观察流程
运行线程控制
默认情况下,发送者和接收者都运行在主线程,但是这显然是不符合实际需求的,我们在日常使用中,通常用的最多的就是在子线程进行各种耗时操作,然后发送到主线程进行,这个时候我们就是需要进行进程切换。
在不指定线程的情况下，RxJava遵循的是线程不变的原则；当需要线程切换时，可以使用Scheduler（线程调度器）。
subscribeOn(),只有在第一次调用的时候生效,之后不管调用多少次,只会以第一次为准，也就是说对所有的操作符生效，是一个群杀技能～～（但是杀伤力比observeOn弱，就很好的理解了下面例子下的解释）
observeOn(),可以被调用多次,每次调用都会更改线程.一般放在just 或者create创建之后，map和flatmap等操作符之前。每个observeOn都可以控制操作符中的操作,作用范围是在下一个observeOn出现前。
Observable.fromArray(...)
.map(1)
.flatMap(2)
.subscribeOn(Schedulers.io())
1、2都是在io线程中执行, 除非在程序中某处执行了observeOn
observeable.just(T...)
.observeOn(Schedulers.io())
.map(1)
.flatMap(2);
.observeOn(Schedulers.newThread())
.map(3)
.subscribe(4)
1、2都是在io线程，3、4在新的线程
常用的Scheduler
| 调度器类型 | 作用范围 |
|---|---|
| AndroidSchedulers.mainThread() | AndroidUI线程 |
| Schedulers.io() | IO线程(读写文件、数据库和网络信息交互等)都可以放在IO线程里来执行,它的模式和newThread()差不多,但效率比newThread()高 |
| Schedulers.newThread() | 开启一个新线程 |
| Schedulers.single() | 单例线程,可以把两段程序放在同一个线程里顺序执行 |
| 调度器类型 | 效果 |
|---|---|
| Schedulers.computation( ) | 用于计算任务，如事件循环或和回调处理，不要用于IO操作(IO操作请使用Schedulers.io())；默认线程数等于处理器的数量 |
| Schedulers.from(executor) | 使用指定的Executor作为调度器 |
| Schedulers.immediate( ) | 在当前线程立即开始执行任务 |
| Schedulers.io( ) | 用于IO密集型任务，如异步阻塞IO操作，这个调度器的线程池会根据需要增长；对于普通的计算任务，请使用Schedulers.computation()；Schedulers.io( )默认是一个CachedThreadScheduler，很像一个有线程缓存的新线程调度器 |
| Schedulers.newThread( ) | 为每个任务创建一个新线程 |
| Schedulers.trampoline( ) | 当其它排队的任务完成后，在当前线程排队开始执行 |
操作符
前面switcher.subscribe(light);是被观察者 .subscribe 观察者，而我们的操作符是创建者模式的链式调用，返回的对象也是被观察者，所以用操作符链式调用代替前面的写法，看下面代码。
//这就是RxJava的流式API调用
Observable.just("On","Off","On","On")
//在传递过程中对事件进行过滤操作
.filter(new Func1<String, Boolean>() {
@Override
public Boolean call(String s) {
return s！=null;
}
})
.subscribe(light);
其他操作符，在后面序列文章里面，结合实际场景进行分析。
网友评论