RxJava2 實(shí)戰(zhàn)知識(shí)梳理(8) - 使用 publish + merge 優(yōu)化先加載緩存，再讀取網(wǎng)絡(luò)數(shù)據(jù)的請(qǐng)求過(guò)程

RxJava2 實(shí)戰(zhàn)系列文章

一、前言

在很多資訊應(yīng)用當(dāng)中汁果，當(dāng)我們進(jìn)入一個(gè)新的頁(yè)面涡拘，為了提升用戶體驗(yàn)，不讓頁(yè)面空白太久据德，我們一般會(huì)先讀取緩存中的數(shù)據(jù)鳄乏，再去請(qǐng)求網(wǎng)絡(luò)。

今天這篇文章棘利，我們將實(shí)現(xiàn)下面這個(gè)效果：同時(shí)發(fā)起讀取緩存橱野、訪問(wèn)網(wǎng)絡(luò)的請(qǐng)求，如果緩存的數(shù)據(jù)先回來(lái)赡译，那么就先展示緩存的數(shù)據(jù)仲吏，而如果網(wǎng)絡(luò)的數(shù)據(jù)先回來(lái)，那么就不再展示緩存的數(shù)據(jù)蝌焚。

為了讓大家對(duì)這一過(guò)程有更深刻的理解裹唆，我們介紹"先加載緩存，再請(qǐng)求網(wǎng)絡(luò)"這種模型的四種實(shí)現(xiàn)方式只洒，其中第四種實(shí)現(xiàn)可以達(dá)到上面我們所說(shuō)的效果许帐，而前面的三種實(shí)現(xiàn)雖然也能夠?qū)崿F(xiàn)相同的需求，并且可以正常工作毕谴，但是在某些特殊情況下成畦，會(huì)出現(xiàn)意想不到的情況：

使用concat實(shí)現(xiàn)
使用concatEager實(shí)現(xiàn)
使用merge實(shí)現(xiàn)
使用publish實(shí)現(xiàn)

二距芬、示例

2.1 準(zhǔn)備工作

我們需要準(zhǔn)備兩個(gè)Observable，分別表示 緩存數(shù)據(jù)源 和 網(wǎng)絡(luò)數(shù)據(jù)源循帐，在其中填入相應(yīng)的緩存數(shù)據(jù)和網(wǎng)絡(luò)數(shù)據(jù)框仔，為了之后演示一些特殊的情況，我們可以在創(chuàng)建它的時(shí)候指定它執(zhí)行的時(shí)間：

   //模擬緩存數(shù)據(jù)源拄养。
    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, "開始加載緩存數(shù)據(jù)");
                    Thread.sleep(simulateTime);
                    List<NewsResultEntity> results = new ArrayList<>();
                    for (int i = 0; i < 10; i++) {
                        NewsResultEntity entity = new NewsResultEntity();
                        entity.setType("緩存");
                        entity.setDesc("序號(hào)=" + i);
                        results.add(entity);
                    }
                    observableEmitter.onNext(results);
                    observableEmitter.onComplete();
                    Log.d(TAG, "結(jié)束加載緩存數(shù)據(jù)");
                } catch (InterruptedException e) {
                    if (!observableEmitter.isDisposed()) {
                        observableEmitter.onError(e);
                    }
                }
            }
        });
    }
    //模擬網(wǎng)絡(luò)數(shù)據(jù)源离斩。
    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, "開始加載網(wǎng)絡(luò)數(shù)據(jù)");
                    Thread.sleep(simulateTime);
                    List<NewsResultEntity> results = new ArrayList<>();
                    for (int i = 0; i < 10; i++) {
                        NewsResultEntity entity = new NewsResultEntity();
                        entity.setType("網(wǎng)絡(luò)");
                        entity.setDesc("序號(hào)=" + i);
                        results.add(entity);
                    }
                    observableEmitter.onNext(results);
                    observableEmitter.onComplete();
                    Log.d(TAG, "結(jié)束加載網(wǎng)絡(luò)數(shù)據(jù)");
                } catch (InterruptedException e) {
                    if (!observableEmitter.isDisposed()) {
                        observableEmitter.onError(e);
                    }
                }
            }
        });
    }

