RxJava的学习与实现

RxJava

要在Android中使用RxJava2, 先添加Gradle配置:

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

原理

在开始学习以前，先来介绍点原理性的东西。

网上也有不少介绍RxJava原理的文章，一般这些文章都从观察者模式开始，先讲观察者，被观察者，订阅关系巴拉巴拉一大堆，说实话，当我第一次看到这些文章的时候已经被这些名词给绕晕了，用了很长的时间才理清楚它们之间的关系。多是我太蠢了，境界不够，领会不到那么多高大上的名词.

今天我用两根水管代替观察者和被观察者, 试图用通俗易懂的话把它们的关系解释清楚, 在这里我将从事件流这个角度来讲明RxJava的基本工做原理。

先假设有两根水管：

上面一根水管为事件产生的水管，叫它上游吧，下面一根水管为事件接收的水管叫它下游吧。

两根水管经过必定的方式链接起来，使得上游每产生一个事件，下游就能收到该事件。注意这里和官网的事件图是反过来的, 这里的事件发送的顺序是先1,后2,后3这样的顺序, 事件接收的顺序也是先1,后2,后3的顺序, 我以为这样更符合咱们普通人的思惟, 简单明了.

这里的上游和下游就分别对应着RxJava中的Observable和Observer，它们之间的链接就对应着subscribe()，所以这个关系用RxJava来表示就是：

//建立一个上游 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);

这个运行的结果就是:

12-02 03:37:17.818 4166-4166/zlc.season.rxjava2demo D/TAG: subscribe
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: 1
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: 2
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: 3
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: complete

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

把这段代码连起来写就成了RxJava引觉得傲的链式操做：

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和Disposable.

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

可是，请注意，并不意味着你能够随意乱七八糟发射事件，须要知足必定的规则：

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

注: 关于onComplete和onError惟一而且互斥这一点, 是须要自行在代码中进行控制, 若是你的代码逻辑中违背了这个规则, 并不必定会致使程序崩溃. 好比发送多个onComplete是能够正常运行的, 依然是收到第一个onComplete就再也不接收了, 但如果发送多个onError, 则收到第二个onError事件会致使程序会崩溃.

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

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

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

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

            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "subscribe");
                mDisposable = d;
            }

            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "onNext: " + value);
                i++;
                if (i == 2) {
                    Log.d(TAG, "dispose");
                    mDisposable.dispose();
                    Log.d(TAG, "isDisposed : " + mDisposable.isDisposed());
                }
            }

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

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

运行结果为:

12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: subscribe
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 1
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: onNext: 1
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 2
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: onNext: 2
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: dispose
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: isDisposed : true
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 3
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit complete
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 4

从运行结果咱们看到, 在收到onNext 2这个事件后, 切断了水管, 可是上游仍然发送了3, complete, 4这几个事件, 并且上游并无由于发送了onComplete而中止. 同时能够看到下游的onSubscribe()方法是最早调用的.

Disposable的用处不止这些, 后面讲解到了线程的调度以后, 咱们会发现它的重要性. 随着后续深刻的讲解, 咱们会在更多的地方发现它的身影.

另外, 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.d(TAG, "emit 1");
                emitter.onNext(1);
                Log.d(TAG, "emit 2");
                emitter.onNext(2);
                Log.d(TAG, "emit 3");
                emitter.onNext(3);
                Log.d(TAG, "emit complete");
                emitter.onComplete();
                Log.d(TAG, "emit 4");
                emitter.onNext(4);
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "onNext: " + integer);
            }
        });