RxJava的简单使用

导入RxJava的相关依赖
创建观察者与被观察者
链式调用
next、error、complete
subscribe的重载方法
指定观察者与被观察者所在的线程以及线程切换
结合Retrofit，实现一个简单的网络请求

一、引入依赖：

implementation "io.reactivex.rxjava2:rxjava:2.2.3"
implementation 'io.reactivex.rxjava2:rxandroid:2.1.0'

二、观察者与被观察者
RxJava中的Observable和Observer通过SubScribe建立连接：

 //创建一个上游 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.onError(new Throwable());
                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);

运行结果：

11-21 16:49:58.443 21398-21398/com.example.qidong.testrxjava D/MainActivity: subscribe
11-21 16:49:58.443 21398-21398/com.example.qidong.testrxjava D/MainActivity: 1
11-21 16:49:58.444 21398-21398/com.example.qidong.testrxjava D/MainActivity: 2
11-21 16:49:58.444 21398-21398/com.example.qidong.testrxjava D/MainActivity: 3
11-21 16:49:58.444 21398-21398/com.example.qidong.testrxjava D/MainActivity: error

注意: 只有当上游和下游建立连接之后, 上游才会开始发送事件. 也就是调用了subscribe() 方法之后才开始发送事件.

三、RxJava的链式调用
使用RxJava的链式操作如下：

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

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {

                emitter.onNext(20);
                emitter.onNext(30);
                emitter.onError(new Throwable());
                emitter.onNext(40);
                emitter.onComplete();

            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

                Log.e(TAG,"OnSubscribe");

            }

            @Override
            public void onNext(Integer integer) {
                Log.e(TAG,"onNext:"+integer);

            }

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

            }

            @Override
            public void onComplete() {
                Log.e(TAG,"OnComplete");

            }
        });

四、next、error、complete事件
从上面的代码中可以看到ObservableEmitter和Disposable。
ObservableEmitter：意思为发射器，可以通过调用onNext()、onError(Throwable)、onComplete()来发射next、error、complete这三种事件。
这其中需要注意的地方有：
上游可以发送无限个onNext, 下游也可以接收无限个onNext.
当上游发送了一个onComplete后, 上游onComplete之后的事件将会继续发送, 而下游收到onComplete事件之后将不再继续接收事件.
当上游发送了一个onError后, 上游onError之后的事件将继续发送, 而下游收到onError事件之后将不再继续接收事件.
上游可以不发送onComplete或onError.
最为关键的是onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError, 也不能先发一个onComplete, 然后再发一个onError, 反之亦然

Disposable, 这个单词的字面意思是一次性用品,用完即可丢弃的. 那么在RxJava中怎么去理解它呢, 对应于上面的水管的例子, 我们可以把它理解成两根管道之间的一个机关, 当调用它的dispose()方法时, 它就会将两根管道切断, 从而导致下游收不到事件.

注意: 调用dispose()并不会导致上游不再继续发送事件, 上游会继续发送剩余的事件.

来看个例子, 我们让上游依次发送1,2,3,complete,4,在下游收到第二个事件之后, 切断水管, 看看运行结果:

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

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {

                Log.e(TAG, "emit 1");
                emitter.onNext(1);
                Log.e(TAG, "emit 2");
                emitter.onNext(2);
                Log.e(TAG, "emit 3");
                emitter.onNext(3);
                Log.e(TAG, "onComplete");
                emitter.onComplete();
                Log.e(TAG, "emit 4");
                emitter.onNext(4);

            }
        }).subscribe(new Observer<Integer>() {

            private Disposable mDispoable;
            private int i;

            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "onSubscribe: ");
                this.mDispoable=d;


            }

            @Override
            public void onNext(Integer integer) {

                Log.e(TAG, "onNext: "+integer);


                i++;

                if(i==2)
                {
                    Log.e(TAG, "dispose");
                    mDispoable.dispose();
                    Log.e(TAG, "isDisposed:"+mDispoable.isDisposed());
                }

            }

            @Override
            public void onError(Throwable e) {

                Log.e(TAG, "onError");

            }

            @Override
            public void onComplete() {

                Log.e(TAG, "onComplete");

            }
        });

