前言

因?yàn)檫@個(gè)RxJava內(nèi)容不算少，而且應(yīng)用場(chǎng)景非常廣，所以這個(gè)關(guān)于RxJava的文章我們會(huì)陸續(xù)更新，今天就來先來個(gè)入門RxJava吧

初識(shí)RxJava

什么是Rx

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

    Rx是一個(gè)函數(shù)庫，讓開發(fā)者可以利用可觀察序列和LINQ風(fēng)格查詢操作符來編寫異步和基于事件的程序

看到這個(gè)定義我只能呵呵，稍微通俗點(diǎn)說是這樣的：

    Rx是微軟.NET的一個(gè)響應(yīng)式擴(kuò)展。Rx借助可觀測(cè)的序列提供一種簡(jiǎn)單的方式來創(chuàng)建異步的，基于事件驅(qū)動(dòng)的程序。

這個(gè)有點(diǎn)清晰了，至少看到我們熟悉的異步與事件驅(qū)動(dòng)，所以簡(jiǎn)單點(diǎn)且不準(zhǔn)確地來說：

     Rx就是一種響應(yīng)式編程，來創(chuàng)建基于事件的異步程序

注意，這個(gè)定義是不準(zhǔn)確的，但是對(duì)于初學(xué)者來說，已經(jīng)可以有個(gè)基本的認(rèn)知了。

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

RxJava是什么

二話不說，先上定義：

     RxJava就是一種用Java語言實(shí)現(xiàn)的響應(yīng)式編程，來創(chuàng)建基于事件的異步程序

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

     一個(gè)在 Java VM 上使用可觀測(cè)的序列來組成異步的、基于事件的程序的庫

反正我剛看這句話的時(shí)候也呵呵了，當(dāng)然現(xiàn)在有所領(lǐng)悟了。

     除此之外，就是：異步，它就是一個(gè)實(shí)現(xiàn)異步操作的庫。

擴(kuò)展的觀察者模式

對(duì)于普通的觀察者模式，這里我就不細(xì)說了。簡(jiǎn)單概括就是，觀察者（Observer）需要在被觀察者(Observable)變化的一瞬間做出反應(yīng)。

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

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

其中這個(gè)Button就是被觀察者（Observable），OnClickListener就是觀察者（Observer），兩者通過setOnClickListener達(dá)成訂閱（Subscribe）關(guān)系，之后當(dāng)Button產(chǎn)生OnClick事件的時(shí)候，會(huì)直接發(fā)送給OnClickListener，它做出相應(yīng)的響應(yīng)處理。

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

而RxJava的觀察者模式呢，跟這個(gè)差不多，但是也有幾點(diǎn)差別：

      Observer與Observable是通過 subscribe() 來達(dá)成訂閱關(guān)系。

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

      如果一個(gè)Observerble沒有任何的Observer，那么這個(gè)Observable是不會(huì)發(fā)出任何事件的。

其中關(guān)于第三點(diǎn)，這里想說明一下，在Rx中，其實(shí)Observable有兩種形式：熱啟動(dòng)Observable和冷啟動(dòng)Observable。

      熱啟動(dòng)Observable任何時(shí)候都會(huì)發(fā)送消息，即使沒有任何觀察者監(jiān)聽它。

      冷啟動(dòng)Observable只有在至少有一個(gè)訂閱者的時(shí)候才會(huì)發(fā)送消息

這個(gè)地方雖然對(duì)于初學(xué)者來說區(qū)別不大，但是要注意一下，所以上面的第三點(diǎn)其實(shí)就針對(duì)于冷啟動(dòng)來說的。

另外，關(guān)于RxJava的回調(diào)事件的總結(jié)：

      onNext() ：基本事件。

      onCompleted() : 事件隊(duì)列完結(jié)。RxJava 不僅把每個(gè)事件單獨(dú)處理，還會(huì)把它們看做一個(gè)隊(duì)列。RxJava 規(guī)定，當(dāng)不會(huì)再有新的 onNext()  發(fā)出時(shí)，需要觸發(fā) onCompleted() 方法作為標(biāo)志。

      onError() : 事件隊(duì)列異常。在事件處理過程中出異常時(shí)，onError() 會(huì)被觸發(fā)，同時(shí)隊(duì)列自動(dòng)終止，不允許再有事件發(fā)出。

