RxJavaAndroid RxjavaRX实战

RxJava2 实战知识梳理(12) - 实战讲解 publis

2017-09-04  本文已影响938人  泽毛

RxJava2 实战系列文章

RxJava2 实战知识梳理(1) - 后台执行耗时操作，实时通知 UI 更新
RxJava2 实战知识梳理(2) - 计算一段时间内数据的平均值
RxJava2 实战知识梳理(3) - 优化搜索联想功能
RxJava2 实战知识梳理(4) - 结合 Retrofit 请求新闻资讯
RxJava2 实战知识梳理(5) - 简单及进阶的轮询操作
RxJava2 实战知识梳理(6) - 基于错误类型的重试请求
RxJava2 实战知识梳理(7) - 基于 combineLatest 实现的输入表单验证
RxJava2 实战知识梳理(8) - 使用 publish + merge 优化先加载缓存，再读取网络数据的请求过程
RxJava2 实战知识梳理(9) - 使用 timer/interval/delay 实现任务调度
RxJava2 实战知识梳理(10) - 屏幕旋转导致 Activity 重建时恢复任务
RxJava2 实战知识梳理(11) - 检测网络状态并自动重试请求
RxJava2 实战知识梳理(12) - 实战讲解 publish & replay & share & refCount & autoConnect
RxJava2 实战知识梳理(13) - 如何使得错误发生时不自动停止订阅关系
RxJava2 实战知识梳理(14) - 在 token 过期时，刷新过期 token 并重新发起请求
RxJava2 实战知识梳理(15) - 实现一个简单的 MVP + RxJava + Retrofit 应用

一、前言

今天，我们来整理以下几个大家容易弄混的概念，并用实际例子来演示，可以从 RxSample 的第十二章中获取：

对于以上这些概念，可以用一幅图来概括：


从图中可以看出，这里面可以供使用者订阅的Observable可以分为四类，下面我们将逐一介绍这几种Observable的特点：

二、Cold Observable

Cold Observable就是我们通过Observable.create、Observable.interval等创建型操作符生成的Observable，它具有以下几个特点：

下面，我们演示一个例子，首先我们创建一个Cold Observable：

    //直接订阅Cold Observable。
    private void createColdSource() {
        mConvertObservable = getSource();
    }

    private Observable<Integer> getSource() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> observableEmitter) throws Exception {
                try {
                    int i = 0;
                    while (true) {
                        Log.d(TAG, "源被订阅者发射数据=" + i + ",发送线程ID=" + Thread.currentThread().getId());
                        mSourceOut.add(i);
                        observableEmitter.onNext(i++);
                        updateMessage();
                        Thread.sleep(1000);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).subscribeOn(Schedulers.io());
    }

在创建两个订阅者，它们可以随时订阅到Cold Observable或者取消对它的订阅：

    private void startSubscribe1() {
        if (mConvertObservable != null && mDisposable1 == null) {
            mDisposable1 = mConvertObservable.subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    Log.d(TAG, "订阅者1收到数据=" + integer + ",接收线程ID=" + Thread.currentThread().getId());
                    mSubscribe1In.add(integer);
                    updateMessage();
                }
            });
        }
    }

    private void disposeSubscribe1() {
        if (mDisposable1 != null) {
            mDisposable1.dispose();
            mDisposable1 = null;
            mSubscribe1In.clear();
            updateMessage();
        }
    }

    private void startSubscribe2() {
        if (mConvertObservable != null && mDisposable2 == null) {
            mDisposable2 = mConvertObservable.subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    Log.d(TAG, "订阅者2收到数据=" + integer + ",接收线程ID=" + Thread.currentThread().getId());
                    mSubscribe2In.add(integer);
                    updateMessage();
                }
            });
        }
    }

    private void disposeSubscribe2() {
        if (mDisposable2 != null) {
            mDisposable2.dispose();
            mDisposable2 = null;
            mSubscribe2In.clear();
            updateMessage();
        }
    }

为了验证之前说到的几个特点，进入程序之后，我们会先创建该Cold Observable，之后进行一系列的操作，效果如下：


在上面的图中，我们做了一下几步操作：

三、由 Cold Observable 转换的 ConnectableObservable

在了解完Cold Observable之后，我们再来看第二类的Observable，它的类型为ConnectableObservable，它是通过Cold Observable经过下面两种方式生成的：

