Rxjava2学习笔记一：RxJava2基本使用

参考链接：
https://www.jianshu.com/p/464fa025229e
Rxjava2学习笔记二：RxJava2进阶使用-zip操作符
https://www.jianshu.com/p/ef8b620fdc4c
Rxjava2学习笔记三：RxJava2进阶使用-map操作符
https://www.jianshu.com/p/f7efc1aeb6c9

1.Gradle配置

• compile 'io.reactivex.rxjava2:rxjava:2.0.1'
  compile 'io.reactivex.rxjava2:rxandroid:2.0.1'

2.原理

• 先假设2根水管： Alt text
• 上面一根水管为事件产生的水管，叫它上游吧，下面一根水管为事件接收的水管叫它下游吧
• 两根水管通过某种方式连接在一起：使得上游每产生一个事件，下游都能收到该事件
  上游事件产生顺序:1->2->3;下游事件接收顺序:1->2->3

基本使用例1

    注：Rxjava2中的emitter-》发射器,用于被观察者发射事件  
    Disposable-》RxJava1.x中的Subscription,用于解除订阅

    //创建一个上游 Observable：
    Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
            emitter.onNext(1);
            emitter.onNext(2);
            emitter.onNext(3);
            emitter.onComplete();
        }
    });
    //创建一个下游 Observer
    Observer<Integer> observer = new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, "subscribe");
        }

        @Override
        public void onNext(Integer value) {
            Log.d(TAG, "" + value);
        }

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

        @Override
        public void onComplete() {
            Log.d(TAG, "complete");
        }
    };
    //建立连接
    observable.subscribe(observer);

基本使用例2-链式调用

    Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
            emitter.onNext(1);
            emitter.onNext(2);
            emitter.onNext(3);
            emitter.onComplete();
        }
    }).subscribe(new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, "subscribe");
        }

        @Override
        public void onNext(Integer value) {
            Log.d(TAG, "" + value);
        }

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

        @Override
        public void onComplete() {
            Log.d(TAG, "complete");
        }
    });

• 注：ObservableEmitter[发射器]:发出事件
  ->调用emitter的onNext(T value)、onComplete()和onError(Throwable error)
  ->发出next事件、complete事件和error事件

Emitter-发送规则

• 1.上游可以发送无限个onNext, 下游也可以接收无限个onNext.
• 2.当上游发送了一个onComplete后, 上游onComplete之后的事件将会继续发送,
  而下游收到onComplete事件之后将不再继续接收事件.
• 3.当上游发送了一个onError后, 上游onError之后的事件将继续发送, 而下游收到onError事件之后将不再继续接收事件.
• 4.上游可以不发送onComplete或onError.
• 5.最为关键的是onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError, 也不能先发一个onComplete, 然后再发一个onError, 反之亦然

发送规则示意图
1.发送onNext()事件

Alt text
2.发送onComplete()事件 Alt text 3.发送发送onError事件事件 Alt text

Disposable-订阅

• 1.相当于RxJava1.x中的Subscription,用于解除订阅

• 2.解除订阅:disposable.dispose();

• 3.多个Disposable时取消订阅,RxJava中已经内置了一个容器CompositeDisposable,每当我们得到一个Disposable时就调用CompositeDisposable.add()将它添加到容器中, 在退出的时候, 调用CompositeDisposable.clear() 即可切断所有的水管.

• 4.eg:

  Disposable disposable = observable.subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                  //这里接收数据项
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
              //这里接收onError
            }
        }, new Action() {
            @Override
            public void run() throws Exception {
              //这里接收onComplete。
            }
        });
    disposable.dispose();//解除订阅
  

Consumer-消费者

• 1.用于接收单个值，其他的如：BiConsumer则是接收两个值，Function用于变换对象，Predicate用于判断用法如上的例子
• 2.替代了RxJava1.x中的Action/Func接口

3.Rxjava线程调度

• 1 subscribeOn(Schedulers.io())//上游发送事件的线程,第一次有效(如网络请求可在IO或子线程发送事件)
  observeOn(AndroidSchedulers.mainThread())//下游接收事件的线程(主线程接收返回信息后更新UI)
  observeOn(Schedulers.io())//线程切换；可多次切换每调用一次observeOn() , 下游的线程就会切换一次.