因為工作需要剛好在學習 RxJava + Retrofit2 + OkHttp3 網絡請求框架，網上搜了一些 RxJava 的教程，並不是很好理解，所幸最後我找到了幾篇有助於初學者瞭解 RxJava 的文章，於是結合自己的理解，重新整理成一篇發給大家，希望通過我的咀嚼，能夠幫助大家更快的瞭解和上手 RxJava，尤其是文章的最後，你將理解是怎樣一種優勢，乃至於我們開始考慮用 RxJava 來替代傳統方式。話不多說，以下正文。

觀察者模式

RxJava 涉及到所謂的觀察者模式。

觀察者模式，按我的理解，就是所謂的被觀察者和訂閱者之間的關係。

其形式是，在作為被觀察者的一個方法中，包含著一段訂閱，該訂閱的本質是抽象類或接口的引用對抽象方法的調用，而該訂閱實際指向的是作為實現類的訂閱者具體實現的方法。

（這實際上就是用到了面向對象編程中“向上轉型”的概念。向上轉型的套路是：實現類實現了抽象類定義的抽象方法；將實現類的對象賦值給抽象類的引用；然後通過抽象類的引用來調用抽象方法，而該調用實際指向實現類具體實現的方法……好吧，我這裡相當於把面向對象的基礎又過了一遍，不懂向上轉型概念的先滾去看 Mars 的 Java4Android 教程 - - ）

//抽象類或接口
public interface Printer{
 void print(String s);//定義抽象函數
}
//實現類實現抽象函數
class CanonPrinter implement Printer{
 @override
 public void print(String s){
 System.out.println(s);
 }
}
//在被觀察者中，通過抽象類或接口的引用來調用抽象函數
private static void printSth(Printer printer){
 printer.print("ha ha ha");
}
//執行被觀察者時，傳入實現類的對象作為實參賦值給抽象類或接口的引用，
//從而在被觀察者中通過抽象類或接口的引用來調用print抽象方法，
//而該抽象方法實際指向實現類實現的print方法
public static void main(String[] args){
 printSth(new CanonPrinter());
}

所以若是真要理解成觀察和被觀察，按照這個邏輯也的確說的過去，也就是說，被觀察者剛好觸發了這裡，然後直接跳轉到訂閱者實現的方法內部去具體執行被觸發後該做出的反應。

被觀察者

Observable<String> myObservable = Observable.create( 
 new Observable.OnSubscribe<String>() { 
 @Override 
 public void call(Subscriber<? super String> sub) { 
 sub.onNext("aaa"); 
 sub.onNext("bbb");
 sub.onCompleted(); 
 } 
 } 
); 

訂閱者

Subscriber<String> mySubscriber = new Subscriber<String>() { 
 @Override 
 public void onNext(String s) { System.out.println(s); } 
 @Override 
 public void onCompleted() { } 
 @Override 
 public void onError(Throwable e) { } 
}; 

關聯起來

myObservable.subscribe(mySubscriber);

有沒有覺得和

btn.setOnClickListener(onClickListener)很相似，是的，就是這樣。

這一步就是將實現類的對象賦值給抽象類或接口的引用。

不過，觀察者和訂閱者的創建也有簡化版。

例如當觀察者只觸發一個事件（onNext）時，可以直接使用 just。

觀察者

Observable<String> myObservable = Observable.just("Hello, world!"); 

訂閱者

當不關心 OnComplete 和 OnError，只需要在 onNext 時做一些處理，可以用 Action1 類

Action1<String> onNextAction = new Action1<String>() { 
 @Override 
 public void call(String s) { 
 System.out.println(s); 
 } 
}; 

subscribe 方法有一個重載版本，接受1~3個 Action1 類型的參數，分別對應 OnNext，OnError，OnComplete

然後我們現在只需要 onNext，就只需要傳入一個參數

myObservable.subscribe(onNextAction);

這樣，上述的代碼最終可以寫成

Observable.just("aaa") 
 .subscribe(new Action1<String>() { 
 @Override 
 public void call(String s) { 
 System.out.println(s); 
 } 
 }); 

注意，泛型類不能直接寫在類中作為對象的實例變量，當它沒有被指定類型時，因為泛型是編譯時，不是運行時。上述代碼因為 Observable 不指定具體<T>，因此，你應寫在方法中，使 Observable 作為局部變量。

而使用 lambda，可以使代碼變得更簡潔

Observable.just("aaa") 
 .subscribe(s -> System.out.println(s)); 