运行结果：

11-21 17:11:50.338 23074-23074/com.example.qidong.testrxjava E/MainActivity: onSubscribe: 
11-21 17:11:50.338 23074-23074/com.example.qidong.testrxjava E/MainActivity: emit 1
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: onNext: 1
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: emit 2
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: onNext: 2
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: dispose
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: isDisposed:true
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: emit 3
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: onComplete
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: emit 4

可以看到，切断观察者和被观察者之间的关联之后，事件3、complete、4事件依然被发送了。

五、subscribe的重载方法
SubScribe有多个重载的方法：

   public final Disposable subscribe() {}
    public final Disposable subscribe(Consumer<? super T> onNext) {}
    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {} 
    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}
    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}
    public final void subscribe(Observer<? super T> observer) {}

最后一个带有Observer参数的我们已经使用过了,这里对其他几个方法进行说明.
不带任何参数的subscribe() 表示下游不关心任何事件,你上游尽管发你的数据去吧, 老子可不管你发什么.
带有一个Consumer参数的方法表示下游只关心onNext事件, 其他的事件我假装没看见, 因此我们如果只需要onNext事件可以这么写:

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

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {

                Log.e(TAG, "onNext:10");
                emitter.onNext(10);
                Log.e(TAG, "onNext:20");
                emitter.onNext(20);
                Log.e(TAG, "OnComplete");
                emitter.onComplete();
                Log.e(TAG, "onNext:30");
                emitter.onNext(30);

            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {

                Log.e(TAG, "accept:"+integer);

            }
        });
    }

运行结果：

11-21 17:29:00.671 23664-23664/? E/MainActivity: onNext:10
11-21 17:29:00.672 23664-23664/? E/MainActivity: accept:10
11-21 17:29:00.672 23664-23664/? E/MainActivity: onNext:20
11-21 17:29:00.672 23664-23664/? E/MainActivity: accept:20
11-21 17:29:00.672 23664-23664/? E/MainActivity: OnComplete
11-21 17:29:00.672 23664-23664/? E/MainActivity: onNext:30

六、指定观察者与被观察者所在的线程以及线程切换
正常情况下，观察者和被观察者是工作在一个线程中的，也就是被观察者在哪个线程发射事件，观察者就在哪个线程接受事件：

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

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {

                Log.e(TAG, "onNext:10");
                emitter.onNext(10);
                Log.e(TAG, "Observable Thread Name:"+Thread.currentThread().getName());

            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {

                Log.e(TAG, "accept:"+integer);
                Log.e(TAG, "Observer Thread Name:"+Thread.currentThread().getName());

            }
        });

运行结果：

11-21 17:43:25.366 25090-25090/? E/MainActivity: onNext:10
11-21 17:43:25.366 25090-25090/? E/MainActivity: accept:10
11-21 17:43:25.366 25090-25090/? E/MainActivity: Observer Thread Name:main
11-21 17:43:25.366 25090-25090/? E/MainActivity: Observable Thread Name:main

一般而言，我们会将耗时操作放在子线程中去做，然后在主线程中根据运行的结果去更新UI。
RxJava中提供了线程调度器实现这样的需求：我们在子线程中发射事件，在主线程中接受事件:

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

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {

                Log.e(TAG, "onNext:10");
                emitter.onNext(10);
                Log.e(TAG, "Observable Thread Name:"+Thread.currentThread().getName());

            }
        }).subscribeOn(Schedulers.newThread())
          .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {

                Log.e(TAG, "accept:"+integer);
                Log.e(TAG, "Observer Thread Name:"+Thread.currentThread().getName());

            }
        });

此时再看一下运行结果：

