資訊專欄INFORMATION COLUMN
摘要：觀察者模式面向的需求是對象觀察者對對象被觀察者的某種變化高度敏感，需要在變化的一瞬間做出反應(yīng)。規(guī)定，當(dāng)不會再有新的發(fā)出時，需要觸發(fā)方法作為標(biāo)志。在事件處理過程中出異常時，會被觸發(fā)，同時隊(duì)列自動終止，不允許再有事件發(fā)出。
我從去年開始使用 RxJava ，到現(xiàn)在一年多了。今年加入了 Flipboard 后，看到 Flipboard 的 Android 項(xiàng)目也在使用 RxJava ，并且使用的場景越來越多 。而最近這幾個月，我也發(fā)現(xiàn)國內(nèi)越來越多的人開始提及 RxJava 。有人說『RxJava 真是太好用了』，有人說『RxJava 真是太難用了』，另外更多的人表示：我真的百度了也谷歌了，但我還是想問： RxJava 到底是什么？
鑒于 RxJava 目前這種既火爆又神秘的現(xiàn)狀，而我又在一年的使用過程中對 RxJava 有了一些理解，我決定寫下這篇文章來對 RxJava 做一個相對詳細(xì)的、針對 Android 開發(fā)者的介紹。
這篇文章的目的有兩個： 1. 給對 RxJava 感興趣的人一些入門的指引 2. 給正在使用 RxJava 但仍然心存疑惑的人一些更深入的解析
RxJava 到底是什么
RxJava 好在哪
API 介紹和原理簡析
1. 概念：擴(kuò)展的觀察者模式
觀察者模式
RxJava 的觀察者模式
2. 基本實(shí)現(xiàn)
1) 創(chuàng)建 Observer
2) 創(chuàng)建 Observable
3) Subscribe (訂閱)
4) 場景示例
a. 打印字符串?dāng)?shù)組
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()
RxJava 的適用場景和使用方式
1. 與 Retrofit 的結(jié)合
2. RxBinding
3. 各種異步操作
4. RxBus
最后
關(guān)于作者：
為什么寫這個？
在正文開始之前的最后，放上 GitHub 鏈接和引入依賴的 gradle 代碼： Github：
https://github.com/ReactiveX/...
https://github.com/ReactiveX/...
引入依賴：
compile "io.reactivex:rxjava:1.0.14"
compile "io.reactivex:rxandroid:1.0.1"
（版本號是文章發(fā)布時的最新穩(wěn)定版）
另外，感謝 RxJava 核心成員流火楓林的技術(shù)支持和內(nèi)測讀者代碼家、鮑永章、drakeet、馬琳、有時放縱、程序亦非猿、大頭鬼、XZoomEye、席德雨、TCahead、Tiiime、Ailurus、宅學(xué)長、妖孽、大大大大大臣哥、NicodeLee的幫助，以及周伯通招聘的贊助。
RxJava 到底是什么一個詞：異步。
RxJava 在 GitHub 主頁上的自我介紹是 "a library for composing asynchronous and event-based programs using observable sequences for the Java VM"（一個在 Java VM 上使用可觀測的序列來組成異步的、基于事件的程序的庫）。這就是 RxJava ，概括得非常精準(zhǔn)。
然而，對于初學(xué)者來說，這太難看懂了。因?yàn)樗且粋€『總結(jié)』，而初學(xué)者更需要一個『引言』。
其實(shí)， RxJava 的本質(zhì)可以壓縮為異步這一個詞。說到根上，它就是一個實(shí)現(xiàn)異步操作的庫，而別的定語都是基于這之上的。
RxJava 好在哪換句話說，『同樣是做異步，為什么人們用它，而不用現(xiàn)成的 AsyncTask / Handler / XXX / ... ？』
一個詞：簡潔。
異步操作很關(guān)鍵的一點(diǎn)是程序的簡潔性，因?yàn)樵谡{(diào)度過程比較復(fù)雜的情況下，異步代碼經(jīng)常會既難寫也難被讀懂。 Android 創(chuàng)造的 AsyncTask 和Handler ，其實(shí)都是為了讓異步代碼更加簡潔。RxJava 的優(yōu)勢也是簡潔，但它的簡潔的與眾不同之處在于，隨著程序邏輯變得越來越復(fù)雜，它依然能夠保持簡潔。
假設(shè)有這樣一個需求：界面上有一個自定義的視圖 imageCollectorView ，它的作用是顯示多張圖片，并能使用 addImage(Bitmap) 方法來任意增加顯示的圖片。現(xiàn)在需要程序?qū)⒁粋€給出的目錄數(shù)組 File[] folders 中每個目錄下的 png 圖片都加載出來并顯示在 imageCollectorView 中。需要注意的是，由于讀取圖片的這一過程較為耗時，需要放在后臺執(zhí)行，而圖片的顯示則必須在 UI 線程執(zhí)行。常用的實(shí)現(xiàn)方式有多種，我這里貼出其中一種：
new Thread() {
@Override
public 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() {
@Override
public void run() {
imageCollectorView.addImage(bitmap);
}
});
}
}
}
}
}.start();
而如果使用 RxJava ，實(shí)現(xiàn)方式是這樣的：
Observable.from(folders)
.flatMap(new Func1>() {
@Override
public Observable call(File file) {
return Observable.from(file.listFiles());
}
})
.filter(new Func1() {
@Override
public Boolean call(File file) {
return file.getName().endsWith(".png");
}
})
.map(new Func1() {
@Override
public Bitmap call(File file) {
return getBitmapFromFile(file);
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1() {
@Override
public void call(Bitmap bitmap) {
imageCollectorView.addImage(bitmap);
}
});
那位說話了：『你這代碼明明變多了啊！簡潔個毛啊！』大兄弟你消消氣，我說的是邏輯的簡潔，不是單純的代碼量少（邏輯簡潔才是提升讀寫代碼速度的必殺技對不？）。觀察一下你會發(fā)現(xiàn)， RxJava 的這個實(shí)現(xiàn)，是一條從上到下的鏈?zhǔn)秸{(diào)用，沒有任何嵌套，這在邏輯的簡潔性上是具有優(yōu)勢的。當(dāng)需求變得復(fù)雜時，這種優(yōu)勢將更加明顯（試想如果還要求只選取前 10 張圖片，常規(guī)方式要怎么辦？如果有更多這樣那樣的要求呢？再試想，在這一大堆需求實(shí)現(xiàn)完兩個月之后需要改功能，當(dāng)你翻回這里看到自己當(dāng)初寫下的那一片迷之縮進(jìn)，你能保證自己將迅速看懂，而不是對著代碼重新捋一遍思路？）。
另外，如果你的 IDE 是 Android Studio ，其實(shí)每次打開某個 Java 文件的時候，你會看到被自動 Lambda 化的預(yù)覽，這將讓你更加清晰地看到程序邏輯：
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) });
如果你習(xí)慣使用 Retrolambda ，你也可以直接把代碼寫成上面這種簡潔的形式。而如果你看到這里還不知道什么是 Retrolambda ，我不建議你現(xiàn)在就去學(xué)習(xí)它。原因有兩點(diǎn)：1. Lambda 是把雙刃劍，它讓你的代碼簡潔的同時，降低了代碼的可讀性，因此同時學(xué)習(xí) RxJava 和 Retrolambda 可能會讓你忽略 RxJava 的一些技術(shù)細(xì)節(jié)；2. Retrolambda 是 Java 6/7 對 Lambda 表達(dá)式的非官方兼容方案，它的向后兼容性和穩(wěn)定性是無法保障的，因此對于企業(yè)項(xiàng)目，使用 Retrolambda 是有風(fēng)險的。所以，與很多 RxJava 的推廣者不同，我并不推薦在學(xué)習(xí) RxJava 的同時一起學(xué)習(xí) Retrolambda。事實(shí)上，我個人雖然很欣賞 Retrolambda，但我從來不用它。
在Flipboard 的 Android 代碼中，有一段邏輯非常復(fù)雜，包含了多次內(nèi)存操作、本地文件操作和網(wǎng)絡(luò)操作，對象分分合合，線程間相互配合相互等待，一會兒排成人字，一會兒排成一字。如果使用常規(guī)的方法來實(shí)現(xiàn)，肯定是要寫得欲仙欲死，然而在使用 RxJava 的情況下，依然只是一條鏈?zhǔn)秸{(diào)用就完成了。它很長，但很清晰。
所以， RxJava 好在哪？就好在簡潔，好在那把什么復(fù)雜邏輯都能穿成一條線的簡潔。
API 介紹和原理簡析這個我就做不到一個詞說明了……因?yàn)檫@一節(jié)的主要內(nèi)容就是一步步地說明 RxJava 到底怎樣做到了異步，怎樣做到了簡潔。
1. 概念：擴(kuò)展的觀察者模式RxJava 的異步實(shí)現(xiàn)，是通過一種擴(kuò)展的觀察者模式來實(shí)現(xiàn)的。
先簡述一下觀察者模式，已經(jīng)熟悉的可以跳過這一段。
觀察者模式面向的需求是：A 對象（觀察者）對 B 對象（被觀察者）的某種變化高度敏感，需要在 B 變化的一瞬間做出反應(yīng)。舉個例子，新聞里喜聞樂見的警察抓小偷，警察需要在小偷伸手作案的時候?qū)嵤┳ゲ丁Ｔ谶@個例子里，警察是觀察者，小偷是被觀察者，警察需要時刻盯著小偷的一舉一動，才能保證不會漏過任何瞬間。程序的觀察者模式和這種真正的『觀察』略有不同，觀察者不需要時刻盯著被觀察者（例如 A 不需要每過 2ms 就檢查一次 B 的狀態(tài)），而是采用注冊(Register)或者稱為訂閱(Subscribe)的方式，告訴被觀察者：我需要你的某某狀態(tài)，你要在它變化的時候通知我。 Android 開發(fā)中一個比較典型的例子是點(diǎn)擊監(jiān)聽器 OnClickListener 。對設(shè)置 OnClickListener 來說， View 是被觀察者， OnClickListener 是觀察者，二者通過 setOnClickListener() 方法達(dá)成訂閱關(guān)系。訂閱之后用戶點(diǎn)擊按鈕的瞬間，Android Framework 就會將點(diǎn)擊事件發(fā)送給已經(jīng)注冊的 OnClickListener 。采取這樣被動的觀察方式，既省去了反復(fù)檢索狀態(tài)的資源消耗，也能夠得到最高的反饋速度。當(dāng)然，這也得益于我們可以隨意定制自己程序中的觀察者和被觀察者，而警察叔叔明顯無法要求小偷『你在作案的時候務(wù)必通知我』。
OnClickListener 的模式大致如下圖：
如圖所示，通過 setOnClickListener() 方法，Button 持有 OnClickListener 的引用（這一過程沒有在圖上畫出）；當(dāng)用戶點(diǎn)擊時，Button 自動調(diào)用 OnClickListener 的 onClick() 方法。另外，如果把這張圖中的概念抽象出來（Button -> 被觀察者、OnClickListener -> 觀察者、setOnClickListener() -> 訂閱，onClick() -> 事件），就由專用的觀察者模式（例如只用于監(jiān)聽控件點(diǎn)擊）轉(zhuǎn)變成了通用的觀察者模式。如下圖：
而 RxJava 作為一個工具庫，使用的就是通用形式的觀察者模式。
RxJava 有四個基本概念：Observable (可觀察者，即被觀察者)、 Observer (觀察者)、 subscribe (訂閱)、事件。Observable 和 Observer 通過 subscribe() 方法實(shí)現(xiàn)訂閱關(guān)系，從而 Observable 可以在需要的時候發(fā)出事件來通知 Observer。
與傳統(tǒng)觀察者模式不同， RxJava 的事件回調(diào)方法除了普通事件 onNext() （相當(dāng)于 onClick() / onEvent()）之外，還定義了兩個特殊的事件：onCompleted() 和 onError()。
onCompleted(): 事件隊(duì)列完結(jié)。RxJava 不僅把每個事件多帶帶處理，還會把它們看做一個隊(duì)列。RxJava 規(guī)定，當(dāng)不會再有新的 onNext() 發(fā)出時，需要觸發(fā) onCompleted() 方法作為標(biāo)志。
onError(): 事件隊(duì)列異常。在事件處理過程中出異常時，onError() 會被觸發(fā)，同時隊(duì)列自動終止，不允許再有事件發(fā)出。
在一個正確運(yùn)行的事件序列中, onCompleted() 和 onError() 有且只有一個，并且是事件序列中的最后一個。需要注意的是，onCompleted() 和 onError() 二者也是互斥的，即在隊(duì)列中調(diào)用了其中一個，就不應(yīng)該再調(diào)用另一個。
RxJava 的觀察者模式大致如下圖：
2. 基本實(shí)現(xiàn)基于以上的概念， RxJava 的基本實(shí)現(xiàn)主要有三點(diǎn)：
Observer 即觀察者，它決定事件觸發(fā)的時候?qū)⒂性鯓拥男袨椤?RxJava 中的 Observer 接口的實(shí)現(xiàn)方式：
Observerobserver = new Observer () { @Override public void onNext(String s) { Log.d(tag, "Item: " + s); } @Override public void onCompleted() { Log.d(tag, "Completed!"); } @Override public void onError(Throwable e) { Log.d(tag, "Error!"); } };
除了 Observer 接口之外，RxJava 還內(nèi)置了一個實(shí)現(xiàn)了 Observer 的抽象類：Subscriber。 Subscriber 對 Observer 接口進(jìn)行了一些擴(kuò)展，但他們的基本使用方式是完全一樣的：
Subscribersubscriber = new Subscriber () { @Override public void onNext(String s) { Log.d(tag, "Item: " + s); } @Override public void onCompleted() { Log.d(tag, "Completed!"); } @Override public void onError(Throwable e) { Log.d(tag, "Error!"); } };
不僅基本使用方式一樣，實(shí)質(zhì)上，在 RxJava 的 subscribe 過程中，Observer 也總是會先被轉(zhuǎn)換成一個 Subscriber 再使用。所以如果你只想使用基本功能，選擇 Observer 和 Subscriber 是完全一樣的。它們的區(qū)別對于使用者來說主要有兩點(diǎn)：
onStart(): 這是 Subscriber 增加的方法。它會在 subscribe 剛開始，而事件還未發(fā)送之前被調(diào)用，可以用于做一些準(zhǔn)備工作，例如數(shù)據(jù)的清零或重置。這是一個可選方法，默認(rèn)情況下它的實(shí)現(xiàn)為空。需要注意的是，如果對準(zhǔn)備工作的線程有要求（例如彈出一個顯示進(jìn)度的對話框，這必須在主線程執(zhí)行）， onStart() 就不適用了，因?yàn)樗偸窃?subscribe 所發(fā)生的線程被調(diào)用，而不能指定線程。要在指定的線程來做準(zhǔn)備工作，可以使用 doOnSubscribe() 方法，具體可以在后面的文中看到。
unsubscribe(): 這是 Subscriber 所實(shí)現(xiàn)的另一個接口 Subscription 的方法，用于取消訂閱。在這個方法被調(diào)用后，Subscriber 將不再接收事件。一般在這個方法調(diào)用前，可以使用 isUnsubscribed() 先判斷一下狀態(tài)。 unsubscribe() 這個方法很重要，因?yàn)樵?subscribe() 之后， Observable 會持有 Subscriber 的引用，這個引用如果不能及時被釋放，將有內(nèi)存泄露的風(fēng)險。所以最好保持一個原則：要在不再使用的時候盡快在合適的地方（例如 onPause() onStop() 等方法中）調(diào)用 unsubscribe() 來解除引用關(guān)系，以避免內(nèi)存泄露的發(fā)生。
Observable 即被觀察者，它決定什么時候觸發(fā)事件以及觸發(fā)怎樣的事件。 RxJava 使用 create() 方法來創(chuàng)建一個 Observable ，并為它定義事件觸發(fā)規(guī)則：
Observable observable = Observable.create(new Observable.OnSubscribe() { @Override public void call(Subscriber subscriber) { subscriber.onNext("Hello"); subscriber.onNext("Hi"); subscriber.onNext("Aloha"); subscriber.onCompleted(); } });
可以看到，這里傳入了一個 OnSubscribe 對象作為參數(shù)。OnSubscribe 會被存儲在返回的 Observable 對象中，它的作用相當(dāng)于一個計(jì)劃表，當(dāng) Observable 被訂閱的時候，OnSubscribe 的 call() 方法會自動被調(diào)用，事件序列就會依照設(shè)定依次觸發(fā)（對于上面的代碼，就是觀察者Subscriber 將會被調(diào)用三次 onNext() 和一次 onCompleted()）。這樣，由被觀察者調(diào)用了觀察者的回調(diào)方法，就實(shí)現(xiàn)了由被觀察者向觀察者的事件傳遞，即觀察者模式。
這個例子很簡單：事件的內(nèi)容是字符串，而不是一些復(fù)雜的對象；事件的內(nèi)容是已經(jīng)定好了的，而不像有的觀察者模式一樣是待確定的（例如網(wǎng)絡(luò)請求的結(jié)果在請求返回之前是未知的）；所有事件在一瞬間被全部發(fā)送出去，而不是夾雜一些確定或不確定的時間間隔或者經(jīng)過某種觸發(fā)器來觸發(fā)的。總之，這個例子看起來毫無實(shí)用價值。但這是為了便于說明，實(shí)質(zhì)上只要你想，各種各樣的事件發(fā)送規(guī)則你都可以自己來寫。至于具體怎么做，后面都會講到，但現(xiàn)在不行。只有把基礎(chǔ)原理先說明白了，上層的運(yùn)用才能更容易說清楚。
create() 方法是 RxJava 最基本的創(chuàng)造事件序列的方法。基于這個方法， RxJava 還提供了一些方法用來快捷創(chuàng)建事件隊(duì)列，例如：
just(T...): 將傳入的參數(shù)依次發(fā)送出來。
Observable observable = Observable.just("Hello", "Hi", "Aloha");
// 將會依次調(diào)用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();
from(T[]) / from(Iterable) : 將傳入的數(shù)組或 Iterable 拆分成具體對象后，依次發(fā)送出來。
String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
// 將會依次調(diào)用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();
上面 just(T...) 的例子和 from(T[]) 的例子，都和之前的 create(OnSubscribe) 的例子是等價的。
創(chuàng)建了 Observable 和 Observer 之后，再用 subscribe() 方法將它們聯(lián)結(jié)起來，整條鏈子就可以工作了。代碼形式很簡單：
observable.subscribe(observer); // 或者： observable.subscribe(subscriber);
有人可能會注意到， subscribe() 這個方法有點(diǎn)怪：它看起來是『observalbe 訂閱了 observer / subscriber』而不是『observer / subscriber 訂閱了 observalbe』，這看起來就像『雜志訂閱了讀者』一樣顛倒了對象關(guān)系。這讓人讀起來有點(diǎn)別扭，不過如果把 API 設(shè)計(jì)成 observer.subscribe(observable) / subscriber.subscribe(observable) ，雖然更加符合思維邏輯，但對流式 API 的設(shè)計(jì)就造成影響了，比較起來明顯是得不償失的。
Observable.subscribe(Subscriber) 的內(nèi)部實(shí)現(xiàn)是這樣的（僅核心代碼）：
// 注意：這不是 subscribe() 的源碼，而是將源碼中與性能、兼容性、擴(kuò)展性有關(guān)的代碼剔除后的核心代碼。
// 如果需要看源碼，可以去 RxJava 的 GitHub 倉庫下載。
public Subscription subscribe(Subscriber subscriber) {
subscriber.onStart();
onSubscribe.call(subscriber);
return subscriber;
}
可以看到，subscriber() 做了3件事：
調(diào)用 Subscriber.onStart() 。這個方法在前面已經(jīng)介紹過，是一個可選的準(zhǔn)備方法。
調(diào)用 Observable 中的 OnSubscribe.call(Subscriber) 。在這里，事件發(fā)送的邏輯開始運(yùn)行。從這也可以看出，在 RxJava 中， Observable 并不是在創(chuàng)建的時候就立即開始發(fā)送事件，而是在它被訂閱的時候，即當(dāng) subscribe() 方法執(zhí)行的時候。
將傳入的 Subscriber 作為 Subscription 返回。這是為了方便 unsubscribe().
整個過程中對象間的關(guān)系如下圖：
或者可以看動圖：
除了 subscribe(Observer) 和 subscribe(Subscriber) ，subscribe() 還支持不完整定義的回調(diào)，RxJava 會自動根據(jù)定義創(chuàng)建出 Subscriber 。形式如下：
Action1onNextAction = new Action1 () { // onNext() @Override public void call(String s) { Log.d(tag, s); } }; Action1 onErrorAction = new Action1 () { // onError() @Override public void call(Throwable throwable) { // Error handling } }; Action0 onCompletedAction = new Action0() { // onCompleted() @Override public void call() { Log.d(tag, "completed"); } }; // 自動創(chuàng)建 Subscriber ，并使用 onNextAction 來定義 onNext() observable.subscribe(onNextAction); // 自動創(chuàng)建 Subscriber ，并使用 onNextAction 和 onErrorAction 來定義 onNext() 和 onError() observable.subscribe(onNextAction, onErrorAction); // 自動創(chuàng)建 Subscriber ，并使用 onNextAction、 onErrorAction 和 onCompletedAction 來定義 onNext()、 onError() 和 onCompleted() observable.subscribe(onNextAction, onErrorAction, onCompletedAction);
簡單解釋一下這段代碼中出現(xiàn)的 Action1 和 Action0。 Action0 是 RxJava 的一個接口，它只有一個方法 call()，這個方法是無參無返回值的；由于 onCompleted() 方法也是無參無返回值的，因此 Action0 可以被當(dāng)成一個包裝對象，將 onCompleted() 的內(nèi)容打包起來將自己作為一個參數(shù)傳入 subscribe() 以實(shí)現(xiàn)不完整定義的回調(diào)。這樣其實(shí)也可以看做將 onCompleted() 方法作為參數(shù)傳進(jìn)了 subscribe()，相當(dāng)于其他某些語言中的『閉包』。 Action1 也是一個接口，它同樣只有一個方法 call(T param)，這個方法也無返回值，但有一個參數(shù)；與 Action0 同理，由于 onNext(T obj) 和 onError(Throwable error) 也是單參數(shù)無返回值的，因此 Action1 可以將 onNext(obj) 和 onError(error) 打包起來傳入 subscribe() 以實(shí)現(xiàn)不完整定義的回調(diào)。事實(shí)上，雖然 Action0 和 Action1 在 API 中使用最廣泛，但 RxJava 是提供了多個 ActionX 形式的接口 (例如 Action2, Action3) 的，它們可以被用以包裝不同的無返回值的方法。
注：正如前面所提到的，Observer 和 Subscriber 具有相同的角色，而且 Observer 在 subscribe() 過程中最終會被轉(zhuǎn)換成 Subscriber 對象，因此，從這里開始，后面的描述我將用 Subscriber 來代替 Observer ，這樣更加嚴(yán)謹(jǐn)。
下面舉兩個例子：
為了把原理用更清晰的方式表述出來，本文中挑選的都是功能盡可能簡單的例子，以至于有些示例代碼看起來會有『畫蛇添足』『明明不用 RxJava 可以更簡便地解決問題』的感覺。當(dāng)你看到這種情況，不要覺得是因?yàn)?RxJava 太啰嗦，而是因?yàn)樵谶^早的時候舉出真實(shí)場景的例子并不利于原理的解析，因此我刻意挑選了簡單的情景。
將字符串?dāng)?shù)組 names 中的所有字符串依次打印出來：
String[] names = ...;
Observable.from(names)
.subscribe(new Action1() {
@Override
public void call(String name) {
Log.d(tag, name);
}
});
由指定的一個 drawable 文件 id drawableRes 取得圖片，并顯示在 ImageView 中，并在出現(xiàn)異常的時候打印 Toast 報錯：
int drawableRes = ...; ImageView imageView = ...; Observable.create(new OnSubscribe() { @Override public void call(Subscriber subscriber) { Drawable drawable = getTheme().getDrawable(drawableRes)); subscriber.onNext(drawable); subscriber.onCompleted(); } }).subscribe(new Observer () { @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(); } });
正如上面兩個例子這樣，創(chuàng)建出 Observable 和 Subscriber ，再用 subscribe() 將它們串起來，一次 RxJava 的基本使用就完成了。非常簡單。
然而，
在 RxJava 的默認(rèn)規(guī)則中，事件的發(fā)出和消費(fèi)都是在同一個線程的。也就是說，如果只用上面的方法，實(shí)現(xiàn)出來的只是一個同步的觀察者模式。觀察者模式本身的目的就是『后臺處理，前臺回調(diào)』的異步機(jī)制，因此異步對于 RxJava 是至關(guān)重要的。而要實(shí)現(xiàn)異步，則需要用到 RxJava 的另一個概念： Scheduler 。
3. 線程控制 —— Scheduler (一)在不指定線程的情況下， RxJava 遵循的是線程不變的原則，即：在哪個線程調(diào)用 subscribe()，就在哪個線程生產(chǎn)事件；在哪個線程生產(chǎn)事件，就在哪個線程消費(fèi)事件。如果需要切換線程，就需要用到 Scheduler （調(diào)度器）。
在RxJava 中，Scheduler ——調(diào)度器，相當(dāng)于線程控制器，RxJava 通過它來指定每一段代碼應(yīng)該運(yùn)行在什么樣的線程。RxJava 已經(jīng)內(nèi)置了幾個 Scheduler ，它們已經(jīng)適合大多數(shù)的使用場景：
Schedulers.immediate(): 直接在當(dāng)前線程運(yùn)行，相當(dāng)于不指定線程。這是默認(rèn)的 Scheduler。
Schedulers.newThread(): 總是啟用新線程，并在新線程執(zhí)行操作。
Schedulers.io(): I/O 操作（讀寫文件、讀寫數(shù)據(jù)庫、網(wǎng)絡(luò)信息交互等）所使用的 Scheduler。行為模式和 newThread() 差不多，區(qū)別在于 io() 的內(nèi)部實(shí)現(xiàn)是是用一個無數(shù)量上限的線程池，可以重用空閑的線程，因此多數(shù)情況下 io() 比 newThread() 更有效率。不要把計(jì)算工作放在 io() 中，可以避免創(chuàng)建不必要的線程。
Schedulers.computation(): 計(jì)算所使用的 Scheduler。這個計(jì)算指的是 CPU 密集型計(jì)算，即不會被 I/O 等操作限制性能的操作，例如圖形的計(jì)算。這個 Scheduler 使用的固定的線程池，大小為 CPU 核數(shù)。不要把 I/O 操作放在 computation() 中，否則 I/O 操作的等待時間會浪費(fèi) CPU。
另外， Android 還有一個專用的 AndroidSchedulers.mainThread()，它指定的操作將在 Android 主線程運(yùn)行。
有了這幾個 Scheduler ，就可以使用 subscribeOn() 和 observeOn() 兩個方法來對線程進(jìn)行控制了。 subscribeOn(): 指定 subscribe() 所發(fā)生的線程，即 Observable.OnSubscribe 被激活時所處的線程。或者叫做事件產(chǎn)生的線程。 observeOn(): 指定 Subscriber 所運(yùn)行在的線程。或者叫做事件消費(fèi)的線程。
文字?jǐn)⑹隹倸w難理解，上代碼：
Observable.just(1, 2, 3, 4)
.subscribeOn(Schedulers.io()) // 指定 subscribe() 發(fā)生在 IO 線程
.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回調(diào)發(fā)生在主線程
.subscribe(new Action1() {
@Override
public void call(Integer number) {
Log.d(tag, "number:" + number);
}
});
上面這段代碼中，由于 subscribeOn(Schedulers.io()) 的指定，被創(chuàng)建的事件的內(nèi)容 1、2、3、4 將會在 IO 線程發(fā)出；而由于 observeOn(AndroidScheculers.mainThread()) 的指定，因此 subscriber 數(shù)字的打印將發(fā)生在主線程 。事實(shí)上，這種在 subscribe() 之前寫上兩句 subscribeOn(Scheduler.io()) 和 observeOn(AndroidSchedulers.mainThread()) 的使用方式非常常見，它適用于多數(shù)的 『后臺線程取數(shù)據(jù)，主線程顯示』的程序策略。
而前面提到的由圖片 id 取得圖片并顯示的例子，如果也加上這兩句：
int drawableRes = ...; ImageView imageView = ...; Observable.create(new OnSubscribe() { @Override public void call(Subscriber subscriber) { Drawable drawable = getTheme().getDrawable(drawableRes)); subscriber.onNext(drawable); subscriber.onCompleted(); } }) .subscribeOn(Schedulers.io()) // 指定 subscribe() 發(fā)生在 IO 線程 .observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回調(diào)發(fā)生在主線程 .subscribe(new Observer () { @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(); } });
那么，加載圖片將會發(fā)生在 IO 線程，而設(shè)置圖片則被設(shè)定在了主線程。這就意味著，即使加載圖片耗費(fèi)了幾十甚至幾百毫秒的時間，也不會造成絲毫界面的卡頓。
RxJava 的 Scheduler API 很方便，也很神奇（加了一句話就把線程切換了，怎么做到的？而且 subscribe() 不是最外層直接調(diào)用的方法嗎，它竟然也能被指定線程？）。然而 Scheduler 的原理需要放在后面講，因?yàn)樗脑硎且韵乱还?jié)《變換》的原理作為基礎(chǔ)的。
好吧這一節(jié)其實(shí)我屁也沒說，只是為了讓你安心，讓你知道我不是忘了講原理，而是把它放在了更合適的地方。
4. 變換終于要到牛逼的地方了，不管你激動不激動，反正我是激動了。
RxJava 提供了對事件序列進(jìn)行變換的支持，這是它的核心功能之一，也是大多數(shù)人說『RxJava 真是太好用了』的最大原因。所謂變換，就是將事件序列中的對象或整個序列進(jìn)行加工處理，轉(zhuǎn)換成不同的事件或事件序列。概念說著總是模糊難懂的，來看 API。
首先看一個 map() 的例子：
Observable.just("images/logo.png") // 輸入類型 String
.map(new Func1() {
@Override
public Bitmap call(String filePath) { // 參數(shù)類型 String
return getBitmapFromPath(filePath); // 返回類型 Bitmap
}
})
.subscribe(new Action1() {
@Override
public void call(Bitmap bitmap) { // 參數(shù)類型 Bitmap
showBitmap(bitmap);
}
});
這里出現(xiàn)了一個叫做 Func1 的類。它和 Action1 非常相似，也是 RxJava 的一個接口，用于包裝含有一個參數(shù)的方法。 Func1 和 Action 的區(qū)別在于， Func1 包裝的是有返回值的方法。另外，和 ActionX 一樣， FuncX 也有多個，用于不同參數(shù)個數(shù)的方法。FuncX 和 ActionX 的區(qū)別在 FuncX 包裝的是有返回值的方法。
可以看到，map() 方法將參數(shù)中的 String 對象轉(zhuǎn)換成一個 Bitmap 對象后返回，而在經(jīng)過 map() 方法后，事件的參數(shù)類型也由 String 轉(zhuǎn)為了 Bitmap。這種直接變換對象并返回的，是最常見的也最容易理解的變換。不過 RxJava 的變換遠(yuǎn)不止這樣，它不僅可以針對事件對象，還可以針對整個事件隊(duì)列，這使得 RxJava 變得非常靈活。我列舉幾個常用的變換：
map(): 事件對象的直接變換，具體功能上面已經(jīng)介紹過。它是 RxJava 最常用的變換。 map() 的示意圖：
flatMap(): 這是一個很有用但非常難理解的變換，因此我決定花多些篇幅來介紹它。 首先假設(shè)這么一種需求：假設(shè)有一個數(shù)據(jù)結(jié)構(gòu)『學(xué)生』，現(xiàn)在需要打印出一組學(xué)生的名字。實(shí)現(xiàn)方式很簡單：
