給初學者的RxJava2.0教程(七)

時間  2019-11-15
標籤 初學者 rxjava2.0 rxjava 教程 欄目 Java 简体版
原文   原文鏈接

Outlinejava

[TOC]react

前言

上一節裏咱們學習了只使用Observable如何去解決上下游流速不均衡的問題, 之因此學習這個是由於Observable仍是有不少它使用的場景, 有些朋友自從據說了Flowable以後就以爲Flowable能解決任何問題, 甚至有拋棄Observable這種想法, 這是萬萬不可的, 它們都有各自的優點和不足. android

在這一節裏咱們先來學習如何使用Flowable, 它東西比較多, 也比較繁瑣, 解釋起來也比較麻煩, 但我仍是儘可能用通俗易懂的話來講清楚, 畢竟, 這是一個通俗易懂的教程.git

正題

咱們仍是以兩根水管舉例子:github

prepare.png

以前咱們所的上游和下游分別是Observable和Observer, 此次不同的是上游變成了Flowable, 下游變成了Subscriber, 可是水管之間的鏈接仍是經過subscribe(), 咱們來看看最基本的用法吧: 多線程

Flowable<Integer> upstream = Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "emit 1");
                emitter.onNext(1);
                Log.d(TAG, "emit 2");
                emitter.onNext(2);
                Log.d(TAG, "emit 3");
                emitter.onNext(3);
                Log.d(TAG, "emit complete");
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR); //增長了一個參數

        Subscriber<Integer> downstream = new Subscriber<Integer>() {

            @Override
            public void onSubscribe(Subscription s) {
                Log.d(TAG, "onSubscribe");
                s.request(Long.MAX_VALUE);  //注意這句代碼
            }

            @Override
            public void onNext(Integer integer) {
                Log.d(TAG, "onNext: " + integer);

            }

            @Override
            public void onError(Throwable t) {
                 Log.w(TAG, "onError: ", t);
            }

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

        upstream.subscribe(downstream);複製代碼

這段代碼中,分別建立了一個上游Flowable和下游Subscriber, 上下游工做在同一個線程中, 和以前的Observable的使用方式只有一點點的區別, 先來看看運行結果吧: app

D/TAG: onSubscribe   
D/TAG: emit 1        
D/TAG: onNext: 1     
D/TAG: emit 2        
D/TAG: onNext: 2     
D/TAG: emit 3        
D/TAG: onNext: 3     
D/TAG: emit complete 
D/TAG: onComplete複製代碼

結果也和咱們預期的是同樣的. 異步

咱們注意到此次和Observable有些不一樣. 首先是建立Flowable的時候增長了一個參數, 這個參數是用來選擇背壓,也就是出現上下游流速不均衡的時候應該怎麼處理的辦法, 這裏咱們直接用BackpressureStrategy.ERROR這種方式, 這種方式會在出現上下游流速不均衡的時候直接拋出一個異常,這個異常就是著名的MissingBackpressureException. 其他的策略後面再來說解.ide

另外的一個區別是在下游的onSubscribe方法中傳給咱們的再也不是Disposable了, 而是Subscription, 它倆有什麼區別呢, 首先它們都是上下游中間的一個開關, 以前咱們說調用Disposable.dispose()方法能夠切斷水管, 一樣的調用Subscription.cancel()也能夠切斷水管, 不一樣的地方在於Subscription增長了一個void request(long n)方法, 這個方法有什麼用呢, 在上面的代碼中也有這麼一句代碼:工具

s.request(Long.MAX_VALUE);複製代碼

這句代碼有什麼用呢, 不要它能夠嗎? 咱們來試試:

Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "emit 1");
                emitter.onNext(1);
                Log.d(TAG, "emit 2");
                emitter.onNext(2);
                Log.d(TAG, "emit 3");
                emitter.onNext(3);
                Log.d(TAG, "emit complete");
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR).subscribe(new Subscriber<Integer>() {

            @Override
            public void onSubscribe(Subscription s) {
                Log.d(TAG, "onSubscribe");
            }

            @Override
            public void onNext(Integer integer) {
                Log.d(TAG, "onNext: " + integer);

            }

            @Override
            public void onError(Throwable t) {
                Log.w(TAG, "onError: ", t);
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete");
            }
        });複製代碼

