對RxJava中Backpressure概念的理解

英文原文地址請點擊這裏

一. Backpressure（背壓、反壓力）

在rxjava中會常常遇到一種狀況就是被觀察者發送消息太快以致於它的操做符或者訂閱者不能及時處理相關的消息。那麼隨之而來的就是如何處理這些未處理的消息。

舉個例子，使用 zip 操做符將兩個無限大的Observable壓縮在一塊兒，其中一個被觀察者發送消息的速度是另外一個的兩倍。一個比較不靠譜的作法就是把發送比較快的消息緩存起來，當比較慢的Observable發送消息的時候取出來並將他們結合在一塊兒。這樣作就使得rxjava變得笨重並且十分佔用系統資源。

在rxjava中有多重控制流以及背壓（backpressure）策略用來應對當一個快速發送消息的被觀察者遇到一個處理消息緩慢的觀察者。下面的解釋將會向你展現你應當怎麼設計屬於你本身的被觀察者和操做符去應對流量控制（flow control）。

二. Hot and cold Observables, and multicasted Observables

Observable 數據流有兩種類型：hot 和 cold。這兩種類型有很大的不一樣。本節介紹他們的區別，以及做爲開發者應該如何正確的使用他們。

1. Cold observables

只有當有訂閱者訂閱的時候， Cold Observable 纔開始執行發射數據流的代碼。而且每一個訂閱者訂閱的時候都獨立的執行一遍數據流代碼。 Observable.interval 就是一個 Cold Observable。每個訂閱者都會獨立的收到他們的數據流。

咱們常常用到的Observable.create 就是 Cold Observable，而 just, range, timer 和 from 這些建立的一樣是 Cold Observable。

2. Hot observables

Hot observable 無論有沒有訂閱者訂閱，他們建立後就開發發射數據流。 一個比較好的示例就是鼠標事件。 無論系統有沒有訂閱者監聽鼠標事件，鼠標事件一直在發生，當有訂閱者訂閱後，從訂閱後的事件開始發送給這個訂閱者，以前的事件這個訂閱者是接受不到的；若是訂閱者取消訂閱了，鼠標事件依然繼續發射。

瞭解更多Hot and cold Observables，參考這裏

當一個cold observable是multicast（多路廣播）（當轉換完成時或者方法被調用）的時候，爲了應對背壓，應當把cold observable轉換成hot observable。

cold observable 至關於響應式拉（就是observer處理完了一個事件就從observable拉取下一個事件），hot observable一般不能很好的處理響應式拉模型，但它倒是處理流量控制問題的不二候選人，例如使用onBackpressureBuffer或者onBackpressureDrop 操做符，和其餘操做符比operations, throttling, buffers, or windows。

三. 能從某種意義上實現背壓的運算符

防止過分建立observable的第一道防線就是使用普通數組去減小observable發送消息的數量，在這一節會使用一些操做符去應對突發的observable發送爆發性數據（一會沒有，一會不少）就像下面的這張圖片所示：

 

            

 

這些操做符能夠經過微調參數確保slow-consuming觀察者不被生產可觀測的。

1. Throttling節流

操做符中好比 sample( ) 、 throttleLast( )、 throttleFirst( )、 throttleWithTimeout( ) 、 debounce( ) 容許你經過調節速率來改變Observable發射消息的速度。

如下圖表展現如何使用這些操做符。

(1). sample (或 throttleLast)

sample 操做符按期收集observable發送的數據items，併發射出最後一個數據item。 

           

Observable<Integer> burstySampled = bursty.sample(500, TimeUnit.MILLISECONDS);

 

上面代碼解釋，按期且一次收集5個item，發射出最後一個item。

(2). throttleFirst

跟sample有點相似，可是並非把觀測到的最後一個item發送出去，而是把該時間段第一個item發送出去。

 

            

Observable<Integer> burstyThrottled = bursty.throttleFirst(500, TimeUnit.MILLISECONDS);

 

(3). debounce (or throttleWithTimeout)