在最終的下游，我們接收數(shù)據(jù)瘪匿，并在頁(yè)面上通過(guò)RecyclerView進(jìn)行展示：

    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, "加載錯(cuò)誤, e=" + throwable);
            }

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

2.2 使用 concat 實(shí)現(xiàn)

concat是很多文章都推薦使用的方式跛梗，因?yàn)樗粫?huì)有任何問(wèn)題，實(shí)現(xiàn)代碼如下：

   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);
    }

上面這段代碼的運(yùn)行結(jié)果為：

從控制臺(tái)的輸出可以看到棋弥，整個(gè)過(guò)程是先取讀取緩存核偿，等緩存的數(shù)據(jù)讀取完畢之后，才開始請(qǐng)求網(wǎng)絡(luò)顽染，因此整個(gè)過(guò)程的耗時(shí)為兩個(gè)階段的相加漾岳，即2500ms。

它的原理圖如下所示：

concat 原理圖

從原理圖中也驗(yàn)證了我們前面的現(xiàn)象家乘，它會(huì)連接多個(gè)Observable蝗羊，并且必須要等到前一個(gè)Observable的所有數(shù)據(jù)項(xiàng)都發(fā)送完之后，才會(huì)開始下一個(gè)Observable數(shù)據(jù)的發(fā)送仁锯。

那么耀找，concat操作符的缺點(diǎn)是什么呢？很明顯业崖，我們白白浪費(fèi)了前面讀取緩存的這段時(shí)間野芒，能不能同時(shí)發(fā)起讀取緩存和網(wǎng)絡(luò)的請(qǐng)求，而不是等到讀取緩存完畢之后双炕，才去請(qǐng)求網(wǎng)絡(luò)呢狞悲？

2.3 使用 concatEager 實(shí)現(xiàn)

為了解決前面沒(méi)有同時(shí)發(fā)起請(qǐng)求的問(wèn)題，我們可以使用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最大的不同就是多個(gè)Observable可以同時(shí)開始發(fā)射數(shù)據(jù)摇锋，如果后一個(gè)Observable發(fā)射完成后，前一個(gè)Observable還有發(fā)射完數(shù)據(jù)站超，那么它會(huì)將后一個(gè)Observable的數(shù)據(jù)先緩存起來(lái)荸恕，等到前一個(gè)Observable發(fā)射完畢后，才將緩存的數(shù)據(jù)發(fā)射出去死相。

上面代碼中融求，請(qǐng)求緩存的時(shí)長(zhǎng)改為500ms，而請(qǐng)求網(wǎng)絡(luò)的時(shí)長(zhǎng)改為2000ms算撮，運(yùn)行結(jié)果為：

那么這種實(shí)現(xiàn)方式的缺點(diǎn)是什么呢生宛？就是在某些異常情況下县昂，如果讀取緩存的時(shí)間要大于網(wǎng)絡(luò)請(qǐng)求的時(shí)間，那么就會(huì)導(dǎo)致出現(xiàn)“網(wǎng)絡(luò)請(qǐng)求的結(jié)果”等待“讀取緩存”這一過(guò)程完成后才能傳遞給下游陷舅，白白浪費(fèi)了一段時(shí)間倒彰。

我們將請(qǐng)求緩存的時(shí)長(zhǎng)改為2000ms，而請(qǐng)求網(wǎng)絡(luò)的時(shí)長(zhǎng)改為500ms蔑赘，查看控制臺(tái)的輸出狸驳，可以驗(yàn)證上面的結(jié)論：

2.4 使用 merge 實(shí)現(xiàn)

下面，我們來(lái)看一下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一樣缩赛，會(huì)讓多個(gè)Observable同時(shí)開始發(fā)射數(shù)據(jù)，但是它不需要Observable之間的互相等待撰糠，而是直接發(fā)送給下游酥馍。

當(dāng)緩存時(shí)間為500ms，而請(qǐng)求網(wǎng)絡(luò)時(shí)間為2000ms時(shí)阅酪，它的結(jié)果為：

