阅读 3171

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()方法将上下游的连接断开了,断开后上游后续生产的饼干事件,下游就接收不到。

运行的结果:

image.png

上图中,也验证了我们的猜想,当使用dispose断开上下游连接后,下游就无法再继续接收事件了。


这一讲就先介绍这么多,这样的方式理解Observable和Observer以及订阅动作subscribe是不是容易多了,希望对你有所帮助,下一讲使用RxJava来切换变化饼干事件处理的线程(主线程、子线程)。

关注下面的标签,发现更多相似文章
评论