给 Android 开发人员的 RxJava 具体解释
前言
我從去年開始使用 RxJava 。到如今一年多了。
今年加入了 Flipboard 后，看到 Flipboard 的 Android 項目也在使用 RxJava 。并且使用的場景越來越多 。
而近期這幾個月。我也發現國內越來越多的人開始提及 RxJava 。
有人說『RxJava 真是太好用了』，有人說『RxJava 真是太難用了』，另外很多其它的人表示：我真的百度了也谷歌了。但我還是想問： RxJava 究竟是什么？
鑒于 RxJava 眼下這種既火爆又神奇的現狀，而我又在一年的使用過程中對 RxJava 有了一些理解，我決定寫下這篇文章來對 RxJava 做一個相對詳細的、針對 Android 開發人員的介紹。
這篇文章的目的有兩個：1. 給對 RxJava 感興趣的人一些入門的指引2. 給正在使用 RxJava 但仍然心存疑惑的人一些更深入的解析
- RxJava 究竟是什么
- RxJava 好在哪
- API 介紹和原理簡析
- 1. 概念：擴展的觀察者模式
- 觀察者模式
- RxJava 的觀察者模式
- 2. 基本實現
- 1) 創建 Observer
- 2) 創建 Observable
- 3) Subscribe (訂閱)
- 4) 場景演示樣例
- a. 打印字符串數組
- b. 由 id 取得圖片并顯示
- 3. 線程控制 —— Scheduler (一)
- 1) Scheduler 的 API (一)
- 2) Scheduler 的原理 (一)
- 4. 變換
- 1) API
- 2) 變換的原理：lift()
- 3) compose: 對 Observable 總體的變換
- 5. 線程控制：Scheduler (二)
- 1) Scheduler 的 API (二)
- 2) Scheduler 的原理（二）
- 3) 延伸：doOnSubscribe()
- 1. 概念：擴展的觀察者模式
- RxJava 的適用場景和使用方式
- 1. 與 Retrofit 的結合
- 2. RxBinding
- 3. 各種異步操作
- 4. RxBus
- 最后
- 關于作者：
- 為什么寫這個？
在正文開始之前的最后。放上 GitHub 鏈接和引入依賴的 gradle 代碼： Github：
https://github.com/ReactiveX/RxJava
https://github.com/ReactiveX/RxAndroid
引入依賴：
compile 'io.reactivex:rxjava:1.0.14'
compile 'io.reactivex:rxandroid:1.0.1'
（版本號號是文章公布時的最新穩定版）
另外，感謝 RxJava 核心成員流火楓林的技術支持和內測讀者代碼家、鮑永章、drakeet、馬琳、有時放縱、程序亦非猿、大頭鬼、XZoomEye、席德雨、TCahead、Tiiime、Ailurus、宅學長、妖孽、大大大大大臣哥、NicodeLee的幫助，以及周伯通招聘的贊助。
RxJava 究竟是什么
一個詞：異步。
RxJava 在 GitHub 主頁上的自我介紹是 "a library for composing asynchronous and event-based programs using observable sequences for the Java VM"（一個在 Java VM 上使用可觀測的序列來組成異步的、基于事件的程序的庫）。這就是 RxJava ，概括得非常精準。
然而，對于剛開始學習的人來說，這太難看懂了。由于它是一個『總結』，而剛開始學習的人更須要一個『引言』。
事實上， RxJava 的本質能夠壓縮為異步這一個詞。
說到根上。它就是一個實現異步操作的庫，而別的定語都是基于這之上的。
RxJava 好在哪
換句話說，『相同是做異步，為什么人們用它，而不用現成的 AsyncTask / Handler / XXX / ... ？』
一個詞：簡潔。
異步操作非常關鍵的一點是程序的簡潔性，由于在調度過程比較復雜的情況下。異步代碼經常會既難寫也難被讀懂。 Android 創造的 AsyncTask 和Handler ，事實上都是為了讓異步代碼更加簡潔。RxJava 的優勢也是簡潔，但它的簡潔的與眾不同之處在于。隨著程序邏輯變得越來越復雜，它依舊能夠保持簡潔。
假設有這樣一個需求：界面上有一個自己定義的視圖 imageCollectorView 。它的作用是顯示多張圖片，并能使用 addImage(Bitmap) 方法來隨意添加顯示的圖片。如今須要程序將一個給出的文件夾數組 File[] folders 中每一個文件夾下的 png 圖片都載入出來并顯示在 imageCollectorView 中。須要注意的是，由于讀取圖片的這一過程較為耗時，須要放在后臺運行。而圖片的顯示則必須在 UI 線程運行。
經常使用的實現方式有多種，我這里貼出當中一種：
new Thread() {@Overridepublic void run() {super.run();for (File folder : folders) {File[] files = folder.listFiles();for (File file : files) {if (file.getName().endsWith(".png")) {final Bitmap bitmap = getBitmapFromFile(file);getActivity().runOnUiThread(new Runnable() {@Overridepublic void run() {imageCollectorView.addImage(bitmap);}});}}}} }.start();而假設使用 RxJava 。實現方式是這種：
Observable.from(folders).flatMap(new Func1<File, Observable<File>>() {@Overridepublic Observable<File> call(File file) {return Observable.from(file.listFiles());}}).filter(new Func1<File, Boolean>() {@Overridepublic Boolean call(File file) {return file.getName().endsWith(".png");}}).map(new Func1<File, Bitmap>() {@Overridepublic Bitmap call(File file) {return getBitmapFromFile(file);}}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Action1<Bitmap>() {@Overridepublic void call(Bitmap bitmap) {imageCollectorView.addImage(bitmap);}});那位說話了：『你這代碼明明變多了啊！
簡潔個毛啊！』大兄弟你消消氣。我說的是邏輯的簡潔。不是單純的代碼量少（邏輯簡潔才是提升讀寫代碼速度的必殺技對不？）。
觀察一下你會發現， RxJava 的這個實現，是一條從上到下的鏈式調用，沒有不論什么嵌套，這在邏輯的簡潔性上是具有優勢的。當需求變得復雜時，這種優勢將更加明顯（試想假設還要求僅僅選取前 10 張圖片，常規方式要怎么辦？假設有很多其它這樣那樣的要求呢？再試想，在這一大堆需求實現完兩個月之后須要改功能，當你翻回這里看到自己當初寫下的那一片迷之縮進，你能保證自己將迅速看懂。而不是對著代碼又一次捋一遍思路？）。
另外。假設你的 IDE 是 Android Studio 。事實上每次打開某個 Java 文件的時候，你會看到被自己主動 Lambda 化的預覽，這將讓你更加清晰地看到程序邏輯：
Observable.from(folders).flatMap((Func1) (folder) -> { Observable.from(file.listFiles()) }).filter((Func1) (file) -> { file.getName().endsWith(".png") }).map((Func1) (file) -> { getBitmapFromFile(file) }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe((Action1) (bitmap) -> { imageCollectorView.addImage(bitmap) });假設你習慣使用 Retrolambda ，你也能夠直接把代碼寫成上面這種簡潔的形式。
而假設你看到這里還不知道什么是 Retrolambda ，我不建議你如今就去學習它。原因有兩點：1. Lambda 是把雙刃劍，它讓你的代碼簡潔的同一時候。減少了代碼的可讀性，因此同一時候學習 RxJava 和 Retrolambda 可能會讓你忽略 RxJava 的一些技術細節。2. Retrolambda 是 Java 6/7 對 Lambda 表達式的非官方兼容方案，它的向后兼容性和穩定性是無法保障的，因此對于企業項目，使用 Retrolambda 是有風險的。所以。與非常多 RxJava 的推廣者不同，我并不推薦在學習 RxJava 的同一時候一起學習 Retrolambda。
事實上，我個人盡管非常贊賞 Retrolambda，但我從來不用它。
在Flipboard 的 Android 代碼中。有一段邏輯非常復雜。包括了多次內存操作、本地文件操作和網絡操作，對象分分合合。線程間相互配合相互等待，一會兒排成人字，一會兒排成一字。
假設使用常規的方法來實現，肯定是要寫得欲仙欲死，然而在使用 RxJava 的情況下。依舊僅僅是一條鏈式調用就完畢了。它非常長。但非常清晰。
所以， RxJava 好在哪？就好在簡潔，好在那把什么復雜邏輯都能穿成一條線的簡潔。
API 介紹和原理簡析
這個我就做不到一個詞說明了……由于這一節的主要內容就是一步步地說明 RxJava 究竟如何做到了異步。如何做到了簡潔。
1. 概念：擴展的觀察者模式
RxJava 的異步實現，是通過一種擴展的觀察者模式來實現的。
觀察者模式
先簡述一下觀察者模式，已經熟悉的能夠跳過這一段。
觀察者模式面向的需求是：A 對象（觀察者）對 B 對象（被觀察者）的某種變化高度敏感，須要在 B 變化的一瞬間做出反應。舉個樣例。新聞里喜聞樂見的警察抓小偷。警察須要在小偷伸手作案的時候實施抓捕。
在這個樣例里，警察是觀察者，小偷是被觀察者，警察須要時刻盯著小偷的一舉一動，才干保證不會漏過不論什么瞬間。
程序的觀察者模式和這種真正的『觀察』略有不同，觀察者不須要時刻盯著被觀察者（比如 A 不須要每過 2ms 就檢查一次 B 的狀態），而是採用注冊(Register)或者稱為訂閱(Subscribe)的方式，告訴被觀察者：我須要你的某某狀態，你要在它變化的時候通知我。
Android 開發中一個比較典型的樣例是點擊監聽器 OnClickListener 。對設置 OnClickListener 來說。 View 是被觀察者， OnClickListener 是觀察者。二者通過 setOnClickListener() 方法達成訂閱關系。訂閱之后用戶點擊按鈕的瞬間。Android Framework 就會將點擊事件發送給已經注冊的 OnClickListener 。採取這樣被動的觀察方式，既省去了重復檢索狀態的資源消耗。也能夠得到最高的反饋速度。
當然，這也得益于我們能夠隨意定制自己程序中的觀察者和被觀察者，而警察叔叔明顯無法要求小偷『你在作案的時候務必通知我』。
OnClickListener 的模式大致例如以下圖：
如圖所看到的，通過 setOnClickListener() 方法，Button 持有 OnClickListener 的引用（這一過程沒有在圖上畫出）。當用戶點擊時，Button 自己主動調用 OnClickListener 的 onClick() 方法。
另外。假設把這張圖中的概念抽象出來（Button -> 被觀察者、OnClickListener -> 觀察者、setOnClickListener() -> 訂閱，onClick() -> 事件）。就由專用的觀察者模式（比如僅僅用于監聽控件點擊）轉變成了通用的觀察者模式。
例如以下圖：
而 RxJava 作為一個工具庫，使用的就是通用形式的觀察者模式。
RxJava 的觀察者模式
RxJava 有四個基本概念：Observable (可觀察者。即被觀察者)、 Observer (觀察者)、 subscribe (訂閱)、事件。
Observable 和 Observer 通過 subscribe() 方法實現訂閱關系。從而 Observable 能夠在須要的時候發出事件來通知 Observer。
與傳統觀察者模式不同， RxJava 的事件回調方法除了普通事件 onNext() （相當于 onClick() / onEvent()）之外。還定義了兩個特殊的事件：onCompleted() 和 onError()。
- onCompleted(): 事件隊列完結。
RxJava 不僅把每一個事件單獨處理，還會把它們看做一個隊列。RxJava 規定。當不會再有新的 onNext() 發出時，須要觸發 onCompleted() 方法作為標志。
- onError(): 事件隊列異常。在事件處理過程中出異常時，onError() 會被觸發，同一時候隊列自己主動終止。不同意再有事件發出。
- 在一個正確運行的事件序列中, onCompleted() 和 onError() 有且僅僅有一個。并且是事件序列中的最后一個。須要注意的是，onCompleted() 和 onError() 二者也是相互排斥的，即在隊列中調用了當中一個。就不應該再調用還有一個。
RxJava 的觀察者模式大致例如以下圖：
2. 基本實現
基于以上的概念。 RxJava 的基本實現主要有三點：
1) 創建 Observer
Observer 即觀察者，它決定事件觸發的時候將有如何的行為。 RxJava 中的 Observer 接口的實現方式：
Observer<String> observer = new Observer<String>() {@Overridepublic void onNext(String s) {Log.d(tag, "Item: " + s);}@Overridepublic void onCompleted() {Log.d(tag, "Completed!");}@Overridepublic void onError(Throwable e) {Log.d(tag, "Error!");} };除了 Observer 接口之外，RxJava 還內置了一個實現了 Observer 的抽象類：Subscriber。 Subscriber 對 Observer 接口進行了一些擴展。但他們的基本使用方式是全然一樣的：
Subscriber<String> subscriber = new Subscriber<String>() {@Overridepublic void onNext(String s) {Log.d(tag, "Item: " + s);}@Overridepublic void onCompleted() {Log.d(tag, "Completed!");}@Overridepublic void onError(Throwable e) {Log.d(tag, "Error!");} };不僅基本使用方式一樣，實質上，在 RxJava 的 subscribe 過程中，Observer 也總是會先被轉換成一個 Subscriber 再使用。
所以假設你僅僅想使用基本功能，選擇 Observer 和 Subscriber 是全然一樣的。它們的差別對于使用者來說主要有兩點：
它會在 subscribe 剛開始，而事件還未發送之前被調用，能夠用于做一些準備工作，比如數據的清零或重置。
這是一個可選方法。默認情況下它的實現為空。
須要注意的是，假設對準備工作的線程有要求（比如彈出一個顯示運行進度的對話框。這必須在主線程運行）， onStart() 就不適用了。由于它總是在 subscribe 所發生的線程被調用，而不能指定線程。要在指定的線程來做準備工作，能夠使用 doOnSubscribe() 方法，詳細能夠在后面的文中看到。
在這種方法被調用后。Subscriber 將不再接收事件。一般在這種方法調用前。能夠使用 isUnsubscribed() 先推斷一下狀態。
unsubscribe() 這種方法非常重要。由于在 subscribe() 之后。 Observable 會持有 Subscriber 的引用，這個引用假設不能及時被釋放，將有內存泄露的風險。所以最好保持一個原則：要在不再使用的時候盡快在合適的地方（比如 onPause() onStop() 等方法中）調用 unsubscribe() 來解除引用關系，以避免內存泄露的發生。
2) 創建 Observable
Observable 即被觀察者，它決定什么時候觸發事件以及觸發如何的事件。 RxJava 使用 create() 方法來創建一個 Observable 。并為它定義事件觸發規則：
Observable observable = Observable.create(new Observable.OnSubscribe<String>() {@Overridepublic void call(Subscriber<? super String> subscriber) {subscriber.onNext("Hello");subscriber.onNext("Hi");subscriber.onNext("Aloha");subscriber.onCompleted();} });能夠看到。這里傳入了一個 OnSubscribe 對象作為參數。
OnSubscribe 會被存儲在返回的 Observable 對象中，它的作用相當于一個計劃表，當 Observable 被訂閱的時候。OnSubscribe 的 call() 方法會自己主動被調用，事件序列就會按照設定依次觸發（對于上面的代碼，就是觀察者Subscriber 將會被調用三次 onNext() 和一次 onCompleted()）。
這樣，由被觀察者調用了觀察者的回調方法。就實現了由被觀察者向觀察者的事件傳遞，即觀察者模式。
這個樣例非常easy：事件的內容是字符串。而不是一些復雜的對象；事件的內容是已經定好了的，而不像有的觀察者模式一樣是待確定的（比如網絡請求的結果在請求返回之前是未知的）；全部事件在一瞬間被全部發送出去，而不是夾雜一些確定或不確定的時間間隔或者經過某種觸發器來觸發的。總之。這個樣例看起來毫無實用價值。
但這是為了便于說明，實質上僅僅要你想，各種各樣的事件發送規則你都能夠自己來寫。至于詳細怎么做。后面都會講到。但如今不行。
僅僅有把基礎原理先說明確了，上層的運用才干更easy說清晰。
create() 方法是 RxJava 最主要的創造事件序列的方法。
基于這種方法。 RxJava 還提供了一些方法用來快捷創建事件隊列，比如：
- just(T...): 將傳入的參數依次發送出來。
- from(T[]) / from(Iterable<? extends T>) : 將傳入的數組或 Iterable 拆分成詳細對象后，依次發送出來。
上面 just(T...) 的樣例和 from(T[]) 的樣例，都和之前的 create(OnSubscribe) 的樣例是等價的。
3) Subscribe (訂閱)
創建了 Observable 和 Observer 之后，再用 subscribe() 方法將它們聯結起來，整條鏈子就能夠工作了。代碼形式非常easy：
observable.subscribe(observer); // 或者： observable.subscribe(subscriber);有人可能會注意到。 subscribe() 這種方法有點怪：它看起來是『observalbe 訂閱了 observer / subscriber』而不是『observer / subscriber 訂閱了 observalbe』。這看起來就像『雜志訂閱了讀者』一樣顛倒了對象關系。
這讓人讀起來有點別扭。只是假設把 API 設計成 observer.subscribe(observable) / subscriber.subscribe(observable) ，盡管更加符合思維邏輯，但對流式 API 的設計就造成影響了，比較起來明顯是得不償失的。
Observable.subscribe(Subscriber) 的內部實現是這種（僅核心代碼）：
// 注意：這不是 subscribe() 的源代碼。而是將源代碼中與性能、兼容性、擴展性有關的代碼剔除后的核心代碼。 // 假設須要看源代碼，能夠去 RxJava 的 GitHub 倉庫下載。 public Subscription subscribe(Subscriber subscriber) { subscriber.onStart(); onSubscribe.call(subscriber); return subscriber; }能夠看到，subscriber() 做了3件事：
在這里。事件發送的邏輯開始運行。從這也能夠看出。在 RxJava 中， Observable 并非在創建的時候就立即開始發送事件，而是在它被訂閱的時候，即當 subscribe() 方法運行的時候。
整個過程中對象間的關系例如以下圖：
或者能夠看動圖：
除了 subscribe(Observer) 和 subscribe(Subscriber) 。subscribe() 還支持不完整定義的回調。RxJava 會自己主動依據定義創建出 Subscriber 。形式例如以下：
Action1<String> onNextAction = new Action1<String>() {// onNext()@Overridepublic void call(String s) {Log.d(tag, s);} }; Action1<Throwable> onErrorAction = new Action1<Throwable>() {// onError()@Overridepublic void call(Throwable throwable) {// Error handling} }; Action0 onCompletedAction = new Action0() {// onCompleted()@Overridepublic void call() {Log.d(tag, "completed");} };// 自己主動創建 Subscriber ，并使用 onNextAction 來定義 onNext() observable.subscribe(onNextAction); // 自己主動創建 Subscriber ，并使用 onNextAction 和 onErrorAction 來定義 onNext() 和 onError() observable.subscribe(onNextAction, onErrorAction); // 自己主動創建 Subscriber ，并使用 onNextAction、 onErrorAction 和 onCompletedAction 來定義 onNext()、 onError() 和 onCompleted() observable.subscribe(onNextAction, onErrorAction, onCompletedAction);簡單解釋一下這段代碼中出現的 Action1 和 Action0。 Action0 是 RxJava 的一個接口，它僅僅有一個方法 call()，這種方法是無參無返回值的。由于 onCompleted() 方法也是無參無返回值的，因此 Action0 能夠被當成一個包裝對象。將 onCompleted() 的內容打包起來將自己作為一個參數傳入 subscribe() 以實現不完整定義的回調。這樣事實上也能夠看做將 onCompleted() 方法作為參數傳進了 subscribe()，相當于其它某些語言中的『閉包』。
Action1 也是一個接口，它相同僅僅有一個方法 call(T param)，這種方法也無返回值，但有一個參數；與 Action0 同理。由于 onNext(T obj) 和 onError(Throwable error) 也是單參數無返回值的。因此 Action1 能夠將 onNext(obj) 和 onError(error) 打包起來傳入 subscribe() 以實現不完整定義的回調。事實上，盡管 Action0 和 Action1 在 API 中使用最廣泛。但 RxJava 是提供了多個 ActionX 形式的接口 (比如 Action2, Action3) 的，它們能夠被用以包裝不同的無返回值的方法。
注：正如前面所提到的。Observer 和 Subscriber 具有相同的角色，并且 Observer 在 subscribe() 過程中最終會被轉換成 Subscriber 對象。因此，從這里開始，后面的描寫敘述我將用 Subscriber 來取代 Observer 。這樣更加嚴謹。
4) 場景演示樣例
以下舉兩個樣例：
為了把原理用更清晰的方式表述出來。本文中挑選的都是功能盡可能簡單的樣例。以至于有些演示樣例代碼看起來會有『畫蛇添足』『明明不用 RxJava 能夠更簡便地解決這個問題』的感覺。當你看到這種情況，不要認為是由于 RxJava 太啰嗦，而是由于在過早的時候舉出真實場景的樣例并不利于原理的解析，因此我刻意挑選了簡單的情景。
a. 打印字符串數組
將字符串數組 names 中的全部字符串依次打印出來：
String[] names = ...; Observable.from(names).subscribe(new Action1<String>() {@Overridepublic void call(String name) {Log.d(tag, name);}});b. 由 id 取得圖片并顯示
由指定的一個 drawable 文件 id drawableRes 取得圖片，并顯示在 ImageView 中。并在出現異常的時候打印 Toast 報錯：
int drawableRes = ...; ImageView imageView = ...; Observable.create(new OnSubscribe<Drawable>() {@Overridepublic void call(Subscriber<? super Drawable> subscriber) { Drawable drawable = getTheme().getDrawable(drawableRes)); subscriber.onNext(drawable); subscriber.onCompleted(); } }).subscribe(new Observer<Drawable>() { @Override public void onNext(Drawable drawable) { imageView.setImageDrawable(drawable); } @Override public void onCompleted() { } @Override public void onError(Throwable e) { Toast.makeText(activity, "Error!", Toast.LENGTH_SHORT).show(); } });正如上面兩個樣例這樣。創建出 Observable 和 Subscriber ，再用 subscribe() 將它們串起來，一次 RxJava 的基本使用就完畢了。非常easy。
然而，
在 RxJava 的默認規則中。事件的發出和消費都是在同一個線程的。也就是說。假設僅僅用上面的方法，實現出來的僅僅是一個同步的觀察者模式。觀察者模式本身的目的就是『后臺處理，前臺回調』的異步機制，因此異步對于 RxJava 是至關重要的。而要實現異步。則須要用到 RxJava 的還有一個概念： Scheduler 。
3. 線程控制 —— Scheduler (一)
在不指定線程的情況下， RxJava 遵循的是線程不變的原則，即：在哪個線程調用 subscribe()。就在哪個線程生產事件；在哪個線程生產事件，就在哪個線程消費事件。
假設須要切換線程，就須要用到 Scheduler （調度器）。
1) Scheduler 的 API (一)
在RxJava 中，Scheduler ——調度器，相當于線程控制器，RxJava 通過它來指定每一段代碼應該運行在什么樣的線程。
RxJava 已經內置了幾個 Scheduler ，它們已經適合大多數的使用場景：
- Schedulers.immediate(): 直接在當前線程運行，相當于不指定線程。這是默認的 Scheduler。
- Schedulers.newThread(): 總是啟用新線程。并在新線程運行操作。
- Schedulers.io(): I/O 操作（讀寫文件、讀寫數據庫、網絡信息交互等）所使用的 Scheduler。
行為模式和 newThread() 幾乎相同。差別在于 io() 的內部實現是是用一個無數量上限的線程池，能夠重用空暇的線程，因此多數情況下 io() 比 newThread() 更有效率。
不要把計算工作放在 io() 中，能夠避免創建不必要的線程。
- Schedulers.computation(): 計算所使用的 Scheduler。
這個計算指的是 CPU 密集型計算。即不會被 I/O 等操作限制性能的操作，比如圖形的計算。這個 Scheduler 使用的固定的線程池，大小為 CPU 核數。
不要把 I/O 操作放在 computation() 中，否則 I/O 操作的等待時間會浪費 CPU。
- 另外。 Android 還有一個專用的 AndroidSchedulers.mainThread()。它指定的操作將在 Android 主線程運行。
有了這幾個 Scheduler ，就能夠使用 subscribeOn() 和 observeOn() 兩個方法來對線程進行控制了。* subscribeOn(): 指定 subscribe() 所發生的線程。即 Observable.OnSubscribe 被激活時所處的線程。或者叫做事件產生的線程。
* observeOn(): 指定 Subscriber 所運行在的線程。
或者叫做事件消費的線程。
文字敘述總歸難理解，上代碼：
Observable.just(1, 2, 3, 4).subscribeOn(Schedulers.io()) // 指定 subscribe() 發生在 IO 線程.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回調發生在主線程.subscribe(new Action1<Integer>() {@Overridepublic void call(Integer number) {Log.d(tag, "number:" + number);}});上面這段代碼中。由于 subscribeOn(Schedulers.io()) 的指定，被創建的事件的內容 1、2、3、4 將會在 IO 線程發出。而由于 observeOn(AndroidScheculers.mainThread()) 的指定，因此 subscriber 數字的打印將發生在主線程 。
事實上，這種在 subscribe() 之前寫上兩句 subscribeOn(Scheduler.io()) 和 observeOn(AndroidSchedulers.mainThread()) 的使用方式非經常見，它適用于多數的 『后臺線程取數據，主線程顯示』的程序策略。
而前面提到的由圖片 id 取得圖片并顯示的樣例。假設也加上這兩句：
int drawableRes = ...; ImageView imageView = ...; Observable.create(new OnSubscribe<Drawable>() {@Overridepublic void call(Subscriber<? super Drawable> subscriber) { Drawable drawable = getTheme().getDrawable(drawableRes)); subscriber.onNext(drawable); subscriber.onCompleted(); } }) .subscribeOn(Schedulers.io()) // 指定 subscribe() 發生在 IO 線程 .observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回調發生在主線程 .subscribe(new Observer<Drawable>() { @Override public void onNext(Drawable drawable) { imageView.setImageDrawable(drawable); } @Override public void onCompleted() { } @Override public void onError(Throwable e) { Toast.makeText(activity, "Error!", Toast.LENGTH_SHORT).show(); } });那么，載入圖片將會發生在 IO 線程，而設置圖片則被設定在了主線程。這就意味著，即使載入圖片耗費了幾十甚至幾百毫秒的時間，也不會造成絲毫界面的卡頓。
2) Scheduler 的原理 (一)
RxJava 的 Scheduler API 非常方便，也非常奇妙（加了一句話就把線程切換了，怎么做到的？并且 subscribe() 不是最外層直接調用的方法嗎，它居然也能被指定線程？）。然而 Scheduler 的原理須要放在后面講。由于它的原理是以下一節《變換》的原理作為基礎的。
好吧這一節事實上我屁也沒說，僅僅是為了讓你安心，讓你知道我不是忘了講原理，而是把它放在了更合適的地方。
4. 變換
最終要到牛逼的地方了，無論你激動不激動，反正我是激動了。
RxJava 提供了對事件序列進行變換的支持。這是它的核心功能之中的一個，也是大多數人說『RxJava 真是太好用了』的最大原因。
所謂變換，就是將事件序列中的對象或整個序列進行加工處理，轉換成不同的事件或事件序列。概念說著總是模糊難懂的，來看 API。
1) API
首先看一個 map() 的樣例：
Observable.just("images/logo.png") // 輸入類型 String.map(new Func1<String, Bitmap>() {@Overridepublic Bitmap call(String filePath) { // 參數類型 Stringreturn getBitmapFromPath(filePath); // 返回類型 Bitmap}}).subscribe(new Action1<Bitmap>() {@Overridepublic void call(Bitmap bitmap) { // 參數類型 BitmapshowBitmap(bitmap);}});這里出現了一個叫做 Func1 的類。它和 Action1 非常類似，也是 RxJava 的一個接口，用于包裝含有一個參數的方法。 Func1 和 Action 的差別在于， Func1 包裝的是有返回值的方法。
另外，和 ActionX 一樣。 FuncX 也有多個。用于不同參數個數的方法。FuncX 和 ActionX 的差別在 FuncX 包裝的是有返回值的方法。
能夠看到，map() 方法將參數中的 String 對象轉換成一個 Bitmap 對象后返回，而在經過 map() 方法后，事件的參數類型也由 String 轉為了 Bitmap。
這種直接變換對象并返回的。是最常見的也最easy理解的變換。只是 RxJava 的變換遠不止這樣。它不僅能夠針對事件對象，還能夠針對整個事件隊列，這使得 RxJava 變得非常靈活。我列舉幾個經常使用的變換：
-
map(): 事件對象的直接變換，詳細功能上面已經介紹過。它是 RxJava 最經常使用的變換。
map() 的示意圖：
-
flatMap(): 這是一個非常實用但非常難理解的變換，因此我決定花多些篇幅來介紹它。
首先假設這么一種需求：假設有一個數據結構『學生』。如今須要打印出一組學生的名字。實現方式非常easy：
非常easy。那么再假設：假設要打印出每一個學生所須要修的全部課程的名稱呢？（需求的差別在于。每一個學生僅僅有一個名字，但卻有多個課程。）首先能夠這樣實現：
Student[] students = ...; Subscriber<Student> subscriber = new Subscriber<Student>() {@Overridepublic void onNext(Student student) {List<Course> courses = student.getCourses();for (int i = 0; i < courses.size(); i++) {Course course = courses.get(i);Log.d(tag, course.getName());}}... }; Observable.from(students).subscribe(subscriber);依舊非常easy。
那么假設我不想在 Subscriber 中使用 for 循環。而是希望 Subscriber 中直接傳入單個的 Course 對象呢（這對于代碼復用非常重要）？用 map() 顯然是不行的，由于 map() 是一對一的轉化，而我如今的要求是一對多的轉化。那怎么才干把一個 Student 轉化成多個 Course 呢？
這個時候，就須要用 flatMap() 了：
Student[] students = ...; Subscriber<Course> subscriber = new Subscriber<Course>() {@Overridepublic void onNext(Course course) {Log.d(tag, course.getName());}... }; Observable.from(students).flatMap(new Func1<Student, Observable<Course>>() {@Overridepublic Observable<Course> call(Student student) {return Observable.from(student.getCourses());}}).subscribe(subscriber);從上面的代碼能夠看出。 flatMap() 和 map() 有一個相同點：它也是把傳入的參數轉化之后返回還有一個對象。但須要注意，和 map() 不同的是。 flatMap() 中返回的是個 Observable 對象，并且這個 Observable 對象并非被直接發送到了 Subscriber 的回調方法中。 flatMap() 的原理是這種：1. 使用傳入的事件對象創建一個 Observable 對象；2. 并不發送這個 Observable, 而是將它激活，于是它開始發送事件；3. 每一個創建出來的 Observable 發送的事件，都被匯入同一個 Observable ，而這個 Observable 負責將這些事件統一交給 Subscriber 的回調方法。這三個步驟，把事件拆成了兩級，通過一組新創建的 Observable 將初始的對象『鋪平』之后通過統一路徑分發了下去。而這個『鋪平』就是 flatMap() 所謂的 flat。
flatMap() 示意圖：
擴展：由于能夠在嵌套的 Observable 中加入異步代碼。 flatMap() 也經常使用于嵌套的異步操作，比如嵌套的網絡請求。演示樣例代碼（Retrofit + RxJava）：
networkClient.token() // 返回 Observable<String>，在訂閱時請求 token。并在響應后發送 token.flatMap(new Func1<String, Observable<Messages>>() {@Overridepublic Observable<Messages> call(String token) {// 返回 Observable<Messages>，在訂閱時請求消息列表，并在響應后發送請求到的消息列表return networkClient.messages();}}).subscribe(new Action1<Messages>() {@Overridepublic void call(Messages messages) {// 處理顯示消息列表showMessages(messages);}});傳統的嵌套請求須要使用嵌套的 Callback 來實現。而通過 flatMap() ，能夠把嵌套的請求寫在一條鏈中，從而保持程序邏輯的清晰。
- throttleFirst(): 在每次事件觸發后的一定時間間隔內丟棄新的事件。經常使用作去抖動過濾。比如按鈕的點擊監聽器：RxView.clickEvents(button) // RxBinding 代碼，后面的文章有解釋.throttleFirst(500, TimeUnit.MILLISECONDS) // 設置防抖間隔為 500ms.subscribe(subscriber);媽媽再也不怕我的用戶手抖點開兩個重復的界面啦。
此外。 RxJava 還提供非常多便捷的方法來實現事件序列的變換。這里就不一一舉例了。
2) 變換的原理：lift()
這些變換盡管功能各有不同，但實質上都是針對事件序列的處理和再發送。
而在 RxJava 的內部。它們是基于同一個基礎的變換方法： lift(Operator)。
首先看一下 lift() 的內部實現（僅核心代碼）：
// 注意：這不是 lift() 的源代碼。而是將源代碼中與性能、兼容性、擴展性有關的代碼剔除后的核心代碼。 // 假設須要看源代碼，能夠去 RxJava 的 GitHub 倉庫下載。 public <R> Observable<R> lift(Operator<? extends R, ? super T> operator) { return Observable.create(new OnSubscribe<R>() { @Override public void call(Subscriber subscriber) { Subscriber newSubscriber = operator.call(subscriber); newSubscriber.onStart(); onSubscribe.call(newSubscriber); } }); }這段代碼非常有意思：它生成了一個新的 Observable 并返回，并且創建新 Observable 所用的參數 OnSubscribe 的回調方法 call() 中的實現居然看起來和前面講過的 Observable.subscribe() 一樣。然而它們并不一樣喲~不一樣的地方關鍵就在于第二行 onSubscribe.call(subscriber) 中的 onSubscribe 所指代的對象不同（高能預警：接下來的幾句話可能會導致身體的嚴重不適）——
- subscribe() 中這句話的 onSubscribe 指的是 Observable 中的 onSubscribe 對象，這個沒有問題。但是 lift() 之后的情況就復雜了點。
- 當含有 lift() 時：
1.lift() 創建了一個 Observable 后，加上之前的原始 Observable，已經有兩個 Observable 了；
2.而相同地，新 Observable 里的新 OnSubscribe 加上之前的原始 Observable 中的原始 OnSubscribe，也就有了兩個 OnSubscribe；
3.當用戶調用經過 lift() 后的 Observable 的 subscribe() 的時候，使用的是 lift() 所返回的新的 Observable ，于是它所觸發的 onSubscribe.call(subscriber)。也是用的新 Observable 中的新 OnSubscribe。即在 lift() 中生成的那個 OnSubscribe；
4.而這個新 OnSubscribe 的 call() 方法中的 onSubscribe ，就是指的原始 Observable 中的原始 OnSubscribe ，在這個 call() 方法里。新 OnSubscribe 利用 operator.call(subscriber) 生成了一個新的 Subscriber（Operator 就是在這里，通過自己的 call() 方法將新 Subscriber 和原始 Subscriber 進行關聯。并插入自己的『變換』代碼以實現變換）。然后利用這個新 Subscriber 向原始 Observable 進行訂閱。
這樣就實現了 lift() 過程，有點像一種代理機制，通過事件攔截和處理實現事件序列的變換。
精簡掉細節的話，也能夠這么說：在 Observable 運行了 lift(Operator) 方法之后，會返回一個新的 Observable，這個新的 Observable 會像一個代理一樣。負責接收原始的 Observable 發出的事件，并在處理后發送給 Subscriber。
假設你更喜歡具象思維，能夠看圖：
或者能夠看動圖：
兩次和多次的 lift() 同理，例如以下圖：
舉一個詳細的 Operator 的實現。以下這是一個將事件中的 Integer 對象轉換成 String 的樣例，僅供參考：
observable.lift(new Observable.Operator<String, Integer>() {@Overridepublic Subscriber<? super Integer> call(final Subscriber<? super String> subscriber) { // 將事件序列中的 Integer 對象轉換為 String 對象 return new Subscriber<Integer>() { @Override public void onNext(Integer integer) { subscriber.onNext("" + integer); } @Override public void onCompleted() { subscriber.onCompleted(); } @Override public void onError(Throwable e) { subscriber.onError(e); } }; } });講述 lift() 的原理僅僅是為了讓你更好地了解 RxJava 。從而能夠更好地使用它。然而無論你是否理解了 lift() 的原理，RxJava 都不建議開發人員自己定義 Operator 來直接使用 lift()，而是建議盡量使用已有的 lift() 包裝方法（如 map() flatMap() 等）進行組合來實現需求，由于直接使用 lift() 非常easy發生一些難以發現的錯誤。
3) compose: 對 Observable 總體的變換
除了 lift() 之外， Observable 還有一個變換方法叫做 compose(Transformer)。它和 lift() 的差別在于。 lift() 是針對事件項和事件序列的，而 compose() 是針對 Observable 自身進行變換。
舉個樣例，假設在程序中有多個 Observable ，并且他們都須要應用一組相同的 lift() 變換。你能夠這么寫：
observable1.lift1().lift2().lift3().lift4().subscribe(subscriber1); observable2.lift1().lift2().lift3().lift4().subscribe(subscriber2); observable3.lift1().lift2().lift3().lift4().subscribe(subscriber3); observable4.lift1().lift2().lift3().lift4().subscribe(subscriber1);你認為這樣太不軟件project了，于是你改成了這樣：
private Observable liftAll(Observable observable) {return observable.lift1().lift2().lift3().lift4(); } ... liftAll(observable1).subscribe(subscriber1); liftAll(observable2).subscribe(subscriber2); liftAll(observable3).subscribe(subscriber3); liftAll(observable4).subscribe(subscriber4);可讀性、可維護性都提高了。但是 Observable 被一個方法包起來，這種方式對于 Observale 的靈活性似乎還是增添了那么點限制。
怎么辦？這個時候。就應該用 compose() 來攻克了：
public class LiftAllTransformer implements Observable.Transformer<Integer, String> {@Overridepublic Observable<String> call(Observable<Integer> observable) {return observable.lift1().lift2().lift3().lift4();} } ... Transformer liftAll = new LiftAllTransformer(); observable1.compose(liftAll).subscribe(subscriber1); observable2.compose(liftAll).subscribe(subscriber2); observable3.compose(liftAll).subscribe(subscriber3); observable4.compose(liftAll).subscribe(subscriber4);像上面這樣，使用 compose() 方法，Observable 能夠利用傳入的 Transformer 對象的 call 方法直接對自身進行處理，也就不必被包在方法的里面了。
compose() 的原理比較簡單，不附圖嘍。
5. 線程控制：Scheduler (二)
除了靈活的變換，RxJava 還有一個牛逼的地方，就是線程的自由控制。
1) Scheduler 的 API (二)
前面講到了，能夠利用 subscribeOn() 結合 observeOn() 來實現線程控制。讓事件的產生和消費發生在不同的線程。
但是在了解了 map() flatMap() 等變換方法后。有些好事的（事實上就是當初剛接觸 RxJava 時的我）就問了：能不能多切換幾次線程？
答案是：能。
由于 observeOn() 指定的是 Subscriber 的線程。而這個 Subscriber 并非（嚴格說應該為『不一定是』。但這里最好還是理解為『不是』）subscribe() 參數中的 Subscriber 。而是 observeOn() 運行時的當前 Observable 所相應的 Subscriber 。即它的直接下級 Subscriber 。換句話說。observeOn() 指定的是它之后的操作所在的線程。
因此假設有多次切換線程的需求，僅僅要在每一個想要切換線程的位置調用一次 observeOn() 就可以。上代碼：
Observable.just(1, 2, 3, 4) // IO 線程，由 subscribeOn() 指定.subscribeOn(Schedulers.io()).observeOn(Schedulers.newThread()).map(mapOperator) // 新線程。由 observeOn() 指定.observeOn(Schedulers.io()).map(mapOperator2) // IO 線程，由 observeOn() 指定.observeOn(AndroidSchedulers.mainThread) .subscribe(subscriber); // Android 主線程，由 observeOn() 指定如上，通過 observeOn() 的多次調用，程序實現了線程的多次切換。
只是，不同于 observeOn() ， subscribeOn() 的位置放在哪里都能夠，但它是僅僅能調用一次的。
又有好事的（事實上還是當初的我）問了：假設我非要調用多次 subscribeOn() 呢？會有什么效果？
這個問題先放著，我們還是從 RxJava 線程控制的原理說起吧。
2) Scheduler 的原理（二）
事實上。 subscribeOn() 和 observeOn() 的內部實現，也是用的 lift()。
詳細看圖（不同顏色的箭頭表示不同的線程）：
subscribeOn() 原理圖：
observeOn() 原理圖：
從圖中能夠看出，subscribeOn() 和 observeOn() 都做了線程切換的工作（圖中的 "schedule..." 部位）。
不同的是。 subscribeOn() 的線程切換發生在 OnSubscribe 中，即在它通知上一級 OnSubscribe 時，這時事件還沒有開始發送。因此 subscribeOn() 的線程控制能夠從事件發出的開端就造成影響；而 observeOn() 的線程切換則發生在它內建的 Subscriber 中，即發生在它即將給下一級 Subscriber 發送事件時，因此 observeOn() 控制的是它后面的線程。
最后，我用一張圖來解釋當多個 subscribeOn() 和 observeOn() 混合使用時。線程調度是怎么發生的（由于圖中對象較多。相對于上面的圖對結構做了一些簡化調整）：
圖中共同擁有 5 處含有對事件的操作。由圖中能夠看出。①和②兩處受第一個 subscribeOn() 影響。運行在紅色線程；③和④處受第一個 observeOn() 的影響。運行在綠色線程；⑤處受第二個 onserveOn() 影響，運行在紫色線程。而第二個 subscribeOn() ，由于在通知過程中線程就被第一個 subscribeOn() 截斷，因此對整個流程并沒有不論什么影響。這里也就回答了前面的問題：當使用了多個 subscribeOn() 的時候，僅僅有第一個 subscribeOn() 起作用。
3) 延伸：doOnSubscribe()
然而，盡管超過一個的 subscribeOn() 對事件處理的流程沒有影響。但在流程之前卻是能夠利用的。
在前面講 Subscriber 的時候。提到過 Subscriber 的 onStart() 能夠用作流程開始前的初始化。然而 onStart() 由于在 subscribe() 發生時就被調用了。因此不能指定線程。而是僅僅能運行在 subscribe() 被調用時的線程。
這就導致假設 onStart() 中含有對線程有要求的代碼（比如在界面上顯示一個 ProgressBar，這必須在主線程運行）。將會有線程非法的風險。由于有時你無法預測 subscribe() 將會在什么線程運行。
而與 Subscriber.onStart() 相相應的。有一個方法 Observable.doOnSubscribe() 。它和 Subscriber.onStart() 相同是在 subscribe() 調用后并且在事件發送前運行，但差別在于它能夠指定線程。
默認情況下。 doOnSubscribe() 運行在 subscribe() 發生的線程；而假設在 doOnSubscribe() 之后有 subscribeOn() 的話，它將運行在離它近期的 subscribeOn() 所指定的線程。
演示樣例代碼：
Observable.create(onSubscribe).subscribeOn(Schedulers.io()).doOnSubscribe(new Action0() {@Overridepublic void call() {progressBar.setVisibility(View.VISIBLE); // 須要在主線程運行}}).subscribeOn(AndroidSchedulers.mainThread()) // 指定主線程.observeOn(AndroidSchedulers.mainThread()).subscribe(subscriber);如上，在 doOnSubscribe()的后面跟一個 subscribeOn() 。就能指定準備工作的線程了。
RxJava 的適用場景和使用方式
1. 與 Retrofit 的結合
Retrofit 是 Square 的一個著名的網絡請求庫。
沒實用過 Retrofit 的能夠選擇跳過這一小節也沒關系，我舉的每種場景都僅僅是個樣例。并且樣例之間并無前后關聯，僅僅是個拋磚引玉的作用，所以你跳過這里看別的場景也能夠的。
Retrofit 除了提供了傳統的 Callback 形式的 API，還有 RxJava 版本號的 Observable 形式 API。以下我用對照的方式來介紹 Retrofit 的 RxJava 版 API 和傳統版本號的差別。
以獲取一個 User 對象的接口作為樣例。使用Retrofit 的傳統 API，你能夠用這種方式來定義請求：
@GET("/user") public void getUser(@Query("userId") String userId, Callback<User> callback);在程序的構建過程中。 Retrofit 會把自己主動把方法實現并生成代碼。然后開發人員就能夠利用以下的方法來獲取特定用戶并處理響應：
getUser(userId, new Callback<User>() {@Overridepublic void success(User user) {userView.setUser(user);}@Overridepublic void failure(RetrofitError error) {// Error handling...} };而使用 RxJava 形式的 API。定義相同的請求是這種：
@GET("/user") public Observable<User> getUser(@Query("userId") String userId);使用的時候是這種：
getUser(userId).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<User>() {@Overridepublic void onNext(User user) {userView.setUser(user);}@Overridepublic void onCompleted() {}@Overridepublic void onError(Throwable error) {// Error handling...}});看到差別了嗎？
當 RxJava 形式的時候，Retrofit 把請求封裝進 Observable ，在請求結束后調用 onNext() 或在請求失敗后調用 onError()。
對照來看， Callback 形式和 Observable 形式長得不太一樣，但本質都幾乎相同。并且在細節上 Observable 形式似乎還比 Callback 形式要差點。那 Retrofit 為什么還要提供 RxJava 的支持呢？
由于它好用啊！
從這個樣例看不出來是由于這僅僅是最簡單的情況。
而一旦情景復雜起來， Callback 形式立即就會開始讓人頭疼。比方：
假設這么一種情況：你的程序取到的 User 并不應該直接顯示，而是須要先與數據庫中的數據進行比對和修正后再顯示。
使用 Callback 方式大概能夠這么寫：
getUser(userId, new Callback<User>() {@Overridepublic void success(User user) {processUser(user); // 嘗試修正 User 數據userView.setUser(user);}@Overridepublic void failure(RetrofitError error) {// Error handling...} };有問題嗎？
非常簡便。但不要這樣做。為什么？由于這樣做會影響性能。數據庫的操作非常重。一次讀寫操作花費 10~20ms 是非經常見的。這種耗時非常easy造成界面的卡頓。
所以通常情況下，假設能夠的話一定要避免在主線程中處理數據庫。
所以為了提升性能。這段代碼能夠優化一下：
getUser(userId, new Callback<User>() {@Overridepublic void success(User user) {new Thread() {@Overridepublic void run() {processUser(user); // 嘗試修正 User 數據runOnUiThread(new Runnable() { // 切回 UI 線程@Overridepublic void run() {userView.setUser(user);}});}).start();}@Overridepublic void failure(RetrofitError error) {// Error handling...} };性能問題解決，但……這代碼實在是太亂了，迷之縮進啊！雜亂的代碼往往不僅僅是美觀問題，由于代碼越亂往往就越難讀懂，而假設項目中充斥著雜亂的代碼，無疑會減少代碼的可讀性，造成團隊開發效率的減少和出錯率的升高。
這時候，假設用 RxJava 的形式，就好辦多了。 RxJava 形式的代碼是這種：
getUser(userId).doOnNext(new Action1<User>() {@Overridepublic void call(User user) {processUser(user);}).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<User>() {@Overridepublic void onNext(User user) {userView.setUser(user);}@Overridepublic void onCompleted() {}@Overridepublic void onError(Throwable error) {// Error handling...}});后臺代碼和前臺代碼全都寫在一條鏈中，明顯清晰了非常多。
再舉一個樣例：假設 /user 接口并不能直接訪問，而須要填入一個在線獲取的 token ，代碼應該怎么寫？
Callback 方式。能夠使用嵌套的 Callback：
@GET("/token") public void getToken(Callback<String> callback);@GET("/user") public void getUser(@Query("token") String token, @Query("userId") String userId, Callback<User> callback);...getToken(new Callback<String>() {@Overridepublic void success(String token) {getUser(token, userId, new Callback<User>() {@Overridepublic void success(User user) {userView.setUser(user);}@Overridepublic void failure(RetrofitError error) {// Error handling...}};}@Overridepublic void failure(RetrofitError error) {// Error handling...} });倒是沒有什么性能問題，但是迷之縮進毀一生，你懂我也懂。做過大項目的人應該更懂。
而使用 RxJava 的話。代碼是這種：
@GET("/token") public Observable<String> getToken();@GET("/user") public Observable<User> getUser(@Query("token") String token, @Query("userId") String userId);...getToken().flatMap(new Func1<String, Observable<User>>() {@Overridepublic Observable<User> onNext(String token) {return getUser(token, userId);}).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<User>() {@Overridepublic void onNext(User user) {userView.setUser(user);}@Overridepublic void onCompleted() {}@Overridepublic void onError(Throwable error) {// Error handling...}});用一個 flatMap() 就搞定了邏輯，依舊是一條鏈。
看著就非常爽。是吧？
2016/03/31 更新，加上我寫的一個 Sample 項目：
rengwuxian RxJava Samples
好。Retrofit 部分就到這里。
2. RxBinding
RxBinding 是 Jake Wharton 的一個開源庫，它提供了一套在 Android 平臺上的基于 RxJava 的 Binding API。
所謂 Binding，就是類似設置 OnClickListener 、設置 TextWatcher 這種注冊綁定對象的 API。
舉個設置點擊監聽的樣例。
使用 RxBinding 。能夠把事件監聽用這種方法來設置：
Button button = ...; RxView.clickEvents(button) // 以 Observable 形式來反饋點擊事件.subscribe(new Action1<ViewClickEvent>() {@Overridepublic void call(ViewClickEvent event) {// Click handling}});看起來除了形式變了沒什么差別，實質上也是這樣。甚至假設你看一下它的源代碼，你會發現它連實現都沒什么驚喜：它的內部是直接用一個包裹著的 setOnClickListener() 來實現的。然而。僅僅這一個形式的改變。卻恰好就是 RxBinding 的目的：擴展性。通過 RxBinding 把點擊監聽轉換成 Observable 之后，就有了對它進行擴展的可能。
擴展的方式有非常多，依據需求而定。
一個樣例是前面提到過的 throttleFirst() ，用于去抖動，也就是消除手抖導致的高速連環點擊：
RxView.clickEvents(button).throttleFirst(500, TimeUnit.MILLISECONDS).subscribe(clickAction);假設想對 RxBinding 有很多其它了解。能夠去它的 GitHub 項目 以下看看。
3. 各種異步操作
前面舉的 Retrofit 和 RxBinding 的樣例，是兩個能夠提供現成的 Observable 的庫。而假設你有某些異步操作無法用這些庫來自己主動生成 Observable。也全然能夠自己寫。比如數據庫的讀寫、大圖片的載入、文件壓縮/解壓等各種須要放在后臺工作的耗時操作。都能夠用 RxJava 來實現。有了之前幾章的樣例。這里應該不用再舉例了。
4. RxBus
RxBus 名字看起來像一個庫，但它并非一個庫，而是一種模式，它的思想是使用 RxJava 來實現了 EventBus ，而讓你不再須要使用 Otto 或者 GreenRobot 的 EventBus。至于什么是 RxBus，能夠看這篇文章。順便說一句，Flipboard 已經用 RxBus 替換掉了 Otto ，眼下為止沒有不良反應。
最后
對于 Android 開發人員來說。 RxJava 是一個非常難上手的庫，由于它對于 Android 開發人員來說有太多陌生的概念了。
但是它真的非常牛逼。因此，我寫了這篇《給 Android 開發人員的 RxJava 詳細解釋》。希望能給始終搞不明確什么是 RxJava 的人一些入門的指引，或者能讓正在使用 RxJava 但仍然心存疑惑的人看到一些更深入的解析。無論如何。僅僅要能給各位同為 Android project師的你們提供一些幫助。這篇文章的目的就達到了。
再次感謝對這篇文章的產出提供支持的各位：
技術支持：流火楓林
內測讀者：代碼家、鮑永章、drakeet、馬琳、有時放縱、程序亦非猿、大頭鬼、XZoomEye、席德雨、TCahead、Tiiime、Ailurus、宅學長、妖孽、大大大大大臣哥、NicodeLee
贊助方：周伯通招聘是他們讓我的文章能夠以不那么丑陋的樣子出如今大家面前。
關于作者
朱凱（扔物線），Flipboard 北京 Android project師。
微博：扔物線
GitHub：rengwuxian
為什么寫這個？
與兩三年前的境況不同，中國如今已經不缺0基礎 Android project師。但中級和高級project師嚴重供不應求。
因此我決定從今天開始不定期地公布我的技術分享。僅僅希望能夠和大家共同提升，通過我們的成長來解決一點點國內互聯網公司人才稀缺的困境。也提升各位技術黨的收入。
所以，不僅要寫這篇，我還會寫很多其它。
至于內容的定位，我計劃僅僅定位真正的干貨，一些邊邊角角的小技巧和炫酷的黑科技應該都不會寫，總之希望每篇文章都能幫讀者提升真正的實力。
轉載于:https://www.cnblogs.com/llguanli/p/8327435.html
總結
以上是生活随笔為你收集整理的给 Android 开发人员的 RxJava 具体解释的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： (转)计算机网络基础知识总结
- 下一篇： Android 9.0新特性