此次咱們取消掉了request這句代碼, 來看看運行結果:

zlc.season.rxjava2demo D/TAG: onSubscribe
zlc.season.rxjava2demo D/TAG: emit 1
zlc.season.rxjava2demo W/TAG: onError: 
                              io.reactivex.exceptions.MissingBackpressureException: create: could not emit value due to lack of requests
                                  at io.reactivex.internal.operators.flowable.FlowableCreate$ErrorAsyncEmitter.onOverflow(FlowableCreate.java:411)
                                  at io.reactivex.internal.operators.flowable.FlowableCreate$NoOverflowBaseAsyncEmitter.onNext(FlowableCreate.java:377)
                                  at zlc.season.rxjava2demo.demo.ChapterSeven$3.subscribe(ChapterSeven.java:77)
                                  at io.reactivex.internal.operators.flowable.FlowableCreate.subscribeActual(FlowableCreate.java:72)
                                  at io.reactivex.Flowable.subscribe(Flowable.java:12218)
                                  at zlc.season.rxjava2demo.demo.ChapterSeven.demo2(ChapterSeven.java:111)
                                  at zlc.season.rxjava2demo.MainActivity$2.onClick(MainActivity.java:36)
                                  at android.view.View.performClick(View.java:5637)
                                  at android.view.View$PerformClick.run(View.java:22429)
                                  at android.os.Handler.handleCallback(Handler.java:751)
                                  at android.os.Handler.dispatchMessage(Handler.java:95)
                                  at android.os.Looper.loop(Looper.java:154)
                                  at android.app.ActivityThread.main(ActivityThread.java:6119)
                                  at java.lang.reflect.Method.invoke(Native Method)
                                  at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:886)
                                  at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:776)
zlc.season.rxjava2demo D/TAG: emit 2
zlc.season.rxjava2demo D/TAG: emit 3
zlc.season.rxjava2demo D/TAG: emit complete複製代碼

哎哎哎, 大兄弟, 怎麼一言不合就拋異常?

從運行結果中能夠看到, 在上游發送第一個事件以後, 下游就拋出了一個著名的MissingBackpressureException異常, 而且下游沒有收到任何其他的事件. 但是這是一個同步的訂閱呀, 上下游工做在同一個線程, 上游每發送一個事件應該會等待下游處理完了纔會繼續發事件啊, 不可能出現上下游流速不均衡的問題呀.

帶着這個疑問, 咱們再來看看異步的狀況: 

Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "emit 1");
                emitter.onNext(1);
                Log.d(TAG, "emit 2");
                emitter.onNext(2);
                Log.d(TAG, "emit 3");
                emitter.onNext(3);
                Log.d(TAG, "emit complete");
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {

                    @Override
                    public void onSubscribe(Subscription s) {
                        Log.d(TAG, "onSubscribe");
                        mSubscription = s;
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG, "onNext: " + integer);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.w(TAG, "onError: ", t);
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");
                    }
                });複製代碼

此次咱們一樣去掉了request這句代碼, 可是讓上下游工做在不一樣的線程, 來看看運行結果:

zlc.season.rxjava2demo D/TAG: onSubscribe
zlc.season.rxjava2demo D/TAG: emit 1
zlc.season.rxjava2demo D/TAG: emit 2
zlc.season.rxjava2demo D/TAG: emit 3
zlc.season.rxjava2demo D/TAG: emit complete複製代碼

哎, 此次上游正確的發送了全部的事件, 可是下游一個事件也沒有收到. 這是由於什麼呢?

