Android RxJava：一文帶你全面瞭解 背壓策略

前言

• Rxjava，因爲其基於事件流的鏈式調用、邏輯簡潔 & 使用簡單的特色，深受各大 Android開發者的歡迎。

 

• 本文主要講解的是RxJava中的 背壓控制策略，但願大家會喜歡。

1.  
2.  

 

目錄

1. 引言

1.1 背景

• 觀察者 & 被觀察者 之間存在2種訂閱關係：同步 & 異步。具體以下：

• 對於異步訂閱關係，存在 被觀察者發送事件速度 與觀察者接收事件速度 不匹配的狀況 

  1.  
  2.  

 

1.2 問題

• 被觀察者 發送事件速度太快，而觀察者 來不及接收全部事件，從而致使觀察者沒法及時響應 / 處理全部發送過來事件的問題，最終致使緩存區溢出、事件丟失 & OOM 

  1.  
  2.  

下面再舉個例子： 

• 被觀察者的發送事件速度 = 10ms / 個 
• 觀察者的接收事件速度 = 5s / 個 

即出現發送 & 接收事件嚴重不匹配的問題

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48

• 結果 
  因爲被觀察者發送事件速度 > 觀察者接收事件速度，因此出現流速不匹配問題，從而致使OOM 
  

1.3 解決方案

採用 背壓策略。

下面，我將開始介紹背壓策略。

2. 背壓策略簡介

2.1 定義

一種 控制事件流速 的策略

2.2 做用

在 異步訂閱關係 中，控制事件發送 & 接收的速度

 

2.3 解決的問題

解決了 因被觀察者發送事件速度 與 觀察者接收事件速度 不匹配（通常是前者 快於 後者），從而致使觀察者沒法及時響應 / 處理全部 被觀察者發送事件 的問題

2.4 應用場景

• 被觀察者發送事件速度 與 觀察者接收事件速度 不匹配的場景
• 具體場景就取決於 該事件的類型，如：網絡請求，那麼具體場景：有不少網絡請求須要執行，但執行者的執行速度沒那麼快，此時就須要使用背壓策略來進行控制。

3. 背壓策略的原理

• 那麼，RxJava實現背壓策略（Backpressure）的原理是什麼呢？
• 解決方案 & 思想主要以下：

• 示意圖以下

• 與 RxJava1.0 中被觀察者的舊實現 Observable 對比

• 好了，那麼上圖中在RxJava 2.0觀察者模型中，Flowable究竟是什麼呢？它實際上是RxJava 2.0中被觀察者的一種新實現，同時也是背壓策略實現的承載者
• 請繼續看下一節的介紹：背壓策略的具體實現 - Flowable

4. 背壓策略的具體實現：Flowable

在 RxJava2.0中，採用 Flowable 實現 背壓策略

 

4.1 Flowable 介紹

• 定義：在 RxJava2.0中，被觀察者（Observable）的一種新實現 

   

• 做用：實現 非阻塞式背壓 策略

4.2 Flowable 特色

• Flowable的特色 具體以下

• 下面再貼出一張RxJava2.0 與RxJava1.0的觀察者模型的對比圖 

   

4.3 與 RxJava1.0 中被觀察者的舊實現 Observable 的關係

• 具體以下圖

• 那麼，爲何要採用新實現Flowable實現背壓，而不採用舊的Observable呢？
• 主要緣由：舊實現Observable沒法很好解決背壓問題。

4.4 Flowable的基礎使用

• Flowable的基礎使用很是相似於 Observable
• 具體以下

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50

• 更加優雅的鏈式調用

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38

• 至此，Flowable的基礎使用講解完
• 關於更深層次的使用會結合 背壓策略的實現 來說解

5. 背壓策略的使用

• 在本節中，我將結合 背壓策略的原理 & Flowable的使用，爲你們介紹在RxJava 2.0 中該如何使用Flowable來實現背壓策略功能，即背壓策略的使用
• Flowable與Observable在功能上的區別主要是 多了背壓的功能
• 下面，我將順着第3節中講解背壓策略實現原理 & 解決方案（以下圖），來說解Flowable在背壓策略功能上的使用

 

