RxJava 2.0 實用操作符總結及原理簡析

大概從2015年開始，RxJava1.0開始快速流行起來，短短兩年時間，RxJava在Android開發中已經算是無人不知無人不曉了，加之它與Retrofit等流行框架的完美結合，已經成為Android項目開發的必備利器。隨手記作為一個大型項目，引入三方框架一直比較慎重，但也從今年初開始，正式引入了RxJava2.0，并配合Retrofit對項目的網絡框架和繁瑣的異步邏輯進行重構。RxJava雖然好用，但伴隨而來的是不可避免的學習成本，為了讓大家快速的了解RxJava的來龍去脈以及快速上手使用，特地總結該篇文章。本文將詳細講解如何快速理解RxJava的操作符，并從源碼角度來分析RxJava操作符的原理。

RxJava的優點

簡單來講RxJava是一個簡化異步調用的庫，但其實它更是一種優雅的編程方式和編程思想，當你熟悉RxJava的使用方式之后，會很容易愛上它。 我總結它的優點主要有兩個方面：

• 簡潔，免除傳統異步代碼邏輯中的callback hell

• 增加業務邏輯代碼的可讀性

關于第一點大家應該都會認同，關于第二點可能有人會有疑惑，因為很多人覺得RxJava大量不明所以的操作符會讓代碼的可讀性變得更差，其實產生這種印象恰恰就是因為沒有掌握RxJava操作符的使用和原理所導致的。 比如隨手記項目中綁定用戶QQ賬號的業務邏輯，這段邏輯的代碼涉及三個異步接口，兩個是QQ登錄SDK的，一個是隨手記后臺的，在使用RxJava重構前，這段代碼使用了3個AsyncTask，也就是三個嵌套的回調，代碼復雜，可讀性非常差。而改造之后，它變成了下面這樣子

如果你對這里面的幾個RxJava操作符比較熟悉的話，你會迅速了解我這段代碼做了什么事情，而且不用再去梳理一堆嵌套回調了，這就是RxJava帶來的可讀性。 所以，學習RxJava，理解和掌握操作符是不可避免的第一步。

RxJava2.0與RxJava1.0的關系

從RxJava1.0到RxJava2.0，基本思想沒有變化，但RxJava2.0按照Reactive-Streams規范對整個架構進行了重新設計，并變更了Maven倉庫依賴地址和包名。所以現在RxJava的github網站中，RxJava1.0和RxJava2.0是兩個獨立的分支，不相互兼容 ，也不能同時使用，而且RxJava1.0再過一段時間也將不再維護。所以，目前還使用RxJava1.0的，建議盡早切換到RxJava2.0，而如果沒有接觸過RxJava1.0，直接使用和學習RxJava2.0就可以了。如果想了解RxJava1.0和RxJava2.0的詳細區別，請參考官方文檔。 為行文方便，從此處開始，本文使用Rx來表示RxJava2.x。

Rx的操作符有哪些

剛接觸Rx的人面對一堆各式各樣的操作符會覺得不知如何去學習記憶，其實你只需要從整體上了解Rx操作符的類別和掌握一些使用頻率較高的操作符就足夠了，至于其他的操作符，你只需要知道它的使用場景和掌握如何快速理解一個操作符的方法，就可以在需要的時候快速拿來用了。 下圖是我根據官方文檔總結的Rx操作符的分類及每個類別下的代表性操作符 從上圖可以看出，Rx的操作符主要十個大類別，每個類別下常用的操作符也就三五個左右，所以只要掌握這些，你就可以應付大部分的業務場景了。

如何快速理解一個Rx操作符

提到Rx操作符，相信很多人都會對描述Rx操作符的花花綠綠的寶石圖有很大印象。 要快速理解Rx操作符，看懂寶石圖是個快捷有效的方式，現在我們就來詳細分析一下構成寶石圖的各個主要元素。 首先，我們有必要回顧一下Rx中的幾個主要的基類

• io.reactivex.Flowable : 事件源(0..N個元素), 支持 Reactive-Streams and 背壓

• io.reactivex.Observable :事件源(0..N個元素), 不支持背壓

• io.reactivex.Single : 僅發射一個元素或產生error的事件源，

• io.reactivex.Completable : 不發射任何元素，只產生completion或error的事件源

• io.reactivex.Maybe : 不發射任何元素，或只發射一個元素，或產生error的事件源

• Subject : 既是事件源，也是事件接受者 可以看到Rx中最重要的概念就是  事件源 了，基本上所有的操作符都是針對事件源來進行一些轉換、組合等操作，而我們最常用的事件源就是  Observable 了。

