淺談RxJava與2.0的新特性

star 。甚至有一些大牛專門為 android 寫了 rxjava 的适配庫，如

<a target="_blank" href="https://github.com/reactivex/rxandroid">rxandroid</a>

<a target="_blank" href="https://github.com/jakewharton/rxbinding">rxbinding</a>

<a target="_blank" href="https://github.com/trello/rxlifecycle">rxlifecycle</a>

為什麼 rxjava 如此受到 android 開發者們的歡迎。我想不外乎兩個原因。 1. 異步 2. 鍊式操作

對 android 線程有所了解的朋友都知道， android的 ui 繪制 與 事件響應是在主線程的，為了保證界面的流暢性，很多耗時操作如讀寫資料庫、讀寫檔案、請求網絡，我們都會挪到異步線程去完成，再回調到主線程。當然在4.0以後主線程直接就不允許請求網絡了。

在過去沒有 rxjava 的時候，開發者一般都是通過 asynctask ， thread ，更好些的就是通過線程池來完成這些任務。而有了 rxjava 以後，簡簡單單的一句話就可以随意的切換線程，簡直不用太舒服。

最典型的 rxjava 中的 <code>observable</code> 類，提供了2個函數， 分别是 <code>subscribeon</code> 與<code>observeon</code> 。前者可以切換被觀察時的線程（如果說資料發射的線程不夠嚴謹，資料并非一定在觀察時發射的，尤其是開發者自定義 <code>onsubscribe</code> 時），後者可以切換資料被消費時的線程。

舉一個切換線程的例子:

這裡我們沒有任何資料，就僅僅發射了一個 <code>oncomplete</code> ，但是在切換線程的代碼中，我們增加了 <code>oncomplte</code> 時要額外執行的代碼，輸出結果如下:

這僅僅是簡單的例子， rxjava 提供了很多便捷的操作符供我們使用，如 <code>map</code> 、 <code>filter</code> 、 <code>flatmap</code> 、 <code>merge</code> 、 <code>concat</code> 等。可見當熟練使用後對我們的程式設計效率确實有很大幫助。尤其是

mvp 模式， rxjava 與之結合可謂是”天作之合”。

上面筆者示範的代碼其實就是 rxjava 的典型使用方式:

發射資料源

中間操作

處理結果

其中中間操作包含諸多用法， 如果切換線程，變換資料等。

為什麼我說鍊式操作很好。第一，鍊式邏輯替代深度回調邏輯，容易編寫，不易出 bug 。第二，rxjava 提供諸多了整體處理資料的操作符，非常實用。第三，配合 java8 的 lambda 表達式，使代碼簡短優雅。

前天， rxjava終于釋出了2.0 rc1 版本，一直關注于此的筆者立刻就進去嘗鮮了。結合官方的介紹，筆者總結并翻譯了一些與 1.x 的異同與大家分享。

首先要說的就是 rxjava 2和1是 互相獨立 的。是以包名與 maven 的依賴也是不一樣的，就類似于 okhttp 3與2一樣。 rxjava 2.x的依賴是全新的 <code>io.reactivex.rxjava2:rxjava:2.x.y</code> ，并且類處于該 <code>io.reactivex</code> 包名下，而不再是 <code>rx</code> 。

publisher

subscriber

subscription

processor

新的實作叫做 <code>flowable</code> ， 同時舊的 <code>observable</code> 也保留了。因為在

rxjava1.x 中，有很多事件不被能正确的背壓，進而抛出 <code>missingbackpressureexception</code> 。

舉個簡單的例子，在 rxjava1.x 中的 <code>observeon</code> ， 因為是切換了消費者的線程，是以内部實作用隊列存儲事件。在 android 中預設的 buffersize

大小是16，是以當消費比生産慢時， 隊列中的數目積累到超過16個，就會抛出 <code>missingbackpressureexception</code> ， 初學者很難明白為什麼會這樣，使得學習曲線異常得陡峭。

