編輯：關於Android編程
終究沒有經受住RxJava的誘惑，只恨自己來的比較晚，走起~
一個在 Java VM 上使用可觀測的序列來組成異步的、基於事件的程序的庫~
我們先來看一下這坨代碼：
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如何實現呢？
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);
}
});
吾皇勿怒，臣妾知道代碼並沒有減少，可是臣妾說的是邏輯簡單，鏈式結構非常清晰不是嗎？
相信您在頓悟了5s中之後也看到了其好處~
1、擴展的觀察者模式
RxJava 的異步實現，是通過一種擴展的觀察者模式來實現的，A 對象（觀察者）對 B 對象（被觀察者）的某種變化高度敏感，需要在 B 變化的一瞬間做出反應。
2、RxJava的四個基本概念：
Observable (可觀察者，即被觀察者) Observer (觀察者) subscribe (訂閱) 事件 ： onNext()、onCompleted() 和 onError()事件
onNext(): 普通事件，每個事件執行之後的event。 onCompleted(): 事件隊列完結。RxJava 不僅把每個事件單獨處理，還會把它們看做一個隊列。RxJava 規定，當不會再有新的onNext() 發出時，需要觸發 onCompleted() 方法作為標志。 onError(): 事件隊列異常。在事件處理過程中出異常時，onError() 會被觸發，同時隊列自動終止，不允許再有事件發出。
注意 ：
在一個正確運行的事件序列中, onCompleted() 和 onError() 有且只有一個，並且是事件序列中的最後一個。需要注意的是，onCompleted() 和 onError() 二者也是互斥的，即在隊列中調用了其中一個，就不應該再調用另一個。
![RxJava觀察者模式](https://www.android5.online/Android/UploadFiles_5356/201702/2017022217194644.png "\")
基於以上的概念， RxJava 的基本實現主要有三點：
1) 創建 Observer
Observer 即觀察者，它決定事件觸發的時候將有怎樣的行為。 RxJava 中的 Observer 接口的實現方式：
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 還內置了一個實現了 Observer 的抽象類：Subscriber。 Subscriber 對 Observer 接口進行了一些擴展，但他們的基本使用方式是完全一樣的：
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!"); }};
不僅基本使用方式一樣，實質上，在 RxJava 的 subscribe 過程中，Observer 也總是會先被轉換成一個 Subscriber 再使用。所以如果你只想使用基本功能，選擇 Observer 和 Subscriber 是完全一樣的。它們的區別對於使用者來說主要有兩點：
onStart(): 這是 Subscriber 增加的方法。它會在 subscribe 剛開始，而事件還未發送之前被調用，可以用於做一些准備工作，例如數據的清零或重置。這是一個可選方法，默認情況下它的實現為空。需要注意的是，如果對准備工作的線程有要求（例如彈出一個顯示進度的對話框，這必須在主線程執行）， onStart() 就不適用了，因為它總是在 subscribe 所發生的線程被調用，而不能指定線程。要在指定的線程來做准備工作，可以使用 doOnSubscribe() 方法，具體可以在後面的文中看到。
unsubscribe(): 這是 Subscriber 所實現的另一個接口 Subscription 的方法，用於取消訂閱。在這個方法被調用後，Subscriber 將不再接收事件。一般在這個方法調用前，可以使用 isUnsubscribed() 先判斷一下狀態。 unsubscribe() 這個方法很重要，因為在 subscribe() 之後， Observable 會持有 Subscriber 的引用，這個引用如果不能及時被釋放，將有內存洩露的風險。所以最好保持一個原則：要在不再使用的時候盡快在合適的地方（例如 onPause() onStop() 等方法中）調用unsubscribe() 來解除引用關系，以避免內存洩露的發生。
2) 創建 Observable
Observable 即被觀察者，它決定什麼時候觸發事件以及觸發怎樣的事件。 RxJava 使用 create() 方法來創建一個 Observable ，並為它定義事件觸發規則：
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 對象作為參數。OnSubscribe 會被存儲在返回的 Observable 對象中，它的作用相當於一個計劃表，當 Observable 被訂閱的時候，OnSubscribe 的 call() 方法會自動被調用，事件序列就會依照設定依次觸發（對於上面的代碼，就是觀察者Subscriber 將會被調用三次 onNext() 和一次 onCompleted()）。這樣，由被觀察者調用了觀察者的回調方法，就實現了由被觀察者向觀察者的事件傳遞，即觀察者模式。
這個例子很簡單：事件的內容是字符串，而不是一些復雜的對象；事件的內容是已經定好了的，而不像有的觀察者模式一樣是待確定的（例如網絡請求的結果在請求返回之前是未知的）；所有事件在一瞬間被全部發送出去，而不是夾雜一些確定或不確定的時間間隔或者經過某種觸發器來觸發的。總之，這個例子看起來毫無實用價值。但這是為了便於說明，實質上只要你想，各種各樣的事件發送規則你都可以自己來寫。至於具體怎麼做，後面都會講到，但現在不行。只有把基礎原理先說明白了，上層的運用才能更容易說清楚。
create() 方法是 RxJava 最基本的創造事件序列的方法。基於這個方法， RxJava 還提供了一些方法用來快捷創建事件隊列，例如：
just(T…): 將傳入的參數依次發送出來。
Observable observable = Observable.just("Hello", "Hi", "Aloha");
// 將會依次調用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();
from(T[]) / from(Iterable
在不指定線程的情況下， 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() {
@Override
public 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() { @Override public void call(Subscriber subscriber) { Drawable drawable = getTheme().getDrawable(drawableRes)); subscriber.onNext(drawable); subscriber.onCompleted(); } }) .subscribeOn(Schedulers.io()) // 指定 subscribe() 發生在 IO 線程 .observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回調發生在主線程 .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(); } });
那麼，加載圖片將會發生在 IO 線程，而設置圖片則被設定在了主線程。這就意味著，即使加載圖片耗費了幾十甚至幾百毫秒的時間，也不會造成絲毫界面的卡頓。
2) Scheduler 的原理基礎
RxJava 的 Scheduler API 很方便，也很神奇（加了一句話就把線程切換了，怎麼做到的？而且 subscribe() 不是最外層直接調用的方法嗎，它竟然也能被指定線程？）。然而 Scheduler 的原理需要放在後面講，因為它的原理是以下一節《變換》的原理作為基礎的。
好吧這一節其實我屁也沒說，只是為了讓你安心，讓你知道我不是忘了講原理，而是把它放在了更合適的地方。
RxJava 提供了對事件序列進行變換的支持，這是它的核心功能之一，也是大多數人說『RxJava 真是太好用了』的最大原因。所謂變換，就是將事件序列中的對象或整個序列進行加工處理，轉換成不同的事件或事件序列。概念說著總是模糊難懂的，來看 API。
1) API
首先看一個 map() 的例子：
Observable.just("images/logo.png") // 輸入類型 String
.map(new Func1() {
@Override
public Bitmap call(String filePath) { // 參數類型 String
return getBitmapFromPath(filePath); // 返回類型 Bitmap
}
})
.subscribe(new Action1() {
@Override
public void call(Bitmap bitmap) { // 參數類型 Bitmap
showBitmap(bitmap);
}
});
這裡出現了一個叫做 Func1 的類。它和 Action1 非常相似，也是 RxJava 的一個接口，用於包裝含有一個參數的方法。 Func1 和Action 的區別在於， Func1 包裝的是有返回值的方法。另外，和 ActionX 一樣， FuncX 也有多個，用於不同參數個數的方法。FuncX和 ActionX 的區別在 FuncX 包裝的是有返回值的方法。
可以看到，map() 方法將參數中的 String 對象轉換成一個 Bitmap 對象後返回，而在經過 map() 方法後，事件的參數類型也由 String轉為了 Bitmap。這種直接變換對象並返回的，是最常見的也最容易理解的變換。不過 RxJava 的變換遠不止這樣，它不僅可以針對事件對象，還可以針對整個事件隊列，這使得 RxJava 變得非常靈活。我列舉幾個常用的變換：
map(): 事件對象的直接變換，具體功能上面已經介紹過。它是 RxJava 最常用的變換。
map() 示意圖：
vcq9uty88rWlo7o8L2NvZGU+PC9jb2RlPjwvY29kZT48L2NvZGU+PC9jb2RlPjwvY29kZT48L2NvZGU+PC9wPg0KPHByZSBjbGFzcz0="brush:java;">
Student[] students = ...;
Subscriber
很簡單。那麼再假設：如果要打印出每個學生所需要修的所有課程的名稱呢？（需求的區別在於，每個學生只有一個名字，但卻有多個課程。）首先可以這樣實現：
Student[] students = ...; Subscribersubscriber = new Subscriber () { @Override public void onNext(Student student) { List 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);
依然很簡單。那麼如果我不想在 Subscriber 中使用 for 循環，而是希望 Subscriber 中直接傳入單個的 Course 對象呢（這對於代碼復用很重要）？用 map() 顯然是不行的，因為 map() 是一對一的轉化，而我現在的要求是一對多的轉化。那怎麼才能把一個 Student 轉化成多個 Course 呢？
這個時候，就需要用 flatMap() 了：
![flatMap 示意圖](https://www.android5.online/Android/UploadFiles_5356/201702/2017022217194729.png "\")
![lift 原理圖](https://www.android5.online/Android/UploadFiles_5356/201702/2017022217194799.png "\")
![兩次和多次的 lift](https://www.android5.online/Android/UploadFiles_5356/201702/2017022217194816.png "\")
![subscribeOn 原理圖](https://www.android5.online/Android/UploadFiles_5356/201702/2017022217194995.png "\")
![observeOn 原理](https://www.android5.online/Android/UploadFiles_5356/201702/2017022217195024.png "\")
![線程控制綜合調用](https://www.android5.online/Android/UploadFiles_5356/201702/2017022217195005.png "\")