在讀取緩存的時(shí)間小于請(qǐng)求網(wǎng)絡(luò)的時(shí)間時(shí)旨袒，這個(gè)操作符能夠很好的工作，但是反之术辐，就會(huì)出現(xiàn)我們先展示了網(wǎng)絡(luò)的數(shù)據(jù)砚尽，然后又被刷新成舊的緩存數(shù)據(jù)。

發(fā)生該異常時(shí)的現(xiàn)象如下所示：

2.5 使用 publish 實(shí)現(xiàn)

使用publish的實(shí)現(xiàn)如下所示：

    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);
    }

這里面一共涉及到了三個(gè)操作符辉词，publish必孤、merge和takeUnti，我們先來(lái)看一下它能否解決我們之前三種方式的缺陷：

讀取緩存的時(shí)間為500ms瑞躺，請(qǐng)求網(wǎng)絡(luò)的時(shí)間為2000ms

讀取緩存的時(shí)間為2000ms敷搪，請(qǐng)求網(wǎng)絡(luò)的時(shí)間為500ms

可以看到，在讀取緩存的時(shí)間大于請(qǐng)求網(wǎng)絡(luò)時(shí)間的時(shí)候幢哨，僅僅只會(huì)展示網(wǎng)絡(luò)的數(shù)據(jù)赡勘，顯示效果為：

并且讀取緩存和請(qǐng)求網(wǎng)絡(luò)是同時(shí)發(fā)起的，很好地解決了前面幾種實(shí)現(xiàn)方式的缺陷捞镰。

這里要感謝簡(jiǎn)友無(wú)心下棋在評(píng)論里提到的問(wèn)題：如果網(wǎng)絡(luò)請(qǐng)求先返回時(shí)發(fā)生了錯(cuò)誤（例如沒(méi)有網(wǎng)絡(luò)等）導(dǎo)致發(fā)送了onError事件闸与，從而使得緩存的Observable也無(wú)法發(fā)送事件，最后界面顯示空白岸售。

針對(duì)這個(gè)問(wèn)題践樱，我們需要對(duì)網(wǎng)絡(luò)的Observable進(jìn)行優(yōu)化，讓其不將onError事件傳遞給下游冰评。其中一種解決方式是通過(guò)使用onErrorResume操作符映胁，它可以接收一個(gè)Func函數(shù)，其形參為網(wǎng)絡(luò)發(fā)送的錯(cuò)誤甲雅，而在上游發(fā)生錯(cuò)誤時(shí)會(huì)回調(diào)該函數(shù)解孙。我們可以根據(jù)錯(cuò)誤的類型來(lái)返回一個(gè)新的Observable坑填，讓訂閱者鏡像到這個(gè)新的Observable，并且忽略onError事件弛姜，從而避免onError事件導(dǎo)致整個(gè)訂閱關(guān)系的結(jié)束脐瑰。

這里為了避免訂閱者在鏡像到新的Observable時(shí)會(huì)收到額外的時(shí)間，我們返回一個(gè)Observable.never()廷臼，它表示一個(gè)永遠(yuǎn)不發(fā)送事件的上游苍在。

    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, "開始加載網(wǎng)絡(luò)數(shù)據(jù)");
                    Thread.sleep(simulateTime);
                    List<NewsResultEntity> results = new ArrayList<>();
                    for (int i = 0; i < 10; i++) {
                        NewsResultEntity entity = new NewsResultEntity();
                        entity.setType("網(wǎng)絡(luò)");
                        entity.setDesc("序號(hào)=" + i);
                        results.add(entity);
                    }
                    //a.正常情況。
                    //observableEmitter.onNext(results);
                    //observableEmitter.onComplete();
                    //b.發(fā)生異常荠商。
                    observableEmitter.onError(new Throwable("netWork Error"));
                    Log.d(TAG, "結(jié)束加載網(wǎng)絡(luò)數(shù)據(jù)");
                } 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, "網(wǎng)絡(luò)請(qǐng)求發(fā)生錯(cuò)誤throwable=" + throwable);
                return Observable.never();
            }
        });
    }

當(dāng)發(fā)生錯(cuò)誤時(shí)寂恬，控制臺(tái)的輸出如下，可以看到緩存仍然正常地發(fā)送給了下游：

