Android拾萃 - RxJava５種響應(yīng)類型的區(qū)別和背壓（三）

Android拾萃 - RxJava最簡單的入門（一）
Android拾萃 - RxJava操作符列表和響應(yīng)類型（二）

本文首先會給出ReactiveX的核心操作符列表和對應(yīng)的文檔鏈接，后面會具體的介紹和實踐所有的操作符。

如果想實現(xiàn)自己的操作符，可以參考：實現(xiàn)自定義操作符

創(chuàng)建操作
用于創(chuàng)建Observable的操作符
Create
 — 通過調(diào)用觀察者的方法從頭創(chuàng)建一個Observable
Defer
 — 在觀察者訂閱之前不創(chuàng)建這個Observable，為每一個觀察者創(chuàng)建一個新的Observable
Empty/Never/Throw
 — 創(chuàng)建行為受限的特殊Observable
From
 — 將其它的對象或數(shù)據(jù)結(jié)構(gòu)轉(zhuǎn)換為Observable
Interval
 — 創(chuàng)建一個定時發(fā)射整數(shù)序列的Observable
Just
 — 將對象或者對象集合轉(zhuǎn)換為一個會發(fā)射這些對象的Observable
Range
 — 創(chuàng)建發(fā)射指定范圍的整數(shù)序列的Observable
Repeat
 — 創(chuàng)建重復(fù)發(fā)射特定的數(shù)據(jù)或數(shù)據(jù)序列的Observable
Start
 — 創(chuàng)建發(fā)射一個函數(shù)的返回值的Observable
Timer
 — 創(chuàng)建在一個指定的延遲之后發(fā)射單個數(shù)據(jù)的Observable

變換操作
這些操作符可用于對Observable發(fā)射的數(shù)據(jù)進行變換，詳細解釋可以看每個操作符的文檔
Buffer
 — 緩存，可以簡單的理解為緩存，它定期從Observable收集數(shù)據(jù)到一個集合，然后把這些數(shù)據(jù)集合打包發(fā)射，而不是一次發(fā)射一個
FlatMap
 — 扁平映射，將Observable發(fā)射的數(shù)據(jù)變換為Observables集合，然后將這些Observable發(fā)射的數(shù)據(jù)平坦化的放進一個單獨的Observable，可以認為是一個將嵌套的數(shù)據(jù)結(jié)構(gòu)展開的過程。
GroupBy
 — 分組，將原來的Observable分拆為Observable集合，將原始Observable發(fā)射的數(shù)據(jù)按Key分組，每一個Observable發(fā)射一組不同的數(shù)據(jù)
Map
 — 映射，通過對序列的每一項都應(yīng)用一個函數(shù)變換Observable發(fā)射的數(shù)據(jù)，實質(zhì)是對序列中的每一項執(zhí)行一個函數(shù)，函數(shù)的參數(shù)就是這個數(shù)據(jù)項
Scan
 — 掃描，對Observable發(fā)射的每一項數(shù)據(jù)應(yīng)用一個函數(shù)，然后按順序依次發(fā)射這些值
Window
 — 窗口，定期將來自O(shè)bservable的數(shù)據(jù)分拆成一些Observable窗口，然后發(fā)射這些窗口，而不是每次發(fā)射一項。類似于Buffer，但Buffer發(fā)射的是數(shù)據(jù)，Window發(fā)射的是Observable，每一個Observable發(fā)射原始Observable的數(shù)據(jù)的一個子集

過濾操作
這些操作符用于從Observable發(fā)射的數(shù)據(jù)中進行選擇
Debounce
 — 只有在空閑了一段時間后才發(fā)射數(shù)據(jù)，通俗的說，就是如果一段時間沒有操作，就執(zhí)行一次操作
Distinct
 — 去重，過濾掉重復(fù)數(shù)據(jù)項
ElementAt
 — 取值，取特定位置的數(shù)據(jù)項
Filter
 — 過濾，過濾掉沒有通過謂詞測試的數(shù)據(jù)項，只發(fā)射通過測試的
First
 — 首項，只發(fā)射滿足條件的第一條數(shù)據(jù)
IgnoreElements
 — 忽略所有的數(shù)據(jù)，只保留終止通知(onError或onCompleted)
Last
 — 末項，只發(fā)射最后一條數(shù)據(jù)
Sample
 — 取樣，定期發(fā)射最新的數(shù)據(jù)，等于是數(shù)據(jù)抽樣，有的實現(xiàn)里叫ThrottleFirst