debounce操做符會只發送兩個在規定間隔內的時間發送的序列的最後一個。

 

              

Observable<Integer> burstyDebounced = bursty.debounce(10, TimeUnit.MILLISECONDS);

 

(4). Buffers and windows 緩衝區和窗口

可使用操做符好比buffer( ) 或者window( ) 收集過分生成消息的Observable的數據項，而後發射出較少使用的數據。緩慢的消費者能夠決定是否處理每一個集合中的某一個特定的項目，或處理集合中的某種組合，或爲集合中的每一項預約計劃工做，這都要視狀況處理。

如下圖表展現如何使用這些操做符。

• buffer

你能夠按期關閉並釋放突發性的 Observable 緩衝區。

 

            

Observable<List<Integer>> burstyBuffered = bursty.buffer(500, TimeUnit.MILLISECONDS);

 

在突發期間你能夠獲得的想要的，並在緩衝區收集數據和最終在突發結束的時候釋放緩存。使用debounce操做符釋放緩存並關閉指示器buffer操做符。

此段超過本人翻譯水平，特提供原文以下，若有好的翻譯建議請提出。 
Or you could get fancy, and collect items in buffers during the bursty periods and emit them at the end of each burst, by using the debounce operator to emit a buffer closing indicator to the buffer operator:

             

// we have to multicast the original bursty Observable so we can use it
// both as our source and as the source for our buffer closing selector:
Observable<Integer> burstyMulticast = bursty.publish().refCount();
// burstyDebounced will be our buffer closing selector:
Observable<Integer> burstyDebounced = burstMulticast.debounce(10, TimeUnit.MILLISECONDS);
// and this, finally, is the Observable of buffers we're interested in:
Observable<List<Integer>> burstyBuffered = burstyMulticast.buffer(burstyDebounced);

• window

與buffer相似，在一個window轉換中容許你發送一個週期性的生成消息的Observable的數據項窗口：

 

             

Observable<Observable<Integer>> burstyWindowed = bursty.window(500, TimeUnit.MILLISECONDS);

當你每次從源被觀察者收集了特定數量的數據項後也能夠選擇從新發送一個新的window:

               

Observable<Observable<Integer>> burstyWindowed = bursty.window(5);

 

四. 使用線程阻塞

處理過快生產item的其餘策略就是使用線程阻塞，可是這麼作違背了響應式設計和非阻塞模型設計，可是它的確是一個可行的選擇。在rxJava中並無操做符能夠作到這一點。

若是observable發送消息，subscriber消耗消息都是在同一個線程這將很好的處理這個問題，可是你要知道，在rxJava中，不少時候生產者和消費者都不在同一個線程。

 

五. 如何創建「響應式拉動（reactive pull）」backpressure

當subscribe訂閱observable的時候能夠經過調用subscribe.request（n），n是你想要的observable發送出來的量。

當在onNext()方法裏處理完某個數據項（或一些數據項）後，你能從新調用 request()方法，通知Observable發射數據項。下面是個例子：

someObservable.subscribe(new Subscriber<t>() {
    @Override
    public void onStart() {
      request(1);
    }

    @Override
    public void onCompleted() {
      // gracefully handle sequence-complete
    }

    @Override
    public void onError(Throwable e) {
      // gracefully handle error
    }

    @Override
    public void onNext(t n) {
      // do something with the emitted item "n"
      // request another item:
      request(1);
    }
});

 

你能夠經過一個神奇數字request， request(Long.MAX_VALUE)，禁用反應拉背力和要求Observable按照本身的步伐發射數據。request(0)是一個合法的調用,但不會奏效。請求值小於零的請求會致使拋出一個異常。

1. Reactive pull backpressure isn’t magic

backpressure 不會使得過分生產的observable的問題消失，這只是提供了一種更好的解決問題的方法。 讓咱們更仔細的研究剛剛說到的zip操做符的問題。