本文中我們就以 Observable 事件源為例來講解Rx的操作符， Observable 發射的事件我們統一稱之為item。首先我們需要詳細了解一下寶石圖中各個圖像元素的含義：

• —> ：  Observable 的時間線，從左至右流動

• ★ ：星星、圓、方塊等表示  Observable 發射的item

• | ：時間線最后的小豎線表示  Observable 的事件流已經成功發射完畢了

• X ：時間線最后的X符合表示由于某種原因  Observable 非正常終止發射，產生了error

上面幾種元素組合在一起代表一個完整的 Observable ，也可以稱為源 Observable

--> 方向朝下的虛線箭頭表示以及中間的長方框表示正在對上面的源 Observable 進行某種轉換。長方框里的文字展示了轉換的性質。下面的 Observable 是對上面的源 Observable 轉換后的結果。

掌握了寶石圖的含義，我們就可以根據某個操作符的寶石圖快速理解這個操作符了。舉幾個例子：

1. map

可以看到，這幅圖表達的意思是一個源 Observable 先后發射了1、2、3的三個item，而經過 map 操作符一轉換，就變成了一個發射了10、20、30三個item的新的 Observable 。描述操作符的長方框中也清楚的說明了該 map 操作符進行了何種具體的轉換操作（圖中的10*x只是一個例子，這個具體的轉換函數是可以自定義的）。 于是，我們就很快速地理解了 map 操作符的含義和用法，簡單來講，它就是通過一個函數將一個 Observable 發射的item逐個進行某種轉換。 示例代碼：

輸出結果：

2. zip 根據 zip 的寶石圖，可以知道zip操作符的作用是把多個源 Observable 發射的item通過特定函數組合在一起，然后發射組合后的item。從圖中還可以看到一個重要的信息是，最終發射的item是對上面的兩個源 Observable 發射的item按照發射順序逐個組合的結果，而且最終發射的 1A 等item的發射時間是由組合它的 1 和 A 等item中發射時間較晚的那個item決定的，也正是如此， zip 操作符經常可以用在需要同時組合處理多個網絡請求的結果的業務場景中。 示例代碼：

輸出結果：

3. concat

從寶石圖可以看出， concat 操作符的作用就是將兩個源 Observable 發射的item連接在一起發射出來。這里的連接指的是整體連接，被 concat 操作后產生的 Observable 會先發射第一個源 Observable 的所有item，然后緊接著再發射第二個源 Observable 的所有的item。 示例代碼：

輸出結果：

大部分操作符都配有這樣的寶石圖，通過官方文檔或者直接在Rx源碼中查看JavaDoc就可以找到，不再過多舉例。你也可以在rxmarbles這樣的網站上查看更多可以動態交互的寶石圖。

Rx操作符的原理

要了解操作符的原理，肯定要從源碼入手嘍。所以我們先來簡單擼一遍Rx的最基本的Create操作符的源碼。 Rx的源碼目錄結構是比較清晰的，我們先從 Observable.create 方法來分析

create 方法如下

代碼很簡單，第一行判空不用管，第二行調用 RxJavaPlugins 的方法是為了實現Rx的hook功能，我們暫時也無需關注，在一般情況下，第二行代碼會直接返回它的入參即 ObservableCreate 對象， ObservableCreate 是 Observable 的子類，實現了 Observable 的一些抽象方法比如 subscribeActual 。事實上Rx的每個操作符都對應 Observable 的一個子類。 這里 create 方法接受的是一個 ObservableOnSubscribe 的接口實現類：

通過注釋可以知道這個接口的作用是通過一個 subscribe 方法接受一個 ObservableEmitter 類型的實例，俗稱發射器。 Observable.create 方法執行時，我們傳入的就是一個 ObservableOnSubscribe 類型的匿名內部類，并實現了它的 subscribe 方法，然后它又被傳入 create 方法的返回對象 ObservableCreate ，最終成為 ObservableCreate 的成員 source

接著我們來看 Observable 的 subscribe 方法，它的入參是一個 Observer (即觀察者，也就是事件接收者)

最終它會調用它的子類 ObservableCreate 的 subscribeActual 方法：

在 subscribeActual 里首先創建了用于發射事件的 CreateEmitter 對象 parent ， CreateEmitter 實現了接口 Emitter 和 Disposable ，并持有 observer 。 這段代碼的關鍵語句是 source.subscribe(parent) ，這行代碼執行后，就會觸發事件源進行發射事件，即 e.onNext("s") 會被調用。細心的同學也會注意到這行代碼之前， parent 先被傳入了 observer 的 onSubscribe() 方法，而在上面我們說過， observer 的 onSubscribe() 方法接受一個 Disposable 類型的參數，可以用于解除訂閱，之所以能夠解除訂閱，正是因為在觸發事件發射之前調用了 observer 的 onSubscribe() ，給了我們調用 CreateEmitter 的解除訂閱的方法 dispose() 的機會。 繼續來看 CreateEmitter 的 onNext() 方法，它最終是通過調用 observer 的 onNext() 方法將事件發射出去的

