RxJava2 实战知识梳理(8) - 使用 publish + merge 优化先加载缓存，再读取网络数据的请求过程

RxJava2 实战系列文章

一、前言

在很多资讯应用当中，当我们进入一个新的页面，为了提升用户体验，不让页面空白太久，我们一般会先读取缓存中的数据，再去请求网络。

今天这篇文章，我们将实现下面这个效果：同时发起读取缓存、访问网络的请求，如果缓存的数据先回来，那么就先展示缓存的数据，而如果网络的数据先回来，那么就不再展示缓存的数据。

为了让大家对这一过程有更深刻的理解，我们介绍"先加载缓存，再请求网络"这种模型的四种实现方式，其中第四种实现可以达到上面我们所说的效果，而前面的三种实现虽然也能够实现相同的需求，并且可以正常工作，但是在某些特殊情况下，会出现意想不到的情况：

使用concat实现
使用concatEager实现
使用merge实现
使用publish实现

二、示例

2.1 准备工作

我们需要准备两个Observable，分别表示 缓存数据源 和 网络数据源，在其中填入相应的缓存数据和网络数据，为了之后演示一些特殊的情况，我们可以在创建它的时候指定它执行的时间：

   //模拟缓存数据源。
    private Observable<List<NewsResultEntity>> getCacheArticle(final long simulateTime) {
        return Observable.create(new ObservableOnSubscribe<List<NewsResultEntity>>() {
            @Override
            public void subscribe(ObservableEmitter<List<NewsResultEntity>> observableEmitter) throws Exception {
                try {
                    Log.d(TAG, "开始加载缓存数据");
                    Thread.sleep(simulateTime);
                    List<NewsResultEntity> results = new ArrayList<>();
                    for (int i = 0; i < 10; i++) {
                        NewsResultEntity entity = new NewsResultEntity();
                        entity.setType("缓存");
                        entity.setDesc("序号=" + i);
                        results.add(entity);
                    }
                    observableEmitter.onNext(results);
                    observableEmitter.onComplete();
                    Log.d(TAG, "结束加载缓存数据");
                } catch (InterruptedException e) {
                    if (!observableEmitter.isDisposed()) {
                        observableEmitter.onError(e);
                    }
                }
            }
        });
    }
    //模拟网络数据源。
    private Observable<List<NewsResultEntity>> getNetworkArticle(final long simulateTime) {
        return Observable.create(new ObservableOnSubscribe<List<NewsResultEntity>>() {
            @Override
            public void subscribe(ObservableEmitter<List<NewsResultEntity>> observableEmitter) throws Exception {
                try {
                    Log.d(TAG, "开始加载网络数据");
                    Thread.sleep(simulateTime);
                    List<NewsResultEntity> results = new ArrayList<>();
                    for (int i = 0; i < 10; i++) {
                        NewsResultEntity entity = new NewsResultEntity();
                        entity.setType("网络");
                        entity.setDesc("序号=" + i);
                        results.add(entity);
                    }
                    observableEmitter.onNext(results);
                    observableEmitter.onComplete();
                    Log.d(TAG, "结束加载网络数据");
                } catch (InterruptedException e) {
                    if (!observableEmitter.isDisposed()) {
                        observableEmitter.onError(e);
                    }
                }
            }
        });
    }

在最终的下游，我们接收数据，并在页面上通过RecyclerView进行展示：

    private DisposableObserver<List<NewsResultEntity>> getArticleObserver() {
        return new DisposableObserver<List<NewsResultEntity>>() {

            @Override
            public void onNext(List<NewsResultEntity> newsResultEntities) {
                mNewsResultEntities.clear();
                mNewsResultEntities.addAll(newsResultEntities);
                mNewsAdapter.notifyDataSetChanged();
            }

            @Override
            public void onError(Throwable throwable) {
                Log.d(TAG, "加载错误, e=" + throwable);
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "加载完成");
            }
        };
    }

2.2 使用 concat 实现

concat是很多文章都推荐使用的方式，因为它不会有任何问题，实现代码如下：

   private void refreshArticleUseContact() {
        Observable<List<NewsResultEntity>> contactObservable = Observable.concat(
                getCacheArticle(500).subscribeOn(Schedulers.io()), getNetworkArticle(2000).subscribeOn(Schedulers.io()));
        DisposableObserver<List<NewsResultEntity>> disposableObserver = getArticleObserver();
        contactObservable.observeOn(AndroidSchedulers.mainThread()).subscribe(disposableObserver);
    }

