Android 開發RxJava 詳解
寫在前面：
感謝拋物線寫出了這篇經典的Rxjava入門的文章1. 概念：擴充套件的觀察者模式
RxJava 的非同步實現，是通過一種擴充套件的觀察者模式來實現的。
觀察者模式
先簡述一下觀察者模式，已經熟悉的可以跳過這一段。
觀察者模式面向的需求是：A 物件（觀察者）對 B 物件（被觀察者）的某種變化高度敏感，需要在 B 變化的一瞬間做出反應。舉個例子，新聞裡喜聞樂見的警察抓小偷，警察需要在小偷伸手作案的時候實施抓捕。在這個例子裡，警察是觀察者，小偷是被觀察者，警察需要時刻盯著小偷的一舉一動，才能保證不會漏過任何瞬間。程式的觀察者模式和這種真正的『觀察』略有不同，觀察者不需要時刻盯著被觀察者（例如 A 不需要每過 2ms 就檢查一次 B 的狀態），而是採用註冊
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 =newObserver<String>(){@Overridepublicvoid onNext(String s){Log.d(tag,"Item: "+ s);}@Overridepublicvoid onCompleted(){Log.d(tag,"Completed!");}@Overridepublicvoid onError(Throwable e){Log.d(tag,"Error!");}};
除了 Observer 介面之外，RxJava 還內建了一個實現了 Observer 的抽象類：Subscriber。 Subscriber 對 Observer 介面進行了一些擴充套件，但他們的基本使用方式是完全一樣的：
Subscriber<String> subscriber =newSubscriber<String>(){@Overridepublicvoid onNext(String s){Log.d(tag,"Item: "+ s);}@Overridepublicvoid onCompleted(){Log.d(tag,"Completed!");}@Overridepublicvoid 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(newObservable.OnSubscribe<String>(){@Overridepublicvoid call(Subscriber<?superString> 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<? extends T>): 將傳入的陣列或Iterable拆分成具體物件後，依次傳送出來。
String[] words ={"Hello","Hi","Aloha"};Observable observable =Observable.from(words);// 將會依次呼叫：// onNext("Hello");// onNext("Hi");// onNext("Aloha");// onCompleted();
上面 just(T...) 的例子和 from(T[]) 的例子，都和之前的 create(OnSubscribe) 的例子是等價的。
3) Subscribe (訂閱)
建立了 Observable 和 Observer 之後，再用 subscribe() 方法將它們聯結起來，整條鏈子就可以工作了。程式碼形式很簡單：
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 倉庫下載。publicSubscription subscribe(Subscriber subscriber){
subscriber.onStart();
onSubscribe.call(subscriber);return subscriber;}
可以看到，subscriber() 做了3件事：
- 呼叫
Subscriber.onStart()。這個方法在前面已經介紹過，是一個可選的準備方法。 - 呼叫
Observable中的OnSubscribe.call(Subscriber)。在這裡，事件傳送的邏輯開始執行。從這也可以看出，在 RxJava 中，Observable並不是在建立的時候就立即開始傳送事件，而是在它被訂閱的時候，即當subscribe()方法執行的時候。 - 將傳入的
Subscriber作為Subscription返回。這是為了方便unsubscribe().
整個過程中物件間的關係如下圖：
或者可以看動圖：
除了 subscribe(Observer) 和 subscribe(Subscriber) ，subscribe() 還支援不完整定義的回撥，RxJava
會自動根據定義創建出 Subscriber 。形式如下：
Action1<String> onNextAction =newAction1<String>(){// onNext()@Overridepublicvoid call(String s){Log.d(tag, s);}};Action1<Throwable> onErrorAction =newAction1<Throwable>(){// onError()@Overridepublicvoid