RxJava入門之介紹與基本運用

前言

因為這個RxJava內容不算少，而且應用場景非常廣，所以這個關於RxJava的文章我們會陸續更新，今天就來先來個入門RxJava吧

初識RxJava

什麼是Rx

很多教程在講解RxJava的時候，上來就介紹了什麼是RxJava。這裡我先說一下什麼是Rx，Rx就是ReactiveX，官方定義是：

    Rx是一個函數庫，讓開發者可以利用可觀察序列和LINQ風格查詢操作符來編寫異步和基於事件的程序

看到這個定義我只能呵呵，稍微通俗點說是這樣的：

    Rx是微軟.NET的一個響應式擴展。Rx借助可觀測的序列提供一種簡單的方式來創建異步的，基於事件驅動的程序。

這個有點清晰了，至少看到我們熟悉的異步與事件驅動，所以簡單點且不准確地來說：

     Rx就是一種響應式編程，來創建基於事件的異步程序

注意，這個定義是不准確的，但是對於初學者來說，已經可以有個基本的認知了。

另外還有一點就是Rx其實是一種編程思想，用很多語言都可以實現，比如RxJava、RxJS、RxPHP等等。而現在我們要說的就是RxJava。

RxJava是什麼

二話不說，先上定義：

     RxJava就是一種用Java語言實現的響應式編程，來創建基於事件的異步程序

有人問你這不是廢話麼，好吧那我上官方定義：

     一個在 Java VM 上使用可觀測的序列來組成異步的、基於事件的程序的庫

反正我剛看這句話的時候也呵呵了，當然現在有所領悟了。

     除此之外，就是：異步，它就是一個實現異步操作的庫。

擴展的觀察者模式

對於普通的觀察者模式，這裡我就不細說了。簡單概括就是，觀察者（Observer）需要在被觀察者(Observable)變化的一瞬間做出反應。

而兩者通過注冊（Register）或者訂閱(Subscribe)的方式進行綁定。

就拿扔物線老師給的例子來說，我豐富了一下如圖所示：

其中這個Button就是被觀察者（Observable），OnClickListener就是觀察者（Observer），兩者通過setOnClickListener達成訂閱（Subscribe）關系，之後當Button產生OnClick事件的時候，會直接發送給OnClickListener，它做出相應的響應處理。

當然還有其他的例子，比如Android四大組件中的ContentProvider與ContentObserver之間也存在這樣的關系。

而RxJava的觀察者模式呢，跟這個差不多，但是也有幾點差別：

      Observer與Observable是通過 subscribe() 來達成訂閱關系。

      RxJava中事件回調有三種：onNext() 、 onCompleted() 、 onError() 。

      如果一個Observerble沒有任何的Observer，那麼這個Observable是不會發出任何事件的。

其中關於第三點，這裡想說明一下，在Rx中，其實Observable有兩種形式：熱啟動Observable和冷啟動Observable。

      熱啟動Observable任何時候都會發送消息，即使沒有任何觀察者監聽它。

      冷啟動Observable只有在至少有一個訂閱者的時候才會發送消息

這個地方雖然對於初學者來說區別不大，但是要注意一下，所以上面的第三點其實就針對於冷啟動來說的。

另外，關於RxJava的回調事件的總結：

      onNext() ：基本事件。

      onCompleted() : 事件隊列完結。RxJava 不僅把每個事件單獨處理，還會把它們看做一個隊列。RxJava 規定，當不會再有新的 onNext()  發出時，需要觸發 onCompleted() 方法作為標志。

      onError() : 事件隊列異常。在事件處理過程中出異常時，onError() 會被觸發，同時隊列自動終止，不允許再有事件發出。

值得注意的是在一個正確運行的事件序列中, onCompleted() 和 onError() 有且只有一個，並且是事件序列中的最後一個。如果在隊列中調用了其中一個，就不應該再調用另一個。

好了，那我們也附一張圖對比一下吧：

如何實現RxJava

關於實現RxJava的步驟，這裡我就大體總結概括一下。

創建Observer

在Java中，一想到要創建一個對象，我們馬上就想要new一個。沒錯，這裡我們也是要new一個Observer出來，其實就是實現Observer的接口，注意String是接收參數的類型：

//創建Observer
Observer<String> observer = new Observer<String>() {
 @Override
 public void onNext(String s) {
  Log.i("onNext ---> ", "Item: " + s);
 }

 @Override
 public void onCompleted() {
  Log.i("onCompleted ---> ", "完成");
 }

 @Override
 public void onError(Throwable e) {
  Log.i("onError ---> ", e.toString());
 }
};

