Android拾萃 - RxJava最简单的入门（一）

环境搭建

在主项目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);

其他操作符，在后面序列文章里面，结合实际场景进行分析。