那如果我想做一些手腳怎麼辦，特別是，觀察者不能被改動，訂閱者也不能被改動的時候

這時我們可以通過操作符 map 來完成改動（操作符還有很多，負責中間過程改動的，是map）

例如我給輸入的文字加個後綴

Observable.just("aaa") 
 .map(new Func1<String, String>() { 
 @Override 
 public String call(String s) { 
 return s + ".jpg"; 
 } 
 }) 
 .subscribe(s -> System.out.println(s));

lambda：

Observable.just("aaa") 
 .map(s -> s + ".jpg") 
 .subscribe(s -> System.out.println(s));

或者在中間過程獲得類型不同的數據

Observable.just("aaa") 
 .map(s -> s.hashCode()) 
 .map(i -> Integer.toString(i)) 
 .subscribe(s -> System.out.println(s));

那麼到這裡為止，我們知道了，RxJava 本質上和我們之前用的“回調”性質是一樣的，就是作為觀察者的一方調用抽象函數，作為訂閱者的另一方實現抽象函數，並且二者通過 subscribe 方法關聯，也就是，在觀察者的 subscribe 方法中實現訂閱者的匿名內部類，來具體實現被觸發的方法。

但是故事說到這裡，好像還是沒有異步什麼事啊，它分明就是“同線程中的身首異處”嘛。

下面來看看操作符

操作符

上述我們已經提到了操作符 map。

然後此處我們有個這樣的需求：輸入一個關鍵字，返回相關結果的 url 列表。相當於搜索引擎。

一般我們會這麼做

Observable<List<String>> query(String text);
query("Hello, world!") 
 .subscribe(urls -> { 
 for (String url : urls) { 
 System.out.println(url); 
 } 
 }); 

有for循環，這樣看起來有點繁雜

我們可以通過操作符 from，對數組中的數據進行逐個處理

Observable.from(urls) 
 .subscribe(url -> System.out.println(url));

也就是

query("Hello, world!") 
 .subscribe(urls -> { 
 Observable.from(urls) 
 .subscribe(url -> System.out.println(url)); 
 }); 

但這樣破壞了 RxJava 的結構，因為訂閱者最好是保持本身單一的功能，而數據的改變最好在中間過程中通過操作符來完成。

此時我們可以通過操作符 flatMap 來處理這個問題。

Observable.flatMap() 接收一個 Observable 的輸出作為輸入，同時輸出另外一個Observable

query("Hello, world!") 
 .flatMap(new Func1<List<String>, Observable<String>>() { 
 @Override 
 public Observable<String> call(List<String> urls) { 
 return Observable.from(urls); 
 } 
 }) 
 .subscribe(url -> System.out.println(url));

lambda：

query("Hello, world!") 
 .flatMap(urls -> Observable.from(urls)) 
 .subscribe(url -> System.out.println(url)); 

flatMap 輸出的新的 Observable 正是我們在 Subscriber 想要接收的。現在 Subscriber 不再收到 List<String>，而是收到一些列單個的字符串，就像 Observable.from() 的輸出一樣

接著前面的例子，現在我不想打印 url 了，而是要打印收到的每個網站的標題。我當然不能直接將 getTitle 方法放進訂閱者， 因為一再強調了，訂閱者實現的功能要單一，不要輕易改動。那我將考慮在中間過程中來完成這個改動，這樣的話，我的 getTitle 方法每次只能傳入一個 url，並且返回值不是一個 String，而是一個輸出 String 的 Observable 對象。flatMap 不正是這麼運作的嗎（通過輸入的 Observable 對象得到裡面的數據，再將數據輸出，以 Observable 對象的方式傳遞），所以還是考慮使用 flatMap。

query("Hello, world!") 
 .flatMap(urls -> Observable.from(urls)) 
 .flatMap(new Func1<String, Observable<String>>() { 
 @Override 
 public Observable<String> call(String url) { 
 return getTitle(url); 
 } 
 }) 
 .subscribe(title -> System.out.println(title));

lambda：

query("Hello, world!") 
 .flatMap(urls -> Observable.from(urls)) 
 .flatMap(url -> getTitle(url)) 
 .subscribe(title -> System.out.println(title)); 

那麼如果我想過濾掉一些情況，例如返回 url 為 null 的，我不要它們的標題了，過濾掉，可以用操作符 filter。

query("Hello, world!") 
 .flatMap(urls -> Observable.from(urls)) 
 .flatMap(url -> getTitle(url)) 
 .filter(title -> title != null) 
 .subscribe(title -> System.out.println(title));

如果想指定輸出元素的數量，可以用 take。