這是由於Flowable在設計的時候採用了一種新的思路也就是響應式拉取的方式來更好的解決上下游流速不均衡的問題, 與咱們以前所講的控制數量和控制速度不太同樣, 這種方式用通俗易懂的話來講就比如是葉問打鬼子, 咱們把上游當作小日本, 把下游看成葉問, 當調用Subscription.request(1)時, 葉問就說我要打一個! 而後小日本就拿出一個鬼子給葉問, 讓他打, 等葉問打死這個鬼子以後, 再次調用request(10), 葉問就又說我要打十個! 而後小日本又派出十個鬼子給葉問, 而後就在邊上看熱鬧, 看葉問能不能打死十個鬼子, 等葉問打死十個鬼子後再繼續要鬼子接着打...

因此咱們把request當作是一種能力, 當成下游處理事件的能力, 下游能處理幾個就告訴上游我要幾個, 這樣只要上游根據下游的處理能力來決定發送多少事件, 就不會形成一窩蜂的發出一堆事件來, 從而致使OOM. 這也就完美的解決以前咱們所學到的兩種方式的缺陷, 過濾事件會致使事件丟失, 減速又可能致使性能損失. 而這種方式既解決了事件丟失的問題, 又解決了速度的問題, 完美 !

可是太完美的東西也就意味着陷阱也會不少, 你可能只是被它的外表所迷惑, 失去了理智, 若是你濫用或者不遵照規則, 同樣會吃到苦頭.

好比這裏須要注意的是, 只有當上游正確的實現了如何根據下游的處理能力來發送事件的時候, 才能達到這種效果, 若是上游根本無論下游的處理能力, 一股腦的瞎他媽發事件, 仍然會產生上下游流速不均衡的問題, 這就比如小日本管他葉問要打幾個, 老子直接拿出1萬個鬼子, 這尼瑪有種打死給我看看? 那麼如何正確的去實現上游呢, 這裏先賣個關子, 以後咱們再來說解.

學習了request, 咱們就能夠解釋上面的兩段代碼了.

首先第一個同步的代碼, 爲何上游發送第一個事件後下遊就拋出了MissingBackpressureException異常, 這是由於下游沒有調用request, 上游就認爲下游沒有處理事件的能力, 而這又是一個同步的訂閱, 既然下游處理不了, 那上游不可能一直等待吧, 若是是這樣, 萬一這兩根水管工做在主線程裏, 界面不就卡死了嗎, 所以只能拋個異常來提醒咱們. 那如何解決這種狀況呢, 很簡單啦, 下游直接調用request(Long.MAX_VALUE)就好了, 或者根據上游發送事件的數量來request就好了, 好比這裏request(3)就能夠了.

而後咱們再來看看第二段代碼, 爲何上下游沒有工做在同一個線程時, 上游卻正確的發送了全部的事件呢? 這是由於在Flowable裏默認有一個大小爲128的水缸, 當上下游工做在不一樣的線程中時, 上游就會先把事件發送到這個水缸中, 所以, 下游雖然沒有調用request, 可是上游在水缸中保存着這些事件, 只有當下遊調用request時, 才從水缸裏取出事件發給下游.

是否是這樣呢, 咱們來驗證一下:

public static void request(long n) {
        mSubscription.request(n); //在外部調用request請求上游
    }

    public static void demo3() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "emit 1");
                emitter.onNext(1);
                Log.d(TAG, "emit 2");
                emitter.onNext(2);
                Log.d(TAG, "emit 3");
                emitter.onNext(3);
                Log.d(TAG, "emit complete");
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {

                    @Override
                    public void onSubscribe(Subscription s) {
                        Log.d(TAG, "onSubscribe");
                        mSubscription = s;  //把Subscription保存起來
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG, "onNext: " + integer);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.w(TAG, "onError: ", t);
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");
                    }
                });
    }複製代碼

這裏咱們把Subscription保存起來, 在界面上增長了一個按鈕, 點擊一次就調用Subscription.request(1), 來看看運行結果:

request.gif

結果彷佛像那麼回事, 上游發送了四個事件保存到了水缸裏, 下游每request一個, 就接收一個進行處理.

剛剛咱們有說到水缸的大小爲128, 有朋友就問了, 你說128就128嗎, 又不是惟品會週年慶, 我不信. 那就來驗證一下:

Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; i < 128; i++) {
                    Log.d(TAG, "emit " + i);
                    emitter.onNext(i);
                }
            }
        }, BackpressureStrategy.ERROR).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {

                    @Override
                    public void onSubscribe(Subscription s) {
                        Log.d(TAG, "onSubscribe");
                        mSubscription = s;
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG, "onNext: " + integer);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.w(TAG, "onError: ", t);
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");
                    }
                });複製代碼

這裏咱們讓上游一次性發送了128個事件, 下游一個也不接收, 來看看運行結果:

zlc.season.rxjava2demo D/TAG: onSubscribe
zlc.season.rxjava2demo D/TAG: emit 0
  ...
zlc.season.rxjava2demo D/TAG: emit 126
zlc.season.rxjava2demo D/TAG: emit 127複製代碼

這段代碼的運行結果很正常, 沒有任何錯誤和異常, 上游僅僅是發送了128個事件.

那來試試129個呢, 把上面代碼中的128改爲129試試:

zlc.season.rxjava2demo D/TAG: onSubscribe
zlc.season.rxjava2demo D/TAG: emit 0
  ...
zlc.season.rxjava2demo D/TAG: emit 126
zlc.season.rxjava2demo D/TAG: emit 127
zlc.season.rxjava2demo D/TAG: emit 128  //這是第129個事件
zlc.season.rxjava2demo W/TAG: onError: 
                              io.reactivex.exceptions.MissingBackpressureException: create: could not emit value due to lack of requests
                                  at io.reactivex.internal.operators.flowable.FlowableCreate$ErrorAsyncEmitter.onOverflow(FlowableCreate.java:411)
                                  at io.reactivex.internal.operators.flowable.FlowableCreate$NoOverflowBaseAsyncEmitter.onNext(FlowableCreate.java:377)
                                  at zlc.season.rxjava2demo.demo.ChapterSeven$7.subscribe(ChapterSeven.java:169)
                                  at io.reactivex.internal.operators.flowable.FlowableCreate.subscribeActual(FlowableCreate.java:72)
                                  at io.reactivex.Flowable.subscribe(Flowable.java:12218)
                                  at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.run(FlowableSubscribeOn.java:82)
                                  at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:59)
                                  at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:51)
                                  at java.util.concurrent.FutureTask.run(FutureTask.java:237)
                                  at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:272)
                                  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1133)
                                  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:607)
                                  at java.lang.Thread.run(Thread.java:761)複製代碼

此次能夠看到, 在上游發送了第129個事件的時候, 就拋出了MissingBackpressureException異常, 提醒咱們發洪水啦. 固然了, 這個128也不是我憑空捏造出來的, Flowable的源碼中就有這個buffersize的大小定義, 能夠自行查看.

注意這裏咱們是把上游發送的事件所有都存進了水缸裏, 下游一個也沒有消費, 因此就溢出了, 若是下游去消費了事件, 可能就不會致使水缸溢出來了. 這裏咱們說的是可能不會, 這也很好理解, 好比剛纔這個例子上游發了129個事件, 下游只要快速的消費了一個事件, 就不會溢出了, 但若是下游過了十秒鐘再來消費一個, 那確定早就溢出了.

好了, 今天的教程就到這裏了, 下一節咱們將會更加深刻的去學習FLowable, 敬請期待.

(哈哈, 給個人RxDownload打個廣告: RxDownload是一個基於RxJava的多線程+斷點續傳的下載工具, 感興趣的來GitHub點個star吧☺. 電梯直達->戳這裏 )

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程