當然這裡也要提一個實現了 Observer 接口的抽象類：Subscriber ，它跟 Observer 接口幾乎完全一樣，只是多了兩個方法，看看總結：

      onStart() :  它會在 subscribe 剛開始，而事件還未發送之前被調用，可以用於做一些准備工作，例如數據的清零或重置。這是一個可選方法，默認情況下它的實現為空。需要注意的是，如果對准備工作的線程有要求（例如彈出一個顯示進度的對話框，這必須在主線程執行）， onStart() 就不適用了，因為它總是在 subscribe 所發生的線程被調用，而不能指定線程。

      unsubscribe() : 用於取消訂閱。在這個方法被調用後，Subscriber 將不再接收事件。一般在這個方法調用前，可以使用 isUnsubscribed() 先判斷一下狀態。 要在不再使用的時候盡快在合適的地方（例如 onPause() onStop() 等方法中）調用 unsubscribe() 來解除引用關系，以避免內存洩露的發生。

雖然多了兩個方法，但是基本實現方式跟Observer是一樣的，所以暫時可以不考慮兩者的區別。不過值得注意的是：

實質上，在 RxJava 的 subscribe 過程中，Observer 也總是會先被轉換成一個 Subscriber 再使用。

創建Observable

與Observer不同的是，Observable是通過 create() 方法來創建的。注意String是發送參數的類型：

//創建Observable
Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
 @Override
 public void call(Subscriber<? super String> subscriber) {
  subscriber.onNext("Hello");
  subscriber.onNext("World");
  subscriber.onCompleted();
 }
});

關於這其中的流程，我們暫且不考慮。

訂閱（Subscribe）

在之前，我們創建了 Observable 和 Observer ，現在就需要用 subscribe() 方法來將它們連接起來，形成一種訂閱關系：

//訂閱
observable.subscribe(observer);

這裡其實確實有點奇怪，為什麼是Observable（被觀察者）訂閱了Observer（觀察者）呢？其實我們想一想之前Button的點擊事件：

Button.setOnClickListener(new View.OnClickListener())

Button是被觀察者，OnClickListener是觀察者，setOnClickListener是訂閱。我們驚訝地發現，也是被觀察者訂閱了觀察者，所以應該是一種流式API的設計吧，也沒啥影響。

完整代碼如下：

 //創建Observer
 Observer<String> observer = new Observer<String>() {
  @Override
  public void onNext(String s) {
   Log.i("onNext ---> ", "Item: " + s);
  }

  @Override
  public void onCompleted() {
   Log.i("onCompleted ---> ", "完成");
  }

  @Override
  public void onError(Throwable e) {
   Log.i("onError ---> ", e.toString());
  }
 };

 //創建Observable
 Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
  @Override
  public void call(Subscriber<? super String> subscriber) {
   subscriber.onNext("Hello");
   subscriber.onNext("World");
   subscriber.onCompleted();
  }
 });

 //訂閱
 observable.subscribe(observer);

運行的結果如下，可以看到Observable中發送的String已經被Observer接收並打印了出來：

線程控制——Scheduler

好了，這裡就是RxJava的精髓之一了。

在RxJava中，Scheduler相當於線程控制器，可以通過它來指定每一段代碼運行的線程。

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。

       AndroidSchedulers.mainThread() ，Android專用線程，指定操作在主線程運行。

那我們如何切換線程呢？RxJava中提供了兩個方法：subscribeOn() 和 observeOn() ，兩者的不同點在於：

       subscribeOn() : 指定subscribe()訂閱所發生的線程，即 call() 執行的線程。或者叫做事件產生的線程。

       observeOn() : 指定Observer所運行在的線程，即onNext()執行的線程。或者叫做事件消費的線程。

具體實現如下：

//改變運行的線程
observable.subscribeOn(Schedulers.io());
observable.observeOn(AndroidSchedulers.mainThread());

這裡確實不好理解，沒關系，下面我們在具體例子中觀察現象。

而這其中的原理，會在之後的源碼級分析的文章中詳細解釋，現在我們暫且擱下。

第一個RxJava案例

好了，當看完之前的所有基礎東西，現在我們就完全可以寫一個基於RxJava的Demo了。

這裡我們用一個基於RxJava的異步加載網絡圖片來演示。