上面这段代码的运行结果为：

从控制台的输出可以看到，整个过程是先取读取缓存，等缓存的数据读取完毕之后，才开始请求网络，因此整个过程的耗时为两个阶段的相加，即2500ms。

它的原理图如下所示：

concat 原理图

从原理图中也验证了我们前面的现象，它会连接多个Observable，并且必须要等到前一个Observable的所有数据项都发送完之后，才会开始下一个Observable数据的发送。

那么，concat操作符的缺点是什么呢？很明显，我们白白浪费了前面读取缓存的这段时间，能不能同时发起读取缓存和网络的请求，而不是等到读取缓存完毕之后，才去请求网络呢？

2.3 使用 concatEager 实现

为了解决前面没有同时发起请求的问题，我们可以使用concatEager，它的使用方法如下：

    private void refreshArticleUseConcatEager() {
        List<Observable<List<NewsResultEntity>>> observables = new ArrayList<>();
        observables.add(getCacheArticle(500).subscribeOn(Schedulers.io()));
        observables.add(getNetworkArticle(2000).subscribeOn(Schedulers.io()));
        Observable<List<NewsResultEntity>> contactObservable = Observable.concatEager(observables);
        DisposableObserver<List<NewsResultEntity>> disposableObserver = getArticleObserver();
        contactObservable.observeOn(AndroidSchedulers.mainThread()).subscribe(disposableObserver);
    }

它和concat最大的不同就是多个Observable可以同时开始发射数据，如果后一个Observable发射完成后，前一个Observable还有发射完数据，那么它会将后一个Observable的数据先缓存起来，等到前一个Observable发射完毕后，才将缓存的数据发射出去。

上面代码中，请求缓存的时长改为500ms，而请求网络的时长改为2000ms，运行结果为：

那么这种实现方式的缺点是什么呢？就是在某些异常情况下，如果读取缓存的时间要大于网络请求的时间，那么就会导致出现“网络请求的结果”等待“读取缓存”这一过程完成后才能传递给下游，白白浪费了一段时间。

我们将请求缓存的时长改为2000ms，而请求网络的时长改为500ms，查看控制台的输出，可以验证上面的结论：

2.4 使用 merge 实现

下面，我们来看一下merge操作符的示例：

    private void refreshArticleUseMerge() {
        Observable<List<NewsResultEntity>> contactObservable = Observable.merge(
                getCacheArticle(500).subscribeOn(Schedulers.io()), getNetworkArticle(2000).subscribeOn(Schedulers.io()));
        DisposableObserver<List<NewsResultEntity>> disposableObserver = getArticleObserver();
        contactObservable.observeOn(AndroidSchedulers.mainThread()).subscribe(disposableObserver);
    }

merge的原理图如下所示：

merge 原理图

它和concatEager一样，会让多个Observable同时开始发射数据，但是它不需要Observable之间的互相等待，而是直接发送给下游。

当缓存时间为500ms，而请求网络时间为2000ms时，它的结果为：

在读取缓存的时间小于请求网络的时间时，这个操作符能够很好的工作，但是反之，就会出现我们先展示了网络的数据，然后又被刷新成旧的缓存数据。

发生该异常时的现象如下所示：

2.5 使用 publish 实现

使用publish的实现如下所示：

    private void refreshArticleUsePublish() {
        Observable<List<NewsResultEntity>> publishObservable = getNetworkArticle(2000).subscribeOn(Schedulers.io()).publish(new Function<Observable<List<NewsResultEntity>>, ObservableSource<List<NewsResultEntity>>>() {

            @Override
            public ObservableSource<List<NewsResultEntity>> apply(Observable<List<NewsResultEntity>> network) throws Exception {
                return Observable.merge(network, getCacheArticle(500).subscribeOn(Schedulers.io()).takeUntil(network));
            }

        });
        DisposableObserver<List<NewsResultEntity>> disposableObserver = getArticleObserver();
        publishObservable.observeOn(AndroidSchedulers.mainThread()).subscribe(disposableObserver);
    }