Skip
 — 跳過前面的若干項數(shù)據(jù)
SkipLast
 — 跳過后面的若干項數(shù)據(jù)
Take
 — 只保留前面的若干項數(shù)據(jù)
TakeLast
 — 只保留后面的若干項數(shù)據(jù)

組合操作
組合操作符用于將多個Observable組合成一個單一的Observable
And/Then/When
 — 通過模式(And條件)和計劃(Then次序)組合兩個或多個Observable發(fā)射的數(shù)據(jù)集
CombineLatest
 — 當(dāng)兩個Observables中的任何一個發(fā)射了一個數(shù)據(jù)時，通過一個指定的函數(shù)組合每個Observable發(fā)射的最新數(shù)據(jù)（一共兩個數(shù)據(jù)），然后發(fā)射這個函數(shù)的結(jié)果
Join
 — 無論何時，如果一個Observable發(fā)射了一個數(shù)據(jù)項，只要在另一個Observable發(fā)射的數(shù)據(jù)項定義的時間窗口內(nèi)，就將兩個Observable發(fā)射的數(shù)據(jù)合并發(fā)射
Merge
 — 將兩個Observable發(fā)射的數(shù)據(jù)組合并成一個
StartWith
 — 在發(fā)射原來的Observable的數(shù)據(jù)序列之前，先發(fā)射一個指定的數(shù)據(jù)序列或數(shù)據(jù)項
Switch
 — 將一個發(fā)射Observable序列的Observable轉(zhuǎn)換為這樣一個Observable：它逐個發(fā)射那些Observable最近發(fā)射的數(shù)據(jù)
Zip
 — 打包，使用一個指定的函數(shù)將多個Observable發(fā)射的數(shù)據(jù)組合在一起，然后將這個函數(shù)的結(jié)果作為單項數(shù)據(jù)發(fā)射

錯誤處理
這些操作符用于從錯誤通知中恢復(fù)
Catch
 — 捕獲，繼續(xù)序列操作，將錯誤替換為正常的數(shù)據(jù)，從onError通知中恢復(fù)
Retry
 — 重試，如果Observable發(fā)射了一個錯誤通知，重新訂閱它，期待它正常終止

輔助操作
一組用于處理Observable的操作符
Delay
 — 延遲一段時間發(fā)射結(jié)果數(shù)據(jù)
Do
 — 注冊一個動作占用一些Observable的生命周期事件，相當(dāng)于Mock某個操作
Materialize/Dematerialize
 — 將發(fā)射的數(shù)據(jù)和通知都當(dāng)做數(shù)據(jù)發(fā)射，或者反過來
ObserveOn
 — 指定觀察者觀察Observable的調(diào)度程序（工作線程）
Serialize
 — 強制Observable按次序發(fā)射數(shù)據(jù)并且功能是有效的
Subscribe
 — 收到Observable發(fā)射的數(shù)據(jù)和通知后執(zhí)行的操作
SubscribeOn
 — 指定Observable應(yīng)該在哪個調(diào)度程序上執(zhí)行
TimeInterval
 — 將一個Observable轉(zhuǎn)換為發(fā)射兩個數(shù)據(jù)之間所耗費時間的Observable
Timeout
 — 添加超時機制，如果過了指定的一段時間沒有發(fā)射數(shù)據(jù)，就發(fā)射一個錯誤通知
Timestamp
 — 給Observable發(fā)射的每個數(shù)據(jù)項添加一個時間戳
Using
 — 創(chuàng)建一個只在Observable的生命周期內(nèi)存在的一次性資源

條件和布爾操作v
這些操作符可用于單個或多個數(shù)據(jù)項，也可用于Observable
All
 — 判斷Observable發(fā)射的所有的數(shù)據(jù)項是否都滿足某個條件
Amb
 — 給定多個Observable，只讓第一個發(fā)射數(shù)據(jù)的Observable發(fā)射全部數(shù)據(jù)
Contains
 — 判斷Observable是否會發(fā)射一個指定的數(shù)據(jù)項
DefaultIfEmpty
 — 發(fā)射來自原始Observable的數(shù)據(jù)，如果原始Observable沒有發(fā)射數(shù)據(jù)，就發(fā)射一個默認數(shù)據(jù)
SequenceEqual
 — 判斷兩個Observable是否按相同的數(shù)據(jù)序列
SkipUntil
 — 丟棄原始Observable發(fā)射的數(shù)據(jù)，直到第二個Observable發(fā)射了一個數(shù)據(jù)，然后發(fā)射原始Observable的剩余數(shù)據(jù)