5.1 控制 觀察者接收事件 的速度

5.1.1 異步訂閱狀況

• 簡介

• 具體原理圖

• 具體使用

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47

• 效果圖

• 有2個結論是須要你們注意的

下圖 = 當緩存區存滿時（128個事件）溢出報錯的原理圖

• 代碼演示1：觀察者不接收事件的狀況下，被觀察者繼續發送事件 & 存放到緩存區；再按需取出

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62

• 代碼演示2：觀察者不接收事件的狀況下，被觀察者繼續發送事件至超出緩存區大小（128）

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34

5.1.2 同步訂閱狀況

同步訂閱 & 異步訂閱 的區別在於： 
- 同步訂閱中，被觀察者 & 觀察者工做於同1線程 
- 同步訂閱關係中沒有緩存區

• 被觀察者在發送1個事件後，必須等待觀察者接收後，才能繼續發下1個事件

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50

因此，實際上並不會出現被觀察者發送事件速度 > 觀察者接收事件速度的狀況。但是，卻會出現被觀察者發送事件數量 > 觀察者接收事件數量的問題。

• 如：觀察者只能接受3個事件，但被觀察者卻發送了4個事件，因此出現了不匹配狀況

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52

因此，對於沒有緩存區概念的同步訂閱關係來講，單純採用控制觀察者的接收事件數量（響應式拉取）實際上就等於 「單相思」，雖然觀察者控制了要接收3個事件，但假設被觀察者須要發送4個事件，仍是會出現問題。

 

• 有1個特殊狀況須要注意

• 代碼演示

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53

在被觀察者發送第1個事件後, 就拋出MissingBackpressureException異常 & 觀察者沒有收到任何事件

5.2 控制 被觀察者發送事件 的速度

• 簡介

• FlowableEmitter類的requested()介紹

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14

• 每一個線程中的requested（）的返回值 = 該線程中的request（a）的a值

• 對應於同步 & 異步訂閱狀況 的原理圖

爲了方便你們理解該策略中的requested（）使用，該節會先講解同步訂閱狀況，再講解異步訂閱狀況

5.2.1 同步訂閱狀況

• 原理說明

即在同步訂閱狀況中，被觀察者 經過 FlowableEmitter.requested()得到了觀察者自身接收事件能力，從而根據該信息控制事件發送速度，從而達到了觀察者反向控制被觀察者的效果

• 具體使用 
  下面的例子 = 被觀察者根據觀察者自身接收事件能力（10個事件），從而僅發送10個事件

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41

• 特別注意 
  在同步訂閱狀況中使用FlowableEmitter.requested()時，有如下幾種使用特性須要注意的：

狀況1：可疊加性

• 即：觀察者可連續要求接收事件，被觀察者會進行疊加並一塊兒發送

• 1
• 2
• 3
• 4

• 代碼演示

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34

狀況2：實時更新性

• 即，每次發送事件後，emitter.requested()會實時更新觀察者能接受的事件 

  1.  
  2.  

 

• 1
• 2
• 3
• 4
• 5
• 6

• 代碼演示

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47

狀況3：異常

• 當FlowableEmitter.requested()減到0時，則表明觀察者已經不可接收事件
• 此時被觀察者若繼續發送事件，則會拋出MissingBackpressureException異常 

   

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44


額外

• 若觀察者沒有設置可接收事件數量，即無調用Subscription.request（）
• 那麼被觀察者默認觀察者可接收事件數量 = 0，即FlowableEmitter.requested()的返回值 = 0

5.2.2 異步訂閱狀況

• 原理說明

從上面能夠看出，因爲兩者處於不一樣線程，因此被觀察者 沒法經過 FlowableEmitter.requested()知道觀察者自身接收事件能力，即 被觀察者不能根據 觀察者自身接收事件的能力 控制發送事件的速度。具體請看下面例子

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35

而在異步訂閱關係中，反向控制的原理是：經過RxJava內部固定調用被觀察者線程中的request(n) 從而 反向控制被觀察者的發送事件速度