下面莱没，我們就來(lái)分析一下它的實(shí)現(xiàn)原理初肉。

2.5.1 takeUntil

takeUntil的原理圖如下所示：

這里，我們給sourceObservable通過(guò)takeUntil傳入了另一個(gè)otherObservable饰躲，它表示sourceObservable在otherObservable發(fā)射數(shù)據(jù)之后牙咏，就不允許再發(fā)射數(shù)據(jù)了，這就剛好滿足了我們前面說(shuō)的“只要網(wǎng)絡(luò)源發(fā)送了數(shù)據(jù)嘹裂，那么緩存源就不應(yīng)再發(fā)射數(shù)據(jù)”妄壶。

之后，我們?cè)儆们懊娼榻B過(guò)的merge操作符寄狼，讓兩個(gè)緩存源和網(wǎng)絡(luò)源同時(shí)開始工作丁寄，去取數(shù)據(jù)。

2.5.2 publish

但是上面有一點(diǎn)缺陷例嘱，就是調(diào)用merge和takeUntil會(huì)發(fā)生兩次訂閱狡逢，這時(shí)候就需要使用publish操作符，它接收一個(gè)Function函數(shù)拼卵，該函數(shù)返回一個(gè)Observable奢浑，該Observable是對(duì)原Observable，也就是上面網(wǎng)絡(luò)源的Observable轉(zhuǎn)換之后的結(jié)果腋腮，該Observable可以被takeUntil和merge操作符所共享雀彼，從而實(shí)現(xiàn)只訂閱一次的效果。

publish的原理圖如下所示：

publish 原理圖

更多文章即寡，歡迎訪問(wèn)我的 Android 知識(shí)梳理系列：

Android 知識(shí)梳理目錄：http://www.reibang.com/p/fd82d18994ce
個(gè)人主頁(yè)：http://lizejun.cn
個(gè)人知識(shí)總結(jié)目錄：http://lizejun.cn/categories/

最后編輯于：2017.12.10 05:00:17

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者