而在 2.0 中，observable 不再支援背壓，而flowable 支援非阻塞式的背壓。并且規範要求，所有的操作符強制支援背壓。幸運的是， flowable 中的操作符大多與舊有的 observable 類似。

single 與 completable 都基于新的 reactive streams 的思想重新設計了接口，主要是消費者的接口， 現在他們是這樣的：

對比一下 subscriber :

我們會發現和以前不一樣的是多了一個 <code>onsubscribe</code> 的方法， <code>subscription</code> 如下:

熟悉 rxjava 1.x 的朋友能發現， 新的 <code>subscription</code> 更像是綜合了舊的 <code>producer</code>與 <code>subscription</code> 的綜合體。他既可以向上遊請求資料，又可以打斷并釋放資源。而舊的 <code>subscription</code> 在這裡因為名字被占，而被重新命名成了 <code>disposable</code>

這裡最大的不同就是這個 <code>onsubscribe</code> ，根據 specification， 這個函數一定是第一個被調用的， 然後就會傳給調用方一個 <code>subscription</code> ，通過這種方式組織新的背壓關系。當我們消費資料時，可以通過 <code>subscription</code> 對象，自己決定請求資料。

這裡就可以解釋上面的非阻塞的背壓。舊的阻塞式的背壓，就是根據下遊的消費速度，中遊可以選擇阻塞住等待下遊的消費，随後向上遊請求資料。而新的非阻塞就不在有中間阻塞的過程，由下遊自己決定取多少，還有背壓政策，如抛棄最新、抛棄最舊、緩存、抛異常等。

而新的接口帶來的新的調用方式與舊的也不太一樣， <code>subscribe</code> 後不再會有 subscription 也就是如今的 disposable，為了保持向後的相容，

flowable 提供了 <code>subscribewith方法</code> 傳回目前的 <code>subscriber</code> 對象，

并且同時提供了 <code>defaultsubscriber</code> , <code>resourcesubscriber</code> ， <code>disposablesubscriber</code> ，讓他們提供了 <code>disposable</code> 接口，

可以完成和以前類似的代碼:

在rxjava 1.x 最明顯的問題就是由于 create 的太過開放，導緻其被開發者濫用，而不是學習使用提供的操作符。

并且使用者對 rxjava 不夠了解，導緻各種各樣的問題，如背壓、異常處理等。

由于規範要求所有的操作符強制支援背壓，是以新的 create 采用了保守的設計，讓使用者實作 <code>flowableonsubscribe</code> 接口，并選取背壓政策，然後在内部實作封裝支援背壓，簡單的例子如下:

新的 <code>actionx</code> 、 <code>functionx</code> 的方法聲明都增加了一個 <code>throws exception</code> ，這帶來了顯而易見的好處，現在我們可以這樣寫:

而在以前是不行的， 因為 <code>files.readlines(name)</code> 會顯式的抛出一個 <code>ioexception</code>。這樣對

lambda 更加友好，而不必再去 try catch 。

在以前是必須要先 <code>createworker</code> ，用 worker 對象去 shedule， 現在可以直接在 <code>scheduler</code> 用這些方法：

這算是一個小優化，友善開發者使用。

如 <code>connectableobservable</code> 、 <code>blockobservable</code> 等，這樣可以直接在 <code>flowable</code> 中寫出這樣的代碼:

還有一些普通開發者不太在意的修改：

hook方式變化，現在可以通過提供接口在 runtime hook

部分在 1.x 中 被标記 <code>@beta</code> 、 <code>@experimental</code> 的操作符現在合并到正式版裡了

由于類結構的變動，一些類名的變化

等其他變動。

rxjava 作為開源的經典之作，筆者一直都有所關注。後續筆者會繼續為大家帶來 rxjava 的源碼解析與進階使用系列等。感謝大家的閱讀，如有不知之處，歡迎讨論交流。