那麼該何時調用被觀察者線程中的request(n) & n 的值該是多少呢？請繼續往下看。

• 具體使用

關於RxJava內部調用request(n)（n = 12八、9六、0）的邏輯以下：

 

• 代碼演示

下面我將用一個例子來演示該原理的邏輯

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65

整個流程 & 測試結果 請看下圖

5.3 採用背壓策略模式：BackpressureStrategy

5.3.1 背壓模式介紹

在Flowable的使用中，會被要求傳入背壓模式參數

• 面向對象：針對緩存區
• 做用：當緩存區大小存滿、被觀察者仍然繼續發送下1個事件時，該如何處理的策略方式 

   

5.3.2 背壓模式類型

 下面我將對每種模式逐一說明。 **模式1：BackpressureStrategy.ERROR** - 問題：發送事件速度 ＞ 接收事件 速度，即流速不匹配

 

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36

 **模式2：BackpressureStrategy.MISSING** - 問題：發送事件速度 ＞ 接收事件 速度，即流速不匹配

 

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36

 **模式3：BackpressureStrategy.BUFFER**

• 問題：發送事件速度 ＞ 接收事件 速度，即流速不匹配 

   

• 處理方式：將緩存區大小設置成無限大 
  1. 即 被觀察者可無限發送事件 觀察者，但其實是存放在緩存區 
  2. 但要注意內存狀況，防止出現OOM

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36

能夠接收超過原先緩存區大小（128）的事件數量了  **模式4： BackpressureStrategy.DROP**

• 問題：發送事件速度 ＞ 接收事件 速度，即流速不匹配 

   

• 處理方式：超過緩存區大小（128）的事件丟棄 
  如發送了150個事件，僅保存第1 - 第128個事件，第129 -第150事件將被丟棄

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47

被觀察者一會兒發送了150個事件，點擊按鈕接收時觀察者接收了128個事件；再次點擊接收時卻沒法接受事件，這說明超過緩存區大小的事件被丟棄了。  **模式5：BackpressureStrategy.LATEST**

• 問題：發送事件速度 ＞ 接收事件 速度，即流速不匹配 

   

• 處理方式：只保存最新（最後）事件，超過緩存區大小（128）的事件丟棄 
  即若是發送了150個事件，緩存區裏會保存129個事件（第1-第128 + 第150事件）

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46

• 被觀察者一會兒發送了150個事件，點擊按鈕接收時觀察者接收了128個事件；
• 再次點擊接收時卻接收到1個事件（第150個事件），這說明超過緩存區大小的事件僅保留最後的事件（第150個事件）

5.3.3 特別注意

在使用背壓策略模式的時候，有1種狀況是須要注意的：

a. 背景 
FLowable 可經過本身建立（如上面例子），或經過其餘方式自動建立，如interval操做符

 

b. 衝突 
- 對於自身手動建立FLowable的狀況，可經過傳入背壓模式參數選擇背壓策略 
（即上面描述的）

• 但是對於自動建立FLowable，卻沒法手動傳入傳入背壓模式參數，那麼出現流速不匹配的狀況下，該如何選擇 背壓模式呢？

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34

c. 解決方案 
RxJava 2.0內部提供 封裝了背壓策略模式的方法 
- onBackpressureBuffer() 
- onBackpressureDrop() 
- onBackpressureLatest()

 

具體使用以下：

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30

從而很好地解決了發送事件 & 接收事件 速度不匹配的問題。

其他方法的做用相似於上面的說背壓模式參數，此處不做過多描述。

背壓策略模式小結

• 至此，對RxJava 2.0的背壓模式終於講解完畢
• 全部代碼Demo均存放在Carson_Ho的Github地址

6. 總結

• 本文主要對 Rxjava 的背壓模式知識進行講解

• 接下來的時間，我將持續推出 Android中 Rxjava 2.0 的一系列文章，包括原理、操做符、應用場景、背壓等等 ，有興趣能夠繼續關注Carson_Ho的安卓開發筆記！！