以下限制輸出 5 個。

query("Hello, world!") 
 .flatMap(urls -> Observable.from(urls)) 
 .flatMap(url -> getTitle(url)) 
 .filter(title -> title != null) 
 .take(5) 
 .subscribe(title -> System.out.println(title)); 

如果想在每次觸及訂閱之前，完成一些事情，可以用 doOnNext。

例如以下在輸出之前，將標題保存到某處。

query("Hello, world!") 
 .flatMap(urls -> Observable.from(urls)) 
 .flatMap(url -> getTitle(url)) 
 .filter(title -> title != null) 
 .take(5) 
 .doOnNext(title -> saveTitle(title)) 
 .subscribe(title -> System.out.println(title)); 

其他的操作符還可以來這篇文章瞭解

https://mrfu.me/2016/01/10/RxWeekend/#tips7

介紹完操作符，現在開始講重點。

為什麼用 RxJava，它和傳統的回調以及異步處理相比存在什麼優勢？

錯誤處理

Observable.just("Hello, world!")
 .map(s -> potentialException(s))
 .map(s -> anotherPotentialException(s))
 .subscribe(new Subscriber<String>() {
 @Override
 public void onNext(String s) { System.out.println(s); }
 @Override
 public void onCompleted() { System.out.println("Completed!"); }
 @Override
 public void onError(Throwable e) { System.out.println("Ouch!"); }
 });

代碼中的 potentialException() 和 anotherPotentialException() 有可能會拋出異常。每一個 Observerable 對象在終結的時候都會調用 onCompleted() 或者 onError() 方法，所以 Demo 中會打印“Completed!”或者“Ouch!”

這種模式有以下幾個優點：

1.只要有異常發生 onError() 一定會被調用

這極大的簡化了錯誤處理。只需要在一個地方處理錯誤即可以。

2.操作符不需要處理異常

將異常處理交給訂閱者來做，Observerable 的操作符調用鏈中一旦有一個拋出了異常，就會直接執行 onError() 方法。

3.你能夠知道什麼時候訂閱者已經接收了全部的數據

知道什麼時候任務結束能夠幫助簡化代碼的流程。

這種錯誤處理方式比傳統的錯誤處理更簡單。傳統的錯誤處理中，通常是在每個回調中處理錯誤。這不僅導致了重複的代碼，並且意味著每個回調都必須知道如何處理錯誤，你的回調代碼將和調用者緊耦合在一起。

使用 RxJava，Observable 對象根本不需要知道如何處理錯誤！操作符也不需要處理錯誤狀態：一旦發生錯誤，就會跳過當前和後續的操作符。所有的錯誤處理都交給訂閱者來做。

（這裡的意思是說，傳統的方式，就算也存在中間過程，那錯誤也需要在中間過程的每一步中分別處理。而 RxJava 不需要在中間過程處理錯誤，有錯誤直接跳到最後，統一由訂閱者處理）

調度器

使用 RxJava，你可以使用 subscribeOn() 指定觀察者代碼運行的線程，使用 observerOn() 指定訂閱者運行的線程：

（這個過程簡單的說就是，訂閱者在主線程觀察被觀察者，所以這個動作被成為 observeOn。而被觀察者在異線程添加了訂閱者的訂閱，所以這個動作叫 subscribeOn，其實說 be subscribe on 更合適）

myObservableServices.retrieveImage(url)
 .subscribeOn(Schedulers.io())
 .observeOn(AndroidSchedulers.mainThread())
 .subscribe(bitmap -> myImageView.setImageBitmap(bitmap));

subscribeOn() 和 observerOn() 可以被添加到任何 Observable 對象上，這兩個也是操作符。我不需要關心 Observable 對象以及它上面有哪些操作符。僅僅運用這兩個操作符就可以實現在不同的線程中調度

如果使用 AsyncTask 或者其他類似的，我將不得不仔細設計我的代碼，找出需要併發執行的部分。使用 RxJava，我可以保持代碼不變，僅僅在需要併發的時候調用這兩個操作符就可以

訂閱

當調用 Observable.subscribe()，會返回一個 Subscription 對象。這個對象代表了被觀察者和訂閱者之間的聯繫

Subscription subscription = Observable.just("Hello, World!")
 .subscribe(s -> System.out.println(s));
subscription.unsubscribe();

RxJava 的另外一個好處就是它處理 unsubscribing 的時候，會停止整個調用鏈。如果你使用了一串很複雜的操作符，調用 unsubscribe 將會在他當前執行的地方終止。不需要做任何額外的工作！