值得注意的是在一個(gè)正確運(yùn)行的事件序列中, onCompleted() 和 onError() 有且只有一個(gè)，并且是事件序列中的最后一個(gè)。如果在隊(duì)列中調(diào)用了其中一個(gè)，就不應(yīng)該再調(diào)用另一個(gè)。

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

如何實(shí)現(xiàn)RxJava

關(guān)于實(shí)現(xiàn)RxJava的步驟，這里我就大體總結(jié)概括一下。

創(chuàng)建Observer

在Java中，一想到要?jiǎng)?chuàng)建一個(gè)對(duì)象，我們馬上就想要new一個(gè)。沒錯(cuò)，這里我們也是要new一個(gè)Observer出來，其實(shí)就是實(shí)現(xiàn)Observer的接口，注意String是接收參數(shù)的類型：

//創(chuàng)建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());
 }
};

當(dāng)然這里也要提一個(gè)實(shí)現(xiàn)了 Observer 接口的抽象類：Subscriber ，它跟 Observer 接口幾乎完全一樣，只是多了兩個(gè)方法，看看總結(jié)：

      onStart() :  它會(huì)在 subscribe 剛開始，而事件還未發(fā)送之前被調(diào)用，可以用于做一些準(zhǔn)備工作，例如數(shù)據(jù)的清零或重置。這是一個(gè)可選方法，默認(rèn)情況下它的實(shí)現(xiàn)為空。需要注意的是，如果對(duì)準(zhǔn)備工作的線程有要求（例如彈出一個(gè)顯示進(jìn)度的對(duì)話框，這必須在主線程執(zhí)行）， onStart() 就不適用了，因?yàn)樗偸窃?subscribe 所發(fā)生的線程被調(diào)用，而不能指定線程。

      unsubscribe() : 用于取消訂閱。在這個(gè)方法被調(diào)用后，Subscriber 將不再接收事件。一般在這個(gè)方法調(diào)用前，可以使用 isUnsubscribed() 先判斷一下狀態(tài)。 要在不再使用的時(shí)候盡快在合適的地方（例如 onPause() onStop() 等方法中）調(diào)用 unsubscribe() 來解除引用關(guān)系，以避免內(nèi)存泄露的發(fā)生。

雖然多了兩個(gè)方法，但是基本實(shí)現(xiàn)方式跟Observer是一樣的，所以暫時(shí)可以不考慮兩者的區(qū)別。不過值得注意的是：

實(shí)質(zhì)上，在 RxJava 的 subscribe 過程中，Observer 也總是會(huì)先被轉(zhuǎn)換成一個(gè) Subscriber 再使用。

創(chuàng)建Observable

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

//創(chuàng)建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();
 }
});

關(guān)于這其中的流程，我們暫且不考慮。

訂閱（Subscribe）

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

//訂閱
observable.subscribe(observer);

這里其實(shí)確實(shí)有點(diǎn)奇怪，為什么是Observable（被觀察者）訂閱了Observer（觀察者）呢？其實(shí)我們想一想之前Button的點(diǎn)擊事件：

Button.setOnClickListener(new View.OnClickListener())

Button是被觀察者，OnClickListener是觀察者，setOnClickListener是訂閱。我們驚訝地發(fā)現(xiàn)，也是被觀察者訂閱了觀察者，所以應(yīng)該是一種流式API的設(shè)計(jì)吧，也沒啥影響。

完整代碼如下：

 //創(chuàng)建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());
  }
 };

 //創(chuàng)建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);

運(yùn)行的結(jié)果如下，可以看到Observable中發(fā)送的String已經(jīng)被Observer接收并打印了出來：

線程控制——Scheduler

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

在RxJava中，Scheduler相當(dāng)于線程控制器，可以通過它來指定每一段代碼運(yùn)行的線程。

