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 应用

一、前言

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

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

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

• 使用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 原理图

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

• Android 知识梳理目录：http://www.jianshu.com/p/fd82d18994ce
• 个人主页：http://lizejun.cn
• 个人知识总结目录：http://lizejun.cn/categories/