SkipWhile
 — 丟棄原始Observable發(fā)射的數(shù)據(jù)，直到一個特定的條件為假，然后發(fā)射原始Observable剩余的數(shù)據(jù)
TakeUntil
 — 發(fā)射來自原始Observable的數(shù)據(jù)，直到第二個Observable發(fā)射了一個數(shù)據(jù)或一個通知
TakeWhile
 — 發(fā)射原始Observable的數(shù)據(jù)，直到一個特定的條件為真，然后跳過剩余的數(shù)據(jù)

算術(shù)和聚合操作
這些操作符可用于整個數(shù)據(jù)序列
Average
 — 計算Observable發(fā)射的數(shù)據(jù)序列的平均值，然后發(fā)射這個結(jié)果
Concat
 — 不交錯的連接多個Observable的數(shù)據(jù)
Count
 — 計算Observable發(fā)射的數(shù)據(jù)個數(shù)，然后發(fā)射這個結(jié)果
Max
 — 計算并發(fā)射數(shù)據(jù)序列的最大值
Min
 — 計算并發(fā)射數(shù)據(jù)序列的最小值
Reduce
 — 按順序?qū)?shù)據(jù)序列的每一個應(yīng)用某個函數(shù)，然后返回這個值
Sum
 — 計算并發(fā)射數(shù)據(jù)序列的和

v連接操作
一些有精確可控的訂閱行為的特殊Observable
Connect
 — 指示一個可連接的Observable開始發(fā)射數(shù)據(jù)給訂閱者
Publish
 — 將一個普通的Observable轉(zhuǎn)換為可連接的
RefCount
 — 使一個可連接的Observable表現(xiàn)得像一個普通的Observable
Replay
 — 確保所有的觀察者收到同樣的數(shù)據(jù)序列，即使他們在Observable開始發(fā)射數(shù)據(jù)之后才訂閱

轉(zhuǎn)換操作
To
 — 將Observable轉(zhuǎn)換為其它的對象或數(shù)據(jù)結(jié)構(gòu)
Blocking
 阻塞Observable的操作符

操作符決策樹
幾種主要的需求
直接創(chuàng)建一個Observable（創(chuàng)建操作）
組合多個Observable（組合操作）
對Observable發(fā)射的數(shù)據(jù)執(zhí)行變換操作（變換操作）
從Observable發(fā)射的數(shù)據(jù)中取特定的值（過濾操作）
轉(zhuǎn)發(fā)Observable的部分值（條件/布爾/過濾操作）
對Observable發(fā)射的數(shù)據(jù)序列求值（算術(shù)/聚合操作）

創(chuàng)建操作

建議在傳遞給create方法的函數(shù)中檢查觀察者的isUnsubscribed狀態(tài)，以便在沒有觀察者的時候，讓你的Observable停止發(fā)射數(shù)據(jù)或者做昂貴的運算。
在rxjava2中（本系列都是在rxjava2的基礎(chǔ)上進行論述的，rxjava1請自行查閱），create接收的類型都為xxxOnSubscrible(xxx為5種類型對應(yīng)的名字),發(fā)射器為xxxEmitter，具體如下表：

RxJava 2.x 類型 create參數(shù)（響應(yīng)接口） 發(fā)射器 Observer
Observable ObservableOnSubscribe ObservableEmitter Observer
Flowable FlowableOnSubscribe FlowableEmitter FlowableSubscriber
Single SingleOnSubscribe SingleEmitter SingleObserver
Completable CompletableOnSubscribe CompletableEmitter CompletableObserver
Maybe MaybeOnSubscribe MaybeEmitter MaybeObserver

再次借用Android拾萃 - RxJava操作符和響應(yīng)類型（二） 的圖片。

RxJava 2.x的5種響應(yīng)類型

下面來看下，五種響應(yīng)類型具體代碼．

1. Observable

Observable是最完整的，擁有四個回調(diào)方法onSubscribe會在收到數(shù)據(jù)之前調(diào)用，可以做一些動畫加載之類的操作．onNext收到數(shù)據(jù)會調(diào)用，onError和onComplete是互斥的，只會調(diào)用一個．

  //ObservableOnSubscribe接口
            Observable observable = Observable.create(new ObservableOnSubscribe() {
                @Override
                public void subscribe(@NonNull ObservableEmitter e) throws Exception {
                    //判斷下是否釋放了
                    if (!e.isDisposed()){
                        for (int i = 0; i<5; i++) {
                            e.onNext(i);
                        }
                        e.onComplete();
                    }
                }
            });
            
            //<Integer>泛型沒有添加的話，默認的是object ，然后onext強制轉(zhuǎn)換一下即可
            observable.subscribe(new Observer<Integer>() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {
                    
                }

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

                @Override
                public void onError(@NonNull Throwable e) {

                }

                @Override
                public void onComplete() {

                }
            });