RxJava已經(jīng)內(nèi)置了幾個(gè)Scheduler，下面是總結(jié)：

      Schedulers.immediate() : 直接在當(dāng)前線程運(yùn)行，相當(dāng)于不指定線程。這是默認(rèn)的Scheduler。

      Schedulers.newThread() : 總是啟用新線程，并在新線程執(zhí)行操作。

      Schedulers.io() : I/O 操作（讀寫文件、讀寫數(shù)據(jù)庫、網(wǎng)絡(luò)信息交互等）所使用的Scheduler。行為模式和newThread()差不多，區(qū)別在于io()的內(nèi)部實(shí)現(xiàn)是是用一個(gè)無數(shù)量上限的線程池，可以重用空閑的線程，因此多數(shù)情況下io()比newThread()更有效率。不要把計(jì)算工作放在io()中，可以避免創(chuàng)建不必要的線程。

      Schedulers.computation() : 計(jì)算所使用的Scheduler。這個(gè)計(jì)算指的是 CPU 密集型計(jì)算，即不會(huì)被 I/O 等操作限制性能的操作，例如圖形的計(jì)算。這個(gè)Scheduler使用的固定的線程池，大小為 CPU 核數(shù)。不要把 I/O 操作放在computation()中，否則 I/O 操作的等待時(shí)間會(huì)浪費(fèi) CPU。

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

那我們?nèi)绾吻袚Q線程呢？RxJava中提供了兩個(gè)方法：subscribeOn() 和 observeOn() ，兩者的不同點(diǎn)在于：

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

       observeOn() : 指定Observer所運(yùn)行在的線程，即onNext()執(zhí)行的線程?；蛘呓凶鍪录M(fèi)的線程。

具體實(shí)現(xiàn)如下：

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

這里確實(shí)不好理解，沒關(guān)系，下面我們?cè)诰唧w例子中觀察現(xiàn)象。

而這其中的原理，會(huì)在之后的源碼級(jí)分析的文章中詳細(xì)解釋，現(xiàn)在我們暫且擱下。

第一個(gè)RxJava案例

好了，當(dāng)看完之前的所有基礎(chǔ)東西，現(xiàn)在我們就完全可以寫一個(gè)基于RxJava的Demo了。

這里我們用一個(gè)基于RxJava的異步加載網(wǎng)絡(luò)圖片來演示。

由于重點(diǎn)在于RxJava對(duì)于異步的處理，所以關(guān)于如何通過網(wǎng)絡(luò)請(qǐng)求獲取圖片，這里就不詳細(xì)說明了。

另外這里采用的是鏈?zhǔn)秸{(diào)用，并為重要位置打上Log日志，觀察方法執(zhí)行的所在線程。

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

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

}

然后按照步驟來，首先通過create創(chuàng)建Observable，注意發(fā)送參數(shù)的類型是Bitmap：

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

然后我們需要?jiǎng)?chuàng)建Observer，并進(jìn)行訂閱，這里是鏈?zhǔn)秸{(diào)用

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

 @Override
 public void onNext(Bitmap bitmap) {
  mainImageView.setImageBitmap(bitmap);
  Log.i(" onNext ---> ", "運(yùn)行在 " + 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());
 }
 });

當(dāng)然網(wǎng)絡(luò)請(qǐng)求是耗時(shí)操作，我們需要在其他線程中執(zhí)行，而更新UI需要在主線程中執(zhí)行，所以需要設(shè)置線程：

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

這樣我們就完成了一個(gè)RxJava的基本編寫，現(xiàn)在整體看一下代碼：

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

 @Override
 public void onNext(Bitmap bitmap) {
  mainImageView.setImageBitmap(bitmap);
  Log.i(" onNext ---> ", "運(yùn)行在 " + 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());
 }
 });

好了，下面是運(yùn)行的動(dòng)態(tài)圖：

RxJava異步加載網(wǎng)絡(luò)圖片

現(xiàn)在來看一下運(yùn)行的Log日志：

Log

可以看到，call方法（事件產(chǎn)生）執(zhí)行在IO線程，而onNext方法（事件消費(fèi)）執(zhí)行在main線程。說明之前分析的是對(duì)的。

總結(jié)

好了，由于本文是一個(gè)RxJava的基礎(chǔ)，所以篇幅稍微過長(zhǎng)了點(diǎn)。即使這樣，很多細(xì)節(jié)性問題都沒有交代清楚。但所幸的是，本文已經(jīng)將RxJava必要的基礎(chǔ)入門知識(shí)講解完了?？赡苡捎诩夹g(shù)水平有限，文中難免會(huì)有錯(cuò)誤或者疏忽之處，歡迎大家指正與交流。希望這篇文章對(duì)大家的學(xué)習(xí)或者工作帶來一定的幫助，小編還會(huì)陸續(xù)更新相關(guān)的文章，感興趣的朋友們請(qǐng)繼續(xù)關(guān)注腳本之家。

最新評(píng)論