RxJava 基礎掃盲

時間 2019-11-16

標籤 rxjava 基礎掃盲欄目 Java 简体版

原文原文鏈接

　　引言：前幾天瀏覽了一下稀土 App，發現有個挺不錯的新聞 App 實戰實例。正好最近想學習一下完整項目的源碼（特別是後臺代碼，就是各類框架等等）。而後想起之前不少大牛都談起過 RxJava、Retrofit、Gson 等等框架，因此如今決定來學習學習這些經常使用的框架。java

RxJava 究竟是什麼

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

RxJava 好在哪裏

　　仍是用一個詞歸納：簡潔。異步操做很關鍵的一點是要注意程序的簡潔性，由於在調度過程比較複雜的狀況下，異步代碼常常會既難寫也難被讀懂。 Android 創造的 AsyncTask 和 Handler ，其實都是爲了讓異步代碼更加簡潔。RxJava 的優點也是簡潔，但它的簡潔的不同凡響之處在於，隨着程序邏輯變得愈來愈複雜，它依然可以保持簡潔。github

　　假設有這樣一個需求：界面上有一個自定義的視圖 ImageCollectorView ，它的做用是顯示多張圖片，並能使用addImage(Bitmap)方法來任意增長顯示的圖片。如今須要程序將一個給出的目錄數組folders中每一個目錄下的.png圖片都加載出來並顯示在 ImageCollectorView 中。須要注意的是，因爲讀取圖片的這一過程較爲耗時，須要放在後臺執行，而圖片的顯示則必須在 UI 線程執行。經常使用的實現方式有多種，我這裏貼出其中一種：數據庫

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(){
              mImageCollcetorView.addImage(bitmap);
            }
          });
        }
      }
    }
  }
}.start();複製代碼

　　而若是使用 RxJava，就能夠寫成：設計模式

Observable.from(folders)
  .flatMap(new Func1<File,Observable<File>>(){
    @Override
    public Observable<File> call(File file){
      return Observable.from(file.listFiles());
    }
  })
  .filter(new Func1<File,Boolean>(){
    @Override
    public Boolean call(File file){
      return file.getName().endsWith(".png");
    }
  })
  .map(new Func1<File,Bitmap>(){
    @Override
    public Bitmap call(File file){
      return getBitmapFromFile(file);
    }
  })
  .subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(new Action1<Bitmap>(){
    @Override
    public 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()兩者也是互斥的，即在隊列中調用了其中一個，就不該該再調用另外一個。ide

使用

1.建立 Observer

　　Observer 即觀察者，它決定事件觸發的時候將有怎樣的行爲。RxJava 中的 Observer 接口的實現方式：

Observer<String> observer = new Observer<String>(){
  @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 中還有一個實現了 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>(){
  @Override
  public 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();複製代碼

　　建立了 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 所運行在的線程。或者叫作事件消費的線程。舉個例子

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

　　上面這段代碼中，因爲subscribeOn(Schedulers.io())的指定，被建立的事件的內容一、二、三、4 將會在 IO 線程發出；而因爲observeOn(AndroidScheculers.mainThread())的指定，所以 subscriber 數字的打印將發生在主線程。事實上，這種使用方式很是常見，它適用於多數的『後臺線程取數據，主線程顯示』的程序策略。