Consumer是簡化版的Observer, 可以只關(guān)心某個步驟(onNext　onError　　onComplete),對應(yīng)的有一個參數(shù)，兩個參數(shù)，三個參數(shù)的方法如下

可以簡單理解微，Action和Consumer可以單獨行動，單獨存在
Action 對應(yīng)　　無參的onComplete()
Consumer　對應(yīng)　單個參數(shù)的onError(Throwable t)　或者　onNext(T t)

            observable.subscribe(new Consumer() {
                @Override
                public void accept(Object o) throws Exception {
                    //對應(yīng)onNext方法
                }
            }, new Consumer<Throwable>() {
                @Override
                public void accept(Throwable throwable) throws Exception {
                    //對應(yīng)onError
                }
            });

            observable.subscribe(new Consumer() {
                @Override
                public void accept(Object o) throws Exception {
                    //對應(yīng)onNext方法
                }
            }, new Consumer<Throwable>() {
                @Override
                public void accept(Throwable throwable) throws Exception {
                    //對應(yīng)onError
                }
            }, new Action() {
                @Override
                public void run() throws Exception {
                    //對應(yīng)onComplete　　無參返回
                }
            });

下面的幾種case, Consumer就不再討論了，具體參考上面的代碼

2. Flowable

Backpressure的策略

我們發(fā)現(xiàn) Flowable.create 多了一個參數(shù)，背壓策略

點進去源碼發(fā)現(xiàn)總共有五種策略

/**
 * Represents the options for applying backpressure to a source sequence.
 */
public enum BackpressureStrategy {
    /**
     * OnNext events are written without any buffering or dropping.
     * Downstream has to deal with any overflow.
     * <p>Useful when one applies one of the custom-parameter onBackpressureXXX operators.
     */
    MISSING,
    /**
     * Signals a MissingBackpressureException in case the downstream can't keep up.
     */
    ERROR,
    /**
     * Buffers <em>all</em> onNext values until the downstream consumes it.
     */
    BUFFER,
    /**
     * Drops the most recent onNext value if the downstream can't keep up.
     */
    DROP,
    /**
     * Keeps only the latest onNext value, overwriting any previous value if the
     * downstream can't keep up.
     */
    LATEST
}

處理Backpressure的策略僅僅是處理Subscriber接收事件的方式，并不影響Flowable發(fā)送事件和事件產(chǎn)生速度。

我們知道只有生產(chǎn)者的速度大于消費者的速度，才會產(chǎn)生Backpressure問題。也就是說只會發(fā)生在異步的情況下。

Flowable的幾種背壓策略：

  1. BackpressureStrategy.ERROR：緩存區(qū)默人大小128，流速不均衡時發(fā)射MissingBackpressureException信號。
  2. BackpressureStrategy.BUFFER：緩存區(qū)不限制大小，使用不當(dāng)仍會OOM。
  3. BackpressureStrategy.DROP：緩存最近的nNext事件。
  4. BackpressureStrategy.LATEST：緩存區(qū)會保留最后的OnNext事件，覆蓋之前緩存的OnNext事件。
  5. BackpressureStrategy.MISSING：OnNext事件沒有任何緩存和丟棄，下游要處理任何溢出。

