引言：前幾天瀏覽了一下稀土 App，發現有個挺不錯的新聞 App 實戰實例。正好最近想學習一下完整項目的源碼（特別是后臺代碼，就是各種框架等等）。然后想起以前很多大牛都談起過 RxJava、Retrofit、Gson 等等框架，所以現在決定來學習學習這些常用的框架。

---

RxJava 到底是什么

　　一個詞概括：異步。說到底它就是一個實現異步操作的庫。那什么是異步操作，我之前在Android 消息傳遞機制這篇博文中講過 Android 是單線程模型。進程啟動時，也就是 App 啟動后默認就只有一個線程運行，而該線程就是我們說的主線程，也叫 UI 線程。顧名思義，UI 線程就是用來更新界面顯示，而如果運行當中具有耗時操作如網絡請求，數據庫讀寫，文件下載等這些耗時操作都需要在其他線程當中去完成，完成之后再更新在 UI 界面中顯示出來。這就是異步加載，而 Android 中我們有現成的方式去完成這一操作，如 AsyncTask 、Handler。那為什么還要用 RxJava？

RxJava 好在哪里

　　還是用一個詞概括：簡潔。異步操作很關鍵的一點是要注意程序的簡潔性，因為在調度過程比較復雜的情況下，異步代碼經常會既難寫也難被讀懂。 Android 創造的 AsyncTask 和 Handler ，其實都是為了讓異步代碼更加簡潔。RxJava 的優勢也是簡潔，但它的簡潔的與眾不同之處在于，隨著程序邏輯變得越來越復雜，它依然能夠保持簡潔。

　　假設有這樣一個需求：界面上有一個自定義的視圖 ImageCollectorView ，它的作用是顯示多張圖片，并能使用addImage(Bitmap)方法來任意增加顯示的圖片?，F在需要程序將一個給出的目錄數組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(){mImageCollcetorView.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){mImageCollectorView.addImage(bitmap);}});復制代碼

　　那看到這里有人可能說了這代碼敲得更多了，哪來的簡潔。注意，我們這里說的簡潔指的是邏輯上的簡潔，不是單純的代碼量少。觀察一下你會發現， RxJava 的這個實現，是一條從上到下的鏈式調用，沒有任何嵌套，這在邏輯的簡潔性上是具有優勢的。當需求變得復雜時，這種優勢將更加明顯。

RxJava 原理與使用簡析

原理

　　RxJava 的異步實現是通過一種通用概念下的觀察者模式來實現的。那么我們了解一下設計模式當中的觀察者模式。

　　觀察者模式面向的需求是：A 對象(觀察者)對 B 對象(被觀察者)的某種變化高度敏感，需要在 B 變化的一瞬間做出反應。舉個例子，新聞里喜聞樂見的警察抓小偷，警察需要在小偷伸手作案的時候實施抓捕。在這個例子里，警察是觀察者，小偷是被觀察者，警察需要時刻盯著小偷的一舉一動，才能保證不會漏過任何瞬間。程序的觀察者模式和這種真正的『觀察』略有不同，觀察者不需要時刻盯著被觀察者(例如 A 不需要每過 2ms 就檢查一次 B 的狀態)，而是采用注冊(Register)或者稱為訂閱(subscribe)的方式，告訴被觀察者：我需要你的某某狀態，你要在它變化的時候通知我。 Android 開發中一個比較典型的例子是點擊監聽器 OnClickListener 。對設置 OnClickListener 來說， View 是被觀察者， OnClickListener 是觀察者，二者通過setOnClickListener()方法達成訂閱關系。訂閱之后用戶點擊按鈕的瞬間，Android Framework 就會將點擊事件發送給已經注冊的 OnClickListener 。采取這樣被動的觀察方式，既省去了反復檢索狀態的資源消耗，也能夠得到最高的反饋速度。

　　RxJava 作為一個工具庫，使用的就是通用形式的觀察者模式。它有四個基本概念：Observable(被觀察者)、Observer(觀察者)、subscribe(訂閱)、事件。Observable與Observer通過subscribe訂閱這一動作聯系在一起，從而使Observable可以在需要的時候發出事件來通知Observer。RxJava 的事件回調方法除了普通事件onNext()之外，還定義了兩個特殊的事件onCompleted()和onError()。這些方法具體內容如下：

• onCompleted()——事件隊列完結。RxJava 不僅把每個事件單獨處理，還會把他們看做一個隊列。RxJava 規定如果隊列處理完畢，需要回調此方法作為標志
• onError()——事件隊列異常。在事件處理過程中出現異常時，此方法會被回調，同時事件停止發出

　　在一個正確運行的事件序列中，onCompleted()和onError()有且只有一個，并且是事件序列中的最后一個。需要注意的是，onCompleted()和onError()二者也是互斥的，即在隊列中調用了其中一個，就不應該再調用另一個。

使用

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!");} }復制代碼

　　其實 RxJava 中還有一個實現了 Observer 接口的抽象類 Subscriber。這兩個類基本使用方式一樣，而且在subscribe()時 Observer 對象先被轉換成 Subscriber 對象再使用。但他們的區別對于使用者來說主要有兩點：

