转载

RxJava 2.x入门新姿势一

经过几年的发展，响应式编程已经是很流行了，在Android开发中的应用也非常的广泛，身为Android开发者，则是必须掌握的技术。

正文

网上已经有很多很多RxJava相关的文章，视频等等教程，但是说实话对于入门，或者新手来说，确实不好理解，上来就是各种，观察者、被观察者、订阅、发布等等概念，一遍看下来直接就晕了，就感觉RxJava很难，难理解，用的时候也只是依葫芦画瓢，晕乎乎的用着，然后就没有然后了。

这里我都不说那些概念，因为讲概念太抽象，难记住，更难理解。我们用另外一个视角来学习。因为RxJava 1.x的版本官方已经停止更新了维护了，没有学习过也没有关系，RxJava 2.x是全新的，直接学习使用就好了。

首先假设我们在工厂里上班，工厂都会有流水线，产品经过流水线生产后来订单了销售出去。

这里假设工厂生产的是一种六边形的“Jerry帅气饼干”，上游是生产车间流水线的事件流，下游是订单产品的销售消费事件流。中间连接上下游关系的暂且叫做“Jerry帅气饼干生产消费订单管理系统”（不要脸，名字写这么长），为了下文方便抒写且用“生产订单管理系统”（PCMS）。以上图上下游对应的就是Obsersvable被观察者也是发布者，下游对应Observer观察者也是订阅者。使用RxJava代码表示上图就是：

public void test1() {

        // 上游生成产品流水线
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                Log.d(TAG, "test1 ====== Observable: ------ onNext: Jerry");
                emitter.onNext("Jerry");
                Log.d(TAG, "test1 ====== Observable: ------ onNext: 就是");
                emitter.onNext("就是");
                Log.d(TAG, "test1 ====== Observable: ------ onNext: 帅");
                emitter.onNext("帅");
                Log.d(TAG, "test1 ====== Observable: ------ onNext: ！！！");
                emitter.onNext("！！！");
                Log.d(TAG, "test1 ====== Observable: ------ onComplete");
                emitter.onComplete();
                Log.d(TAG, "test1 ====== Observable: ------ onNext: Jerry帅炸天！！！");
                emitter.onNext("Jerry帅炸天！！！");
            }
        });

        // 下游订单产品销售
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "test1 ====== Observer: onSubscribe");
            }

            @Override
            public void onNext(String value) {
                Log.d(TAG, "test1 ====== Observer: onNext: " + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "test1 ====== Observer: onError: " + e.getMessage());
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "test1 ====== Observer: onComplete");
            }
        };

        // 连接上下游的订单管理系统
        observable.subscribe(observer);
    }
复制代码

上述代码，上游生产车间流水线就是Observable，下游订单销售就是Observer，中间通过“生产订单管理系统”subscribe来将上下游连接起来。

运行后输出结果是：

从输出结果来看，当上游Observable发出一个生产的饼干产品事件，下游订单销售的Observer就销售一个饼干产品事件，而且当上游调用了onComplete方法后，上游的生产事件还是生产饼干事件（继续生产了“Jerry帅炸天”饼干事件），但是下游的订单销售却没有消费掉。也就是事件产生方调用onComplete方法后，之后的事件还会继续发送，但是事件接收方就不会接收了。

我们来看看Observable的subscribe方法的参数：ObservableEmitter，Emitter顾名思义是发射器的意思，ObservableEmitter接口继承自Emitter接口：

public interface Emitter<T> {

    /**
     * Signal a normal value.
     * @param value the value to signal, not null
     */
    void onNext(T value);

    /**
     * Signal a Throwable exception.
     * @param error the Throwable to signal, not null
     */
    void onError(Throwable error);

    /**
     * Signal a completion.
     */
    void onComplete();
}
复制代码

接口定义很简单，就三个方法，onNext我们上门已经用过了，是用来发射发送事件的，onComplete是用来表示事件发送完了，后面如果有新的事件发送，下游接收者可以不用处理，onError方法看注释说是发送一个异常事件给下游接收者。到底是不是这样，我们来试试就晓得了。