至此，Rx事件源的創建和訂閱的流程就走通了。

下面我們從 map 操作符來入手看一下Rx操作符的原理， map 方法如下

map 方法接受一個Function類型的參數 mapper ，返回了一個 ObservableMap 對象，它也是繼承自 Observable ，而 mapper 被傳給了 ObservableMap 的成員 function ,同時當前的源 Observable 被傳給 ObservableMap 的成員 source ，進入 ObservableMap 類

可以看到這里用到了裝飾者模式， ObservableMap 持有來自它上游的事件源 source ， MapObserver 持有來自它下游的事件接收者和我們實現的轉換方法 function ，在 subscribeActual() 方法中完成 ObservableMap 對 source 的訂閱，觸發 MapObserver 的 onNext() 方法，繼而將來自 source 的原始數據經過函數 mapper 轉換后再發射給下游的事件接收者，從而實現map這一功能。

現在我們終于能夠來總結一下包含多個操作符時的訂閱流程了，以下面這段代碼為例

執行代碼時，自上而下每一步操作符都會創建一個新的 Observable (均為 Observable 的子類，對應不同的操作符)，當執行 create 時，創建并返回了 ObservableCreate ，當執行 map 時，創建并返回了 ObservableMap ，并且每一個新的 Observable 都持有它上游的源 Observable (即 source )及當前涉及到的操作函數 function 。當最后一步執行訂閱方法 subscribe 時會觸發 ObservableMap 的 subscribeActual() 方法，并將最下游的 Observer 包裝成 MapObserver ，同時該方法又會繼續調用它所持有 ObservableCreate 的訂閱方法（即執行 source.subscribe ），由此也會觸發 ObservableCreate 的 subscribeActual() 方法，此時我們的發射器 CreateEmitter 才會調用它的 onNext() 方法發射事件，再依次調用 MapObserver 的操作函數 mapper 和 onNext() 方法，最終將事件傳遞給了最下游的 Observer 的 onNext() 方法。

我簡單的將這段邏輯用下面這幅圖來表示

操作符 lift 和  compose

lift 和 compose 在Rx中是兩個比較特殊的操作符。 lift 讓我們可以對 Observer 進行封裝，在RxJava1.0中大部分變換都基于 lift 這個神奇的操作符。

lift 操作符接受一個 ObservableOperator 對象

看注釋可以知道，這是一個將下游訂閱者包裝成一個上游訂閱者的接口。類似Map操作符中的MapObserver。

而 compose 操作符讓我們可以對 Observable 進行封裝

wrap 方法如下，僅僅是走了 RxJavaPlugins 的流程

compose 方法接受一個 ObservableTransformer 對象

ObservableSource 即為我們的基類 Observable 繼承的唯一接口。看注釋可以知道， ObservableTransformer 是一個組合多個 Observable 的接口，它通過一個 apply() 方法接收上游的 Observable ，進行一些操作后，返回新的 Observable 。 這里組合多個 Observable 的意思其實就是組合多個操作符，比如我們經常會需要在使用Rx進行網絡異步請求時進行線程變化，這個操作一般都是差不多的，每次都寫會比較煩，這時我們就可以使用 compose 把常用的線程變換的幾個操作符組合起來

關于 compose 的典型應用，大家有興趣還可以去看一下開源項目RxLifecycle，它就是巧妙地利用 compose 操作符來解決了使用Rx可能會出現的內存泄露問題。

Rx操作符的應用場景

說了這么多，其實我們最關心的還是Rx操作符的應用場景。其實只要存在異步的地方，都可以優雅地使用Rx操作符。比如很多流行的Rx周邊開源項目

而針對自己想要實現的功能情景，如何去選擇特定的操作符，官網的文檔中也列出了一些指導——Rx操作符決策樹。

當然除了這些，我們在開發項目時，還會有各種具體的業務場景需要選擇合適的操作符，這里我總結了一些經常遇到的場景以及適合它們的操作符

只要我們理解了Rx操作符的原理，熟練掌握了一些使用頻率較高的操作符，就能夠在以上場景中輕松地使用，不再讓自己的代碼被復雜的業務邏輯搞得混亂。

以上就是本文的全部內容，關于Rx還有很多東西值得深入地學習研究，后續有機會再跟大家分享更多Rx的使用心得。

參考

• RxJava2Examples

 

來自：http://mp.weixin.qq.com/s/OJCEyH1gJ-JiRIDC3LLi3w