• onStart()——這是 Subscriber 增加的方法。它會在subscribe()剛開始，而事件還未發送之前被調用，可以用于做一些準備工作，例如數據的清零或重置。這是一個可選方法，默認情況下它的實現為空。需要注意的是，如果對準備工作的線程有要求（例如彈出一個顯示進度的對話框，這必須在主線程執行）， 該方法就不適用了，因為它總是在subscribe()所發生的線程被調用，而不能指定線程。要在指定的線程來做準備工作，可以使用doOnSubscribe()方法。
• unsubscribe()——這是 Subscriber 所實現的另一個接口 Subscription 的方法，用于取消訂閱。在這個方法被調用后，Subscriber 將不再接收事件。一般在這個方法調用前，可以使用isUnsubscribed()先判斷一下狀態。 unsubscribe()這個方法很重要，因為在subscribe()之后， Observable 會持有 Subscriber 的引用，這個引用如果不能及時被釋放，將有內存泄露的風險。所以最好保持一個原則：要在不再使用的時候盡快在合適的地方（例如onPause()、onStop()等方法中）調用unsubscribe()來解除引用關系，以避免內存泄露的發生。

2.創建 Observable

　　Observable 即被觀察者，它決定什么時候觸發事件以及觸發怎樣的事件。它有三種方法來創建一個 Observable，并為它定義事件觸發規則

• create()

  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();} });復制代碼

  create()方法的參數 OnSubscribe 相當于一個計劃表，當 Observable 被訂閱時 OnSubscribe 的call()方法會被自動調用，事件序列就會依照設定依次觸發(對于上面的代碼，就是觀察者 Subscriber 將會被調用三次onNext()和一次onCompleted())。這樣，由被觀察者調用了觀察者的回調方法，就實現了由被觀察者向觀察者的事件傳遞，即觀察者模式。

• just()

  Observable observable = Observable.just("Hello","Hi","Aloha"); //將會依次調用 //onNext("Hello"); //onNext("Hi"); //onNext("Aloha"); //onCompleted();復制代碼

• from(T[] t)

  String[] words = {"Hello","Hi","Aloha"}; Observable observable = Observable.from(words); //將會依次調用 //onNext("Hello"); //onNext("Hi"); //onNext("Aloha"); //onCompleted();復制代碼

3.subscribe()訂閱

　　創建了 Observable 和 Observer 之后，再用subscribe()方法將它們聯結起來，整條鏈子就可以工作了。代碼形式很簡單：

observable.subscribe(observer); //或者 observable.subscribe(subscriber);復制代碼

　　我們再來關注一下subscribe()方法的內部實現

public Subscription subscribe(Subscriber subscriber){......subscriber.onStart();onSubscriber.call(subscriber); //事件發送邏輯開始運行return subscriber; }復制代碼

4.線程控制

　　在不指定線程的情況下， RxJava 遵循的是線程不變的原則，即：在哪個線程調用subscribe()，就在哪個線程生產事件；在哪個線程生產事件，就在哪個線程消費事件。如果需要切換線程，就需要用到 Scheduler (調度器)。

　　在 RxJava 中，Scheduler ——調度器，相當于線程控制器，RxJava 通過它來指定每一段代碼應該運行在什么樣的線程。RxJava 已經內置了幾個 Scheduler ，它們已經適合大多數的使用場景：

• Schedulers.immediate()——直接在當前線程運行，相當于不指定線程。這是默認的情況。
• Schedulers.newThread()——總是啟用新線程，并在新線程執行操作。
• Schedulers.io()——I/O 操作（讀寫文件、讀寫數據庫、網絡信息交互等）所使用的 Scheduler。行為模式和newThread()差不多，區別在于該方法的內部實現是是用一個無數量上限的線程池，可以重用空閑的線程，因此多數情況下該方法比newThread()更有效率。不要把計算工作放在該方法中，可以避免創建不必要的線程。
• Schdeulers.computation()——計算所使用的 Scheduler。這個計算指的是 CPU 密集型計算，即不會被 I/O 等操作限制性能的操作，例如圖形的計算。這個 Scheduler 使用的固定的線程池，大小為 CPU 核數。不要把 I/O 操作放在computation()中，否則 I/O 操作的等待時間會浪費 CPU。
• AndroidSchedulers.mainThread()——它指定的操作將在 Android 主線程運行。

　　有了這幾個 Scheduler ，就可以使用subscribeOn()和observeOn()兩個方法來對線程進行控制了。 subscribeOn()指定subscribe()所發生的線程，即 Observable.OnSubscribe 被激活時所處的線程。或者叫做事件產生的線程。 observeOn():指定 Subscriber 所運行在的線程?；蛘呓凶鍪录M的線程。舉個例子

Observable.just(1,2,3,4).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Action<Integer>(){@Overridepublic void call(Integer number){Log.d(TAG,"number:"+number);}})復制代碼

　　上面這段代碼中，由于subscribeOn(Schedulers.io())的指定，被創建的事件的內容 1、2、3、4 將會在 IO 線程發出；而由于observeOn(AndroidScheculers.mainThread())的指定，因此 subscriber 數字的打印將發生在主線程 。事實上，這種使用方式非常常見，它適用于多數的『后臺線程取數據，主線程顯示』的程序策略。

總結

　　對于 RxJava，我們應該記住兩個關鍵字：異步、簡潔。而 RxJava 還有一些比較重要的關鍵點需要理解，但本篇僅僅用于向初學者普及 RxJava 的一些簡單原理和使用方法，更多內容大家可以查看給 Android 開發者的 RxJava 詳解。

最后是廣告時間，我的博文將同步更新在三大平臺上，歡迎大家點擊閱讀！謝謝

劉志宇的新天地

簡書

稀土掘金

轉載于:https://juejin.im/post/595a034a6fb9a06bbe7db635

總結

以上是生活随笔為你收集整理的RxJava 基础扫盲的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。