人面猴
序言：七十年代末徊哑，一起剝皮案震驚了整個(gè)濱河市，隨后出現(xiàn)的幾起案子聪富，更是在濱河造成了極大的恐慌莺丑，老刑警劉巖，帶你破解...
沈念sama閱讀 206,126評(píng)論 6贊 481
死咒
序言：濱河連續(xù)發(fā)生了三起死亡事件，死亡現(xiàn)場(chǎng)離奇詭異梢莽，居然都是意外死亡萧豆，警方通過(guò)查閱死者的電腦和手機(jī)，發(fā)現(xiàn)死者居然都...
沈念sama閱讀 88,254評(píng)論 2贊 382
救了他兩次的神仙讓他今天三更去死
文/潘曉璐我一進(jìn)店門昏名，熙熙樓的掌柜王于貴愁眉苦臉地迎上來(lái)涮雷，“玉大人，你說(shuō)我怎么就攤上這事轻局『檠迹” “怎么了？”我有些...
開封第一講書人閱讀 152,445評(píng)論 0贊 341
道士緝兇錄：失蹤的賣姜人
文/不壞的土叔我叫張陵仑扑，是天一觀的道長(zhǎng)览爵。經(jīng)常有香客問(wèn)我，道長(zhǎng)夫壁，這世上最難降的妖魔是什么拾枣？我笑而不...
開封第一講書人閱讀 55,185評(píng)論 1贊 278
?港島之戀（遺憾婚禮）
正文為了忘掉前任，我火速辦了婚禮盒让，結(jié)果婚禮上，老公的妹妹穿的比我還像新娘司蔬。我一直安慰自己邑茄，他們只是感情好，可當(dāng)我...
茶點(diǎn)故事閱讀 64,178評(píng)論 5贊 371
惡毒庶女頂嫁案：這布局不是一般人想出來(lái)的
文/花漫我一把揭開白布俊啼。她就那樣靜靜地躺著肺缕，像睡著了一般。火紅的嫁衣襯著肌膚如雪授帕。梳的紋絲不亂的頭發(fā)上同木，一...
開封第一講書人閱讀 48,970評(píng)論 1贊 284
城市分裂傳說(shuō)
那天，我揣著相機(jī)與錄音跛十，去河邊找鬼彤路。笑死，一個(gè)胖子當(dāng)著我的面吹牛芥映，可吹牛的內(nèi)容都是我干的洲尊。我是一名探鬼主播，決...
沈念sama閱讀 38,276評(píng)論 3贊 399
雙鴛鴦連環(huán)套：你想象不到人心有多黑
文/蒼蘭香墨我猛地睜開眼奈偏，長(zhǎng)吁一口氣：“原來(lái)是場(chǎng)噩夢(mèng)啊……” “哼坞嘀！你這毒婦竟也來(lái)了？” 一聲冷哼從身側(cè)響起惊来，我...
開封第一講書人閱讀 36,927評(píng)論 0贊 259
萬(wàn)榮殺人案實(shí)錄
序言：老撾萬(wàn)榮一對(duì)情侶失蹤丽涩，失蹤者是張志新（化名）和其女友劉穎，沒(méi)想到半個(gè)月后裁蚁，有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體矢渊，經(jīng)...
沈念sama閱讀 43,400評(píng)論 1贊 300
?護(hù)林員之死
正文獨(dú)居荒郊野嶺守林人離奇死亡继准，尸身上長(zhǎng)有42處帶血的膿包…… 初始之章·張勛以下內(nèi)容為張勛視角年9月15日...
茶點(diǎn)故事閱讀 35,883評(píng)論 2贊 323
?白月光啟示錄
正文我和宋清朗相戀三年，在試婚紗的時(shí)候發(fā)現(xiàn)自己被綠了昆淡。大學(xué)時(shí)的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片锰瘸。...
茶點(diǎn)故事閱讀 37,997評(píng)論 1贊 333
活死人
序言：一個(gè)原本活蹦亂跳的男人離奇死亡，死狀恐怖昂灵，靈堂內(nèi)的尸體忽然破棺而出避凝，到底是詐尸還是另有隱情，我是刑警寧澤眨补，帶...
沈念sama閱讀 33,646評(píng)論 4贊 322
?日本核電站爆炸內(nèi)幕
正文年R本政府宣布管削，位于F島的核電站，受9級(jí)特大地震影響撑螺，放射性物質(zhì)發(fā)生泄漏含思。R本人自食惡果不足惜，卻給世界環(huán)境...
茶點(diǎn)故事閱讀 39,213評(píng)論 3贊 307
男人毒藥：我在死后第九天來(lái)索命
文/蒙蒙一甘晤、第九天我趴在偏房一處隱蔽的房頂上張望含潘。院中可真熱鬧，春花似錦线婚、人聲如沸遏弱。這莊子的主人今日做“春日...
開封第一講書人閱讀 30,204評(píng)論 0贊 19
一樁弒父案塞弊，背后竟有這般陰謀
文/蒼蘭香墨我抬頭看了看天上的太陽(yáng)漱逸。三九已至，卻和暖如春游沿，著一層夾襖步出監(jiān)牢的瞬間饰抒，已是汗流浹背。一陣腳步聲響...
開封第一講書人閱讀 31,423評(píng)論 1贊 260
情欲美人皮
我被黑心中介騙來(lái)泰國(guó)打工诀黍，沒(méi)想到剛下飛機(jī)就差點(diǎn)兒被人妖公主榨干…… 1. 我叫王不留袋坑，地道東北人。一個(gè)月前我還...
沈念sama閱讀 45,423評(píng)論 2贊 352
代替公主和親
正文我出身青樓蔗草，卻偏偏與公主長(zhǎng)得像咒彤，于是被迫代替她去往敵國(guó)和親。傳聞我的和親對(duì)象是個(gè)殘疾皇子咒精，可洞房花燭夜當(dāng)晚...
茶點(diǎn)故事閱讀 42,722評(píng)論 2贊 345

RxJava2 實(shí)戰(zhàn)知識(shí)梳理(8) - 使用 publish + merge 優(yōu)化先加載緩存遮精，再讀取網(wǎng)絡(luò)數(shù)據(jù)的請(qǐng)求過(guò)程