11-21 17:50:37.408 25766-25934/com.example.qidong.testrxjava E/MainActivity: onNext:10
11-21 17:50:37.409 25766-25934/com.example.qidong.testrxjava E/MainActivity: Observable Thread Name:RxNewThreadScheduler-1
11-21 17:50:37.453 25766-25766/com.example.qidong.testrxjava E/MainActivity: accept:10
11-21 17:50:37.453 25766-25766/com.example.qidong.testrxjava E/MainActivity: Observer Thread Name:main

简单的来说, subscribeOn() 指定的是上游发送事件的线程, observeOn() 指定的是下游接收事件的线程.
多次指定被观察者的线程只有第一次指定的有效, 也就是说多次调用subscribeOn() 只有第一次的有效, 其余的会被忽略.
多次指定观察者的线程是可以的, 也就是说每调用一次observeOn() , 下游的线程就会切换一次.

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

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {

                Log.e(TAG, "onNext:10");
                emitter.onNext(10);
                Log.e(TAG, "Observable Thread Name:"+Thread.currentThread().getName());

            }
        }).subscribeOn(Schedulers.newThread())
                .subscribeOn(Schedulers.io())
          .observeOn(AndroidSchedulers.mainThread())
                .observeOn(Schedulers.io())
                .subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {

                Log.e(TAG, "accept:"+integer);
                Log.e(TAG, "Observer Thread Name:"+Thread.currentThread().getName());

            }
        });

再来看下运行结果：

11-21 17:54:36.940 26399-26522/com.example.qidong.testrxjava E/MainActivity: onNext:10
11-21 17:54:36.941 26399-26522/com.example.qidong.testrxjava E/MainActivity: Observable Thread Name:RxNewThreadScheduler-1
11-21 17:54:36.973 26399-26524/com.example.qidong.testrxjava E/MainActivity: accept:10
11-21 17:54:36.974 26399-26524/com.example.qidong.testrxjava E/MainActivity: Observer Thread Name:RxCachedThreadScheduler-2

可以看出观察者所在的线程被修改成功。

为了看出观察者所在线程的切换过程，我们可以添加Log去查看：

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

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.e(TAG, "emit 10");
                emitter.onNext(10);
                Log.e(TAG, "emit 20");
                emitter.onNext(20);

            }
        }).subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "After Change to MainThread,thread name:"+Thread.currentThread().getName());

                    }
                }).observeOn(Schedulers.io()).doOnNext(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "After Change to IO,thread name:"+Thread.currentThread().getName());
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "accept: "+integer);
            }
        });

运行结果：

11-21 18:08:37.913 28956-29018/com.example.qidong.testrxjava E/MainActivity: emit 10
11-21 18:08:37.914 28956-29018/com.example.qidong.testrxjava E/MainActivity: emit 20
11-21 18:08:37.945 28956-28956/com.example.qidong.testrxjava E/MainActivity: After Change to MainThread,thread name:main
11-21 18:08:37.946 28956-28956/com.example.qidong.testrxjava E/MainActivity: After Change to MainThread,thread name:main
11-21 18:08:37.946 28956-29020/com.example.qidong.testrxjava E/MainActivity: After Change to IO,thread name:RxCachedThreadScheduler-1
11-21 18:08:37.946 28956-29020/com.example.qidong.testrxjava E/MainActivity: accept: 10
11-21 18:08:37.947 28956-29020/com.example.qidong.testrxjava E/MainActivity: After Change to IO,thread name:RxCachedThreadScheduler-1
11-21 18:08:37.947 28956-29020/com.example.qidong.testrxjava E/MainActivity: accept: 20

在RxJava中, 已经内置了很多线程选项供我们选择, 例如有
Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作
Schedulers.newThread() 代表一个常规的新线程
AndroidSchedulers.mainThread() 代表Android的主线程
这些内置的Scheduler已经足够满足我们开发的需求, 因此我们应该使用内置的这些选项, 在RxJava内部使用的是线程池来维护这些线程, 所有效率也比较高.