由於重點在於RxJava對於異步的處理，所以關於如何通過網絡請求獲取圖片，這裡就不詳細說明了。

另外這裡采用的是鏈式調用，並為重要位置打上Log日志，觀察方法執行的所在線程。

首先需要添加依賴，這沒什麼好說的：

dependencies {
 compile fileTree(include: ['*.jar'], dir: 'libs')
 testCompile 'junit:junit:4.12'
 ...
 compile 'io.reactivex:rxjava:1.1.6'

}

然後按照步驟來，首先通過create創建Observable，注意發送參數的類型是Bitmap：

//創建被觀察者
Observable.create(new Observable.OnSubscribe<Bitmap>() {
 /**
 * 復寫call方法
 *
 * @param subscriber 觀察者對象
 */
 @Override
 public void call(Subscriber<? super Bitmap> subscriber) {
  //通過URL得到圖片的Bitmap對象
  Bitmap bitmap = GetBitmapForURL.getBitmap(url);
  //回調觀察者方法
  subscriber.onNext(bitmap);
  subscriber.onCompleted();
  Log.i(" call ---> ", "運行在 " + Thread.currentThread().getName() + " 線程");
 }
})

然後我們需要創建Observer，並進行訂閱，這裡是鏈式調用

.subscribe(new Observer<Bitmap>() { //訂閱觀察者（其實是觀察者訂閱被觀察者）

 @Override
 public void onNext(Bitmap bitmap) {
  mainImageView.setImageBitmap(bitmap);
  Log.i(" onNext ---> ", "運行在 " + Thread.currentThread().getName() + " 線程");
 }

 @Override
 public void onCompleted() {
  mainProgressBar.setVisibility(View.GONE);
  Log.i(" onCompleted ---> ", "完成");
 }

 @Override
 public void onError(Throwable e) {
  Log.e(" onError --->", e.toString());
 }
 });

當然網絡請求是耗時操作，我們需要在其他線程中執行，而更新UI需要在主線程中執行，所以需要設置線程：

.subscribeOn(Schedulers.io()) // 指定subscribe()發生在IO線程
.observeOn(AndroidSchedulers.mainThread()) // 指定Subscriber的回調發生在UI線程

這樣我們就完成了一個RxJava的基本編寫，現在整體看一下代碼：

//創建被觀察者
Observable.create(new Observable.OnSubscribe<Bitmap>() {
 /**
 * 復寫call方法
 *
 * @param subscriber 觀察者對象
 */
 @Override
 public void call(Subscriber<? super Bitmap> subscriber) {
  //通過URL得到圖片的Bitmap對象
  Bitmap bitmap = GetBitmapForURL.getBitmap(url);
  //回調觀察者方法
  subscriber.onNext(bitmap);
  subscriber.onCompleted();
  Log.i(" call ---> ", "運行在 " + Thread.currentThread().getName() + " 線程");
 }
})
.subscribeOn(Schedulers.io()) // 指定subscribe()發生在IO線程
.observeOn(AndroidSchedulers.mainThread()) // 指定Subscriber的回調發生在UI線程
.subscribe(new Observer<Bitmap>() { //訂閱觀察者（其實是觀察者訂閱被觀察者）

 @Override
 public void onNext(Bitmap bitmap) {
  mainImageView.setImageBitmap(bitmap);
  Log.i(" onNext ---> ", "運行在 " + Thread.currentThread().getName() + " 線程");
 }

 @Override
 public void onCompleted() {
  mainProgressBar.setVisibility(View.GONE);
  Log.i(" onCompleted ---> ", "完成");
 }

 @Override
 public void onError(Throwable e) {
  Log.e(" onError --->", e.toString());
 }
 });

好了，下面是運行的動態圖：

RxJava異步加載網絡圖片

現在來看一下運行的Log日志：

Log

可以看到，call方法（事件產生）執行在IO線程，而onNext方法（事件消費）執行在main線程。說明之前分析的是對的。

總結

好了，由於本文是一個RxJava的基礎，所以篇幅稍微過長了點。即使這樣，很多細節性問題都沒有交代清楚。但所幸的是，本文已經將RxJava必要的基礎入門知識講解完了。可能由於技術水平有限，文中難免會有錯誤或者疏忽之處，歡迎大家指正與交流。希望這篇文章對大家的學習或者工作帶來一定的幫助，小編還會陸續更新相關的文章，感興趣的朋友們請繼續關注本站。