Rxjava 2学习笔记（一）

Rxjava 在这些年在android开发中非常的火爆，它和Retrofit 的结合堪称完美，他们可以把我们从以前我们进行网络请求中各种线程切换，各种接口回掉中解放出来了。可以让我们的逻辑变的非常清晰，便于代码维护。我们公司的项目目前使用的Rxjava 版本还是1.0版本的，准备打算升级到使用Rx2.0。所以打算学习一下Rxjava 2

Observable.create(new ObservableOnSubscribe<String>() {

            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("hello world");
                emitter.onComplete();
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "onSubscribe");
            }

            @Override
            public void onNext(String s) {
                Log.e(TAG, "onNext "+s);
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError");
            }

            @Override
            public void onComplete() {
                Log.e(TAG, "onComplete");
            }
        });
        
10-13 16:24:39.176 11578-11578/? E/huangjie: onSubscribe
10-13 16:24:39.176 11578-11578/? E/huangjie: onNext hello world
10-13 16:24:39.176 11578-11578/? E/huangjie: onComplete

上面代码中我们看到了两个陌生的对象，ObservableEmitter和Disposable.

ObservableEmitter：

Emitter是发射器的意思，那就很好猜了，这个就是用来发出事件的，它可以发出三种类型的事件，通过调用emitter的onNext(T value)、onComplete()和onError(Throwable error)就可以分别发出next事件、complete事件和error事件

Disposable：

这个单词的字面意思是一次性用品,用完即可丢弃的. 那么在RxJava中怎么去理解它呢, 可以想象成一条河流当调用它的dispose()方法时, 它就会放下栅栏, 从而导致下游收不到事件.

在Rxjava 2 中，Observable不再支持订阅Subscriber了，而是需要使用Observer作为观察者。但是有一点是没有变化的，不管是在Rxjava 1 还是在Rxjava 2中，被观察者，观察者，订阅三者缺一不可。只有在使用了subscribe()方法过后（就是被观察者订阅观察者），被观察者才会发送数据。

在Rxjava 2中有五种观察者模式

1.png

类型	描述
Observale	能够发射0或者n 个数据，并以成功或者错误事件终止
Flowable	能够发射0或n个数据，并以成功或者错误事件终止，支持背压，可以控制数据源发射的速度
Single	只能发射单个数据或错误事件
Completable	从来不发射数据，只处理onComplete 和onError事件，可以看成Rxjava的Runnable
Maybe	能够发射0或者n个数据，要么成功，要么失败。有点类似于Optional

do操作符

do操作符可以给Observable的生命周期的各个阶段加上一系列的监听，当Observable执行到各个阶段的时，这些回掉就会触发。

Observable.just("hello")
                .doOnNext(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.e(TAG, "doOnNext" + s);
                    }
                })
                .doAfterNext(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.e(TAG, "doAfterNext" + s);
                    }
                })
                .doOnComplete(new Action() {
                    @Override
                    public void run() throws Exception {
                        Log.e(TAG, "doOnComplete");
                    }
                })
                .doOnSubscribe(new Consumer<Disposable>() {
                    @Override
                    public void accept(Disposable disposable) throws Exception {
                        Log.e(TAG, "doOnSubscribe");
                    }
                })
                .doAfterTerminate(new Action() {
                    @Override
                    public void run() throws Exception {
                        Log.e(TAG, "doAfterTerminate");
                    }
                })
                .doFinally(new Action() {
                    @Override
                    public void run() throws Exception {
                        Log.e(TAG, "doFinally");

                    }
                })
                .doOnEach(new Consumer<Notification<String>>() {
                    @Override
                    public void accept(Notification<String> stringNotification) throws Exception {
                        Log.e(TAG, "doOnEach" + (stringNotification.isOnNext() ? "onNext" : stringNotification.isOnComplete() ? "onComplete" : "onError"));

                    }
                })
                .doOnLifecycle(new Consumer<Disposable>() {
                    @Override
                    public void accept(Disposable disposable) throws Exception {
                        Log.e(TAG, "doOnLifecycle");
                    }
                }, new Action() {
                    @Override
                    public void run() throws Exception {
                        Log.e(TAG, "doOnLifecycle run");
                    }
                })
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {
                        Log.e(TAG, " onNext" + s);

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, " onError");
                    }

                    @Override
                    public void onComplete() {

                    }
                });
        
  E/huangjie: doOnSubscribe
 E/huangjie: doOnLifecycle
 E/huangjie: doOnNext   hello
 E/huangjie: doOnEach  onNext
 E/huangjie:  onNext  hello
 E/huangjie: doAfterNext  hello
 E/huangjie: doOnComplete
 E/huangjie: doOnEach  onComplete
 E/huangjie: doFinally
 E/huangjie: doAfterTerminate

根据打印的日志我们看出Rxjava 内部数据的流向

操作符	用途
doOnSubscribe	一旦观察者订阅了Observable,它就会调用
doOnLifecycle	可以在观察者订阅之后，设置是否取消订阅
doOnNext	它产生的Observable 每发射一项就会调用它一次，它的Consumer接受发射的数据项，一般用于在subscrible之前的数据进行处理
doOnEach	它产生的Observable每发射一项数据就会调用它一次，不仅包括onNext,还包括onError和onCompleted
doAfterNext	在onNext之后执行，而doOnNext()是在onNext之前执行
doOnComplete	当它产生的Observable在正常终止调用onComplete时会被调用
doFinally	在当它产生的Observable终止之后会被调用，无论是正常终止，还是异常终止，doFinally 优于doAfterTerminate的调用
doAfterTerminate	注册一个Action,当Observable调用onComplete或onError时触发

上面的代码中我们发现了Consumer 和Action,我可以认为其实Consumer 就是Rxjava 1.x 的Action1, Consumer是单一参数，Action是无参数类型。