这里面一共涉及到了三个操作符，publish、merge和takeUnti，我们先来看一下它能否解决我们之前三种方式的缺陷：

读取缓存的时间为500ms，请求网络的时间为2000ms

读取缓存的时间为2000ms，请求网络的时间为500ms

可以看到，在读取缓存的时间大于请求网络时间的时候，仅仅只会展示网络的数据，显示效果为：

并且读取缓存和请求网络是同时发起的，很好地解决了前面几种实现方式的缺陷。

这里要感谢简友无心下棋在评论里提到的问题：如果网络请求先返回时发生了错误（例如没有网络等）导致发送了onError事件，从而使得缓存的Observable也无法发送事件，最后界面显示空白。

针对这个问题，我们需要对网络的Observable进行优化，让其不将onError事件传递给下游。其中一种解决方式是通过使用onErrorResume操作符，它可以接收一个Func函数，其形参为网络发送的错误，而在上游发生错误时会回调该函数。我们可以根据错误的类型来返回一个新的Observable，让订阅者镜像到这个新的Observable，并且忽略onError事件，从而避免onError事件导致整个订阅关系的结束。

这里为了避免订阅者在镜像到新的Observable时会收到额外的时间，我们返回一个Observable.never()，它表示一个永远不发送事件的上游。

    private Observable<List<NewsResultEntity>> getNetworkArticle(final long simulateTime) {
        return Observable.create(new ObservableOnSubscribe<List<NewsResultEntity>>() {
            @Override
            public void subscribe(ObservableEmitter<List<NewsResultEntity>> observableEmitter) throws Exception {
                try {
                    Log.d(TAG, "开始加载网络数据");
                    Thread.sleep(simulateTime);
                    List<NewsResultEntity> results = new ArrayList<>();
                    for (int i = 0; i < 10; i++) {
                        NewsResultEntity entity = new NewsResultEntity();
                        entity.setType("网络");
                        entity.setDesc("序号=" + i);
                        results.add(entity);
                    }
                    //a.正常情况。
                    //observableEmitter.onNext(results);
                    //observableEmitter.onComplete();
                    //b.发生异常。
                    observableEmitter.onError(new Throwable("netWork Error"));
                    Log.d(TAG, "结束加载网络数据");
                } catch (InterruptedException e) {
                    if (!observableEmitter.isDisposed()) {
                        observableEmitter.onError(e);
                    }
                }
            }
        }).onErrorResumeNext(new Function<Throwable, ObservableSource<? extends List<NewsResultEntity>>>() {

            @Override
            public ObservableSource<? extends List<NewsResultEntity>> apply(Throwable throwable) throws Exception {
                Log.d(TAG, "网络请求发生错误throwable=" + throwable);
                return Observable.never();
            }
        });
    }

当发生错误时，控制台的输出如下，可以看到缓存仍然正常地发送给了下游：

下面，我们就来分析一下它的实现原理。

2.5.1 takeUntil

takeUntil的原理图如下所示：

这里，我们给sourceObservable通过takeUntil传入了另一个otherObservable，它表示sourceObservable在otherObservable发射数据之后，就不允许再发射数据了，这就刚好满足了我们前面说的“只要网络源发送了数据，那么缓存源就不应再发射数据”。

之后，我们再用前面介绍过的merge操作符，让两个缓存源和网络源同时开始工作，去取数据。

2.5.2 publish

但是上面有一点缺陷，就是调用merge和takeUntil会发生两次订阅，这时候就需要使用publish操作符，它接收一个Function函数，该函数返回一个Observable，该Observable是对原Observable，也就是上面网络源的Observable转换之后的结果，该Observable可以被takeUntil和merge操作符所共享，从而实现只订阅一次的效果。

publish的原理图如下所示：

publish 原理图

RxJava2 实战知识梳理(8) - 使用 publish + merge 优化先加载缓存，再读取网络数据的请求过程

RxJava2 实战知识梳理(8) - 使用 publish + merge 优化先加载缓存，再读取网络数据的请求过程

RxJava2 实战系列文章

一、前言

二、示例

2.1 准备工作

2.2 使用 concat 实现

2.3 使用 concatEager 实现

2.4 使用 merge 实现

2.5 使用 publish 实现

2.5.1 takeUntil

2.5.2 publish

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

推荐阅读更多精彩内容