onBackpressureBuffer是不丟棄數(shù)據(jù)的處理方式。把上游收到的全部緩存下來，等下游來請求再發(fā)給下游。相當(dāng)于一個水庫。但上游太快，水庫（buffer）就會溢出。
onBackpressureDrop和onBackpressureLatest比較類似，都會丟棄數(shù)據(jù)。這兩種策略相當(dāng)于一種令牌機制（或者配額機制），下游通過request請求產(chǎn)生令牌（配額）給上游，上游接到多少令牌，就給下游發(fā)送多少數(shù)據(jù)。當(dāng)令牌數(shù)消耗到0的時候，上游開始丟棄數(shù)據(jù)。但這兩種策略在令牌數(shù)為0的時候有一點微妙的區(qū)別：onBackpressureDrop直接丟棄數(shù)據(jù)，不緩存任何數(shù)據(jù)；而onBackpressureLatest則緩存最新的一條數(shù)據(jù)，這樣當(dāng)上游接到新令牌的時候，它就先把緩存的上一條“最新”數(shù)據(jù)發(fā)送給下游。

 //我們發(fā)現(xiàn)Flowable 多了一個參數(shù)，背壓策略
            Flowable flowable = Flowable.create(new FlowableOnSubscribe<Object>() {
                @Override
                public void subscribe(@NonNull FlowableEmitter<Object> e) throws Exception {

  //判斷下是否釋放了
                    if (!e.isCancelled()){
                        for (int i = 0; i<5; i++) {
                            e.onNext(i);
                        }
                        e.onComplete();
                    }
                }
            }, BackpressureStrategy.DROP);

            //這里建立關(guān)系都是使用了subscribe，只不過對應(yīng)的方法不一樣，請參照上面的表
            flowable
           .subscribeOn(Schedulers.io())
           .observeOn(AndroidSchedulers.mainThread())

          .subscribe(new FlowableSubscriber() {
                @Override
                public void onSubscribe(@NonNull Subscription s) {

                }

                @Override
                public void onNext(Object o) {
                    Log.d(TAG, o + "");
                }

                @Override
                public void onError(Throwable t) {

                }

                @Override
                public void onComplete() {

                }
            });
3. Single

Single 是沒有complete的，因為，只發(fā)送一個數(shù)據(jù)之后，默認就會調(diào)用complete

Single single = Single.create(new SingleOnSubscribe<Object>() {
                @Override
                public void subscribe(@NonNull SingleEmitter<Object> e) throws Exception {
                    e.onSuccess(1);
                }
            });

            single.subscribeOn(Schedulers.io()).subscribe(new SingleObserver() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {

                }

                @Override
                public void onSuccess(@NonNull Object o) {
                    Log.d(TAG, o + "");
                }

                @Override
                public void onError(@NonNull Throwable e) {

                }
            });
4. Completable

Completable只關(guān)心onComplete，所以上游發(fā)射的數(shù)據(jù)，都不會進行處理的．

Completable completable =  Completable.create(new CompletableOnSubscribe() {
                @Override
                public void subscribe(@NonNull CompletableEmitter e) throws Exception {
                    e.onComplete();
                }
            });
            completable.subscribe(new CompletableObserver() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {
                    
                }

                @Override
                public void onComplete() {

                }

                @Override
                public void onError(@NonNull Throwable e) {

                }
            });
5. Maybe

Maybe是Single　和　Completable的合體，只會處理一條數(shù)據(jù)，但是onComplete不會自動調(diào)用

            Maybe maybe = Maybe.create(new MaybeOnSubscribe() {
                @Override
                public void subscribe(@NonNull MaybeEmitter e) throws Exception {
                    e.onSuccess(12);
                    e.onComplete();
                }
            });
            
            maybe.subscribe(new MaybeObserver() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {
                    
                }

                @Override
                public void onSuccess(@NonNull Object o) {
                    Log.d(TAG, o + "");
                }

                @Override
                public void onError(@NonNull Throwable e) {

                }

                @Override
                public void onComplete() {

                }
            });
最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

  • 響應(yīng)式編程簡介 響應(yīng)式編程是一種基于異步數(shù)據(jù)流概念的編程模式。數(shù)據(jù)流就像一條河：它可以被觀測，被過濾，被操作，或者...
    說碼解字閱讀 3,573評論 0贊 5
  • 記錄RxJava操作符，方便查詢（2.2.2版本） 英文文檔地址：http://reactivex.io/docu...
    凌云飛魚閱讀 1,055評論 0贊 0
  • 一、RxJava操作符概述 RxJava中的操作符就是為了提供函數(shù)式的特性，函數(shù)式最大的好處就是使得數(shù)據(jù)處理簡潔易...
    無求_95dd閱讀 3,513評論 0贊 21
  • 1. 響應(yīng)式編程 1.1 響應(yīng)式編程概念 響應(yīng)式編程是一種通過異步和數(shù)據(jù)流來構(gòu)建事物關(guān)系的編程模型。 事物的關(guān)系 ...
    師傅_有妖氣閱讀 1,494評論 0贊 0
  • 目錄 一：創(chuàng)建操作 二：合并操作 三：過濾操作 四：切換線程 五：條件/布爾操作 六：聚合操作 七：轉(zhuǎn)換操作 八：...
    Allens_Jiang閱讀 10,252評論 11贊 32

友情鏈接更多精彩內(nèi)容