這裏有兩個observable，a和b，b發射item比a更加的頻繁，當你想zip這兩個observable的時候，你須要把a發送出來的第n個和b發送出來的第n個對象處理，然而因爲b發送出來的速率更快，這時候b已經發送出了n+1~n+m個消息了，這時候你要想要把a的n+1~n+m個消息結合的話，就必須持有b已經發送出來的n+1~n+m消息，同時，這意味着緩存的數量在不斷的增加。

固然你能夠給b添加操做符throttling，可是這意味着你將丟失某些從b發送出來的項，你真正想要作的其實就是告訴b：「b你須要慢下來，可是你要保持你給個人數據是完整的」。

響應式拉（reective pull）模型能夠幫你作到這一點，subscriber從observable那裏拉取數據，這與一般狀況下從observable那裏推送數據這種模式相比造成鮮明的對比。

在rxJava中，zip操做符正是使用了這種技巧。它給每一個源observable維護了一個小的緩存池，當它的緩存池滿了之後，它將不會從源observable那裏拉取item。每當zip發送一個item的時候，他從它的緩存池裏面移除相應的項，並從源observable那裏拉取下一個項。

在rxJava中，不少操做符都使用了這種模式（響應式拉），可是有的操做符並無使用這種模式，由於他們也許執行的操做跟源observable處於相同的進程。在這種狀況下，因爲消耗事件會阻塞本進程，因此這一項的工做完成後，纔有機會收到下一項。還有另一種狀況，backpressure也是不適合的，由於他們有指定的其餘方式去處理流量控制，這些特殊的狀況在rxJava的javadocs裏面都會有詳細說明。

可是，observable a和b必須正確的響應request()方法，若是一個observable尚未被支持響應式拉（並非每一個observable都會支持），你能夠採起如下其中一種操做均可以達到backpressure的行爲：

(1). onBackpressurebuffer

爲全部從源observable發送出來的數據項維護一個緩存區，根據他們生成的request發送給下層流。

 

             

這個操做符還有一個實驗性的版本容許去設置這個緩存池的大小，但當緩存池滿了之後將會終止執行並拋出異常。

(2). onBackpressureDrop

終止發送來自源observable的事件，除非來自下層流的subscriber即將調用request（n）方法的時候，此時纔會發送足夠的數據項給以知足requst。

             

(3). onBackpressureBlock (實驗性的,   不支持RxJava 1.0)

阻塞源Observable正在操做的線程的線程直到某個Subscriber發出請求，而後只要有即將發出的請求就結束阻塞。

 

             

若是你不容許這些操做符操做用在一個不支持背壓的Observable上，而且 若是做爲Subscriber的你或者在你和Observable之間的一些操做符嘗試去應用響應式拉背壓，你將會在onError回調事件中遭遇 MissBackpresssureException的警告。

六. Flowable與Observable

最後，爲了你們更好的理解backpressure概念，這裏補充說一下Flowable。

Observable在RxJava2.0中新的實現叫作Flowable， 同時舊的Observable也保留了。由於在 RxJava1.x 中，有不少事件不被能正確的背壓，從而拋出MissingBackpressureException。

舉個簡單的例子，在 RxJava1.x 中的 observeOn， 由於是切換了消費者的線程，所以內部實現用隊列存儲事件。在 Android 中默認的 buffersize 大小是16，所以當消費比生產慢時， 隊列中的數目積累到超過16個，就會拋出MissingBackpressureException， 初學者很難明白爲何會這樣，使得學習曲線異常得陡峭。

而在2.0 中，Observable 再也不支持背壓，而Flowable 支持非阻塞式的背壓。Flowable是RxJava2.0中專門用於應對背壓（Backpressure）問題。所謂背壓，即生產者的速度大於消費者的速度帶來的問題，好比在Android中常見的點擊事件，點擊過快則常常會形成點擊兩次的效果。其中，Flowable默認隊列大小爲128。而且規範要求，全部的操做符強制支持背壓。幸運的是， Flowable 中的操做符大多與舊有的 Observable 相似。