如果使用.publish()创建，那么订阅者只能收到在订阅之后Cold Observable发出的数据，而如果使用reply(int N)创建，那么订阅者在订阅后可以收到Cold Observable在订阅之前发送的N个数据。

我们先以publish()为例，介绍ConnectableObservable的几个特点：

下面，我们创建一个ConnectableObservable，两个订阅者之后会订阅到它，而不是Cold Observable：

    //.publish()将源Observable转换成为HotObservable，当调用它的connect方法后，无论此时有没有订阅者，源Observable都开始发送数据，订阅者订阅后将可以收到数据，并且订阅者解除订阅不会影响源Observable数据的发射。
    public void createPublishSource() {
        mColdObservable = getSource();
        mConvertObservable = mColdObservable.publish();
        mConvertDisposable = ((ConnectableObservable<Integer>) mConvertObservable).connect();
    }

和上面一样，还是用一个例子来演示，该例子的效果为：


上面这些现象发生的根本原因在于：现在Observer和Observer2都是订阅到ConnectableObservable，真正产生数据的Cold Observable并不知道他们的存在，和它交互的是ConnectableObservable，ConnectableObservable相当于一个中介，它完成下面两项任务：

四、由 ConnectableObservable 转换成 Observable

由ConnectableObservable转换成Observable有两种方法，我们分为两节介绍下当订阅到转换后的Observable时的现象：

4.1 ConnectableObservable 由 refCount 转换成 Observable

经过refCount方法，ConnectableObservable可以转换成正常的Observable，我们称为refObservable，这里我们假设ConnectableObservable是由Cold Observable通过publish()方法转换的，对于它的订阅者，有以下几个特点：

接着上例子，我们创建一个refObservable：

    //.share()相当于.publish().refCount()，当有订阅者订阅时，源订阅者会开始发送数据，如果所有的订阅者都取消订阅，源Observable就会停止发送数据。
    private void createShareSource() {
        mColdObservable = getSource();
        mConvertObservable = mColdObservable.publish().refCount();
    }

示例如下：



操作分为以下几步：

最后说明一点：订阅到Cold Observable的.publish().refCount()和Cold Observable的share()所返回的Observable是等价的。

4.2 ConnectableObservable 由 autoConnect(int N) 转换成 Observable

autoConnect(int N)和refCount很类似，都是将ConnectableObservable转换成普通的Observable，我们称为autoObservable，同样我们先假设ConnectableObservable是由Cold Observable通过publish()方法生成的，它有以下几个特点：

其创建方法如下所示：

    //.autoConnect在有指定个订阅者时开始让源Observable发送消息，但是订阅者是否取消订阅不会影响到源Observable的发射。
    private void createAutoConnectSource() {
        mColdObservable = getSource();
        mConvertObservable = mColdObservable.publish().autoConnect(1, new Consumer<Disposable>() {
            @Override
            public void accept(Disposable disposable) throws Exception {
                mConvertDisposable = disposable;
            }
        });
    }

示例效果如下：



我们进行了如下几步操作：

五、publish 和 reply(int N) 的区别

在上面的例子当中，所有总结的特点都是建立在ConnectableObservable是由publish()生成，只所以这么做，是为了方便大家理解，无论是订阅到ConnectableObservable，还是由ConnectableObservable转换的refObservable和autoObservable，使用这两种方式创建的唯一区别就是，订阅者在订阅后，如果是通过publish()创建的，那么订阅者之后收到订阅后Cold Observable发送的数据；而如果是reply(int N)创建的，那么订阅者还能额外收到N个之前Cold Observable发送的数据，我们用下面一个小例子来演示，订阅者订阅到的Observable如下：

    //.reply会让缓存源Observable的N个数据项，当有新的订阅者订阅时，它会发送这N个数据项给它。
    private void createReplySource() {
        mColdObservable = getSource();
        mConvertObservable = mColdObservable.replay(3);
        mConvertDisposable = ((ConnectableObservable<Integer>) mConvertObservable).connect();
    }

示例演示效果：



操作步骤：

最后再提一下，更详细的代码大家可以从 RxSample 的第十二章中获取。

更多文章，欢迎访问我的 Android 知识梳理系列：

上一篇下一篇

猜你喜欢

热点阅读