public void test2() {

        // 上游生成产品流水线
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                Log.d(TAG, "test2 ====== Observable: ------ onNext: Jerry");
                emitter.onNext("Jerry");
                Log.d(TAG, "test2 ====== Observable: ------ onNext: 就是");
                emitter.onNext("就是");
                Log.d(TAG, "test2 ====== Observable: ------ onNext: 帅");
                emitter.onNext("帅");
                Log.d(TAG, "test2 ====== Observable: ------ onNext: ！！！");
                emitter.onNext("！！！");
                Log.d(TAG, "test2 ====== Observable: ------ onError");
                emitter.onError(new IllegalStateException("Jerry饼干烤焦了，卖出去会被打！"));
                Log.d(TAG, "test2 ====== Observable: ------ onNext: Jerry帅炸天！！！");
                emitter.onNext("Jerry帅炸天！！！");
            }
        });

        // 下游订单产品销售
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "test2 ====== Observer: onSubscribe");
            }

            @Override
            public void onNext(String value) {
                Log.d(TAG, "test2 ====== Observer: onNext: " + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "test2 ====== Observer: onError: " + e.getMessage());
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "test2 ====== Observer: onComplete");
            }
        };

        // 连接上下游的订单管理系统
        observable.subscribe(observer);
    }
复制代码

在上游生产饼干的时候就生产了一个“Jerry饼干烤焦了，卖出去会被打！”的错误饼干事件，下游订单销售的onError出错状态会消费这个事件。而上游在出错事件后发送的“Jerry帅炸天！！！”饼干事件，同样也只是把事件发送了处理，下游订单销售并没有接收处理这个事件。

运行后输出结果：

细心的小伙伴应该会发现，每次执行的时候都会先调用下游的onSubscribe方法，这个方法里有个参数Disposable（用完即可丢弃）意思可以理解成，将上下游的连接切断，让上游的生产的饼干不打包放入下游订单销售环节，实际开发中是有这种需求的，当发送事件出问题的时候就需要断开事件接收处理。不像最近的疫苗事件，一些不要脸的生物疫苗公司把生产不合格的疫苗上市销售，伤天害理，谋财害命。下面举个例子：

public void test3() {

        // 上游生成产品流水线
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                Log.d(TAG, "test3 ====== Observable: ------ onNext: Jerry");
                emitter.onNext("Jerry");
                Log.d(TAG, "test3 ====== Observable: ------ onNext: 就是");
                emitter.onNext("就是");
                Log.d(TAG, "test3 ====== Observable: ------ onNext: 帅");
                emitter.onNext("帅");
                Log.d(TAG, "test3 ====== Observable: ------ onNext: ！！！");
                emitter.onNext("！！！");
                Log.d(TAG, "test3 ====== Observable: ------ onComplete");
                emitter.onComplete();
                Log.d(TAG, "test3 ====== Observable: ------ onNext: Jerry帅炸天！！！");
                emitter.onNext("Jerry帅炸天！！！");
            }
        });

        // 下游订单产品销售
        Observer<String> observer = new Observer<String>() {

            private Disposable mDisposable;
            private int i;

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

            @Override
            public void onNext(String value) {
                Log.d(TAG, "test3 ====== Observer: onNext: " + value);
                i++;
                // 第一个事件接收后，就断开上下游连接
                if (i == 1) {
                    Log.d(TAG, "test3 ====== Observer: start disposable");
                    mDisposable.dispose();
                    Log.d(TAG, "test3 ====== Observer: isDisposable: " + mDisposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "test3 ====== Observer: onError: " + e.getMessage());
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "test3 ====== Observer: onComplete");
            }
        };

        // 连接上下游的订单管理系统
        observable.subscribe(observer);
    }
复制代码

这里我们在下游订单销售的onNext方法中，当接收完第一个饼干事件后，就使用mDisposable.dispose()方法将上下游的连接断开了，断开后上游后续生产的饼干事件，下游就接收不到。

运行的结果：