開場(chǎng)白
如果你以前學(xué)過(guò)RxJava1.x，那么對(duì)于RxJava2.x只需要看看更新了哪些東西就行了，其最核心的思想并沒(méi)有變化，如果你沒(méi)學(xué)過(guò)RxJava1.x，也沒(méi)有關(guān)系，直接學(xué)習(xí)RxJava2。本文中闡述的所有的名詞都是關(guān)于RxJava2.x，不涉及RxJava.1.x.

開始

• 在使用RxJava2.x之前我們需要引入函數(shù)庫(kù)，在Android Studio Module級(jí)別的Gradle配置:

  compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
  compile 'io.reactivex.rxjava2:rxjava:2.1.0'
  

在開始學(xué)習(xí)之前，先來(lái)介紹點(diǎn)原理性的東西。

網(wǎng)上也有很多介紹RxJava原理的文章，通常這些文章都從觀察者模式開始，先講觀察者，被觀察者，訂閱關(guān)系巴拉巴拉一大堆，說(shuō)實(shí)話，當(dāng)我第一次看到這些文章的時(shí)候已經(jīng)被這些名詞給繞暈了，用了很長(zhǎng)的時(shí)間才理清楚它們之間的關(guān)系?？赡苁俏姨懒?，境界不夠，領(lǐng)會(huì)不到那么多高大上的名詞.

今天我用兩根水管代替觀察者和被觀察者, 試圖用通俗易懂的話把它們的關(guān)系解釋清楚, 在這里我將從事件流這個(gè)角度來(lái)說(shuō)明RxJava的基本工作原理。

先假設(shè)有兩根水管：

image.png

上面一根水管為事件產(chǎn)生的水管，叫它上游吧，下面一根水管為事件接收的水管叫它下游吧。

兩根水管通過(guò)一定的方式連接起來(lái)，使得上游每產(chǎn)生一個(gè)事件，下游就能收到該事件。注意這里和官網(wǎng)的事件圖是反過(guò)來(lái)的, 這里的事件發(fā)送的順序是先1,后2,后3這樣的順序, 事件接收的順序也是先1,后2,后3的順序, 我覺(jué)得這樣更符合我們普通人的思維, 簡(jiǎn)單明了.

這里的上游和下游就分別對(duì)應(yīng)著RxJava中的Observable（被觀察者）和Observer（觀察者），它們之間的關(guān)聯(lián)就對(duì)應(yīng)著subscribe()方法，即訂閱關(guān)系，就好比我們訂閱報(bào)紙，那我們就是觀察者，報(bào)社就是被觀察者，那么郵局就是subscribe()，把報(bào)社和用戶關(guān)聯(lián)起來(lái)，因此這個(gè)關(guān)系用RxJava來(lái)表示就是：

   //創(chuàng)建一個(gè)上游 Observable：
    Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
            //就是說(shuō)這個(gè)Observable會(huì)發(fā)射4個(gè)事件包括complete事件
            emitter.onNext(1);//發(fā)送一個(gè)integer事件
            emitter.onNext(2);;//發(fā)送一個(gè)integer事件
            emitter.onNext(3);;//發(fā)送一個(gè)integer事件
            emitter.onComplete();;//發(fā)送一個(gè)integer事件
        }
    });
    //創(chuàng)建一個(gè)下游 Observer
    Observer<Integer> observer = new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, "subscribe");
        }

        @Override
        public void onNext(Integer value) {
            Log.d(TAG, "" + value);
        }

        @Override
        public void onError(Throwable e) {
            Log.d(TAG, "error");
        }

        @Override
        public void onComplete() {
            Log.d(TAG, "complete");
        }
    };
    //建立關(guān)聯(lián)，訂閱關(guān)系
    observable.subscribe(observer);

這個(gè)運(yùn)行的結(jié)果就是:

12-02 03:37:17.818 4166-4166/zlc.season.rxjava2demo D/TAG: subscribe
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: 1
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: 2
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: 3
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: complete

注意: 只有當(dāng)上游和下游建立連接之后, 上游才會(huì)開始發(fā)送事件. 也就是調(diào)用了subscribe() 方法之后才開始發(fā)送事件（代碼實(shí)踐）.

把上面這端邏輯寫就成了RxJava鏈?zhǔn)讲僮鳎?/p>

Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
            emitter.onNext(1);
            emitter.onNext(2);
            emitter.onNext(3);
            emitter.onComplete();
        }
    }).subscribe(new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, "subscribe");
        }

        @Override
        public void onNext(Integer value) {
            Log.d(TAG, "" + value);
        }

        @Override
        public void onError(Throwable e) {
            Log.d(TAG, "error");
        }

        @Override
        public void onComplete() {
            Log.d(TAG, "complete");
        }
    });

ObservableEmitter：

接下來(lái)解釋一下其中兩個(gè)陌生的玩意：ObservableEmitter和Disposable.
ObservableEmitter： Emitter是發(fā)射器的意思，那就很好猜了，這個(gè)就是用來(lái)發(fā)出事件的，它可以發(fā)出三種類型的事件，通過(guò)調(diào)用emitter的onNext(T value)、onComplete()和onError(Throwable error)就可以分別發(fā)出next事件、complete事件和error事件。

但是，請(qǐng)注意，并不意味著你可以隨意亂七八糟發(fā)射事件，需要滿足一定的規(guī)則：

• 上游可以發(fā)送無(wú)限個(gè)onNext, 下游也可以接收無(wú)限個(gè)onNext.
• 當(dāng)上游發(fā)送了一個(gè)onComplete后, 上游onComplete之后的事件將會(huì)繼續(xù)發(fā)送, 而下游收到onComplete事件之后將不再繼續(xù)接收事件，發(fā)送complete事件告訴事件接收者，我已經(jīng)發(fā)送完成，那接收者將不會(huì)在接收.
• 當(dāng)上游發(fā)送了一個(gè)onError后, 上游onError之后的事件將繼續(xù)發(fā)送, 而下游收到onError事件之后將不再繼續(xù)接收事件.
• 上游可以不發(fā)送onComplete或onError.
• 最為關(guān)鍵的是onComplete和onError必須唯一并且互斥, 即不能發(fā)多個(gè)onComplete, 也不能發(fā)多個(gè)onError, 也不能先發(fā)一個(gè)onComplete, 然后再發(fā)一個(gè)onError, 反之亦然。

注: 關(guān)于onComplete和onError唯一并且互斥這一點(diǎn), 是需要自行在代碼中進(jìn)行控制, 如果你的代碼邏輯中違背了這個(gè)規(guī)則, 并不一定會(huì)導(dǎo)致程序崩潰. 比如發(fā)送多個(gè)onComplete是可以正常運(yùn)行的, 依然是收到第一個(gè)onComplete就不再接收了, 但若是發(fā)送多個(gè)onError, 則收到第二個(gè)onError事件會(huì)導(dǎo)致程序會(huì)崩潰.

以上幾個(gè)規(guī)則用示意圖表示如下:

示意圖

只發(fā)送onNext事件

發(fā)送onComplete事件

發(fā)送onError事件

Disposable

介紹了ObservableEmitter, 接下來(lái)介紹Disposable, 這個(gè)單詞的字面意思是一次性用品,用完即可丟棄的. 那么在RxJava中怎么去理解它呢, 對(duì)應(yīng)于上面的水管的例子, 我們可以把它理解成兩根管道之間的一個(gè)機(jī)關(guān), 當(dāng)調(diào)用它的dispose()方法時(shí), 它就會(huì)將兩根管道切斷, 從而導(dǎo)致下游收不到事件.

注意: 調(diào)用dispose()并不會(huì)導(dǎo)致上游不再繼續(xù)發(fā)送事件, 上游會(huì)繼續(xù)發(fā)送剩余的事件，相當(dāng)于解除訂閱關(guān)系.

來(lái)看個(gè)例子, 我們讓上游依次發(fā)送1,2,3,complete,4,在下游收到第二個(gè)事件之后, 切斷水管, 看看運(yùn)行結(jié)果:

 Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
            Log.d(TAG, "emit 1");
            emitter.onNext(1);
            Log.d(TAG, "emit 2");
            emitter.onNext(2);
            Log.d(TAG, "emit 3");
            emitter.onNext(3);
            Log.d(TAG, "emit complete");
            emitter.onComplete();
            Log.d(TAG, "emit 4");
            emitter.onNext(4);
        }
    }).subscribe(new Observer<Integer>() {
        private Disposable mDisposable;
        private int i;

        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, "subscribe");
            mDisposable = d;
        }

        @Override
        public void onNext(Integer value) {
            Log.d(TAG, "onNext: " + value);
            i++;
            if (i == 2) {
                Log.d(TAG, "dispose");
                mDisposable.dispose();
                Log.d(TAG, "isDisposed : " + mDisposable.isDisposed());
            }
        }

        @Override
        public void onError(Throwable e) {
            Log.d(TAG, "error");
        }

        @Override
        public void onComplete() {
            Log.d(TAG, "complete");
        }
    });

運(yùn)行結(jié)果為:

12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: subscribe
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 1
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: onNext: 1
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 2
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: onNext: 2
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: dispose
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: isDisposed : true
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 3
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit complete
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 4

從運(yùn)行結(jié)果我們看到, 在收到onNext 2這個(gè)事件后, 切斷了水管, 但是上游仍然發(fā)送了3, complete, 4這幾個(gè)事件, 而且上游并沒(méi)有因?yàn)榘l(fā)送了onComplete而停止. 同時(shí)可以看到下游的onSubscribe()方法是最先調(diào)用的.

Disposable的用處不止這些, 后面講解到了線程的調(diào)度之后, 我們會(huì)發(fā)現(xiàn)它的重要性. 隨著后續(xù)深入的講解, 我們會(huì)在更多的地方發(fā)現(xiàn)它的身影.

另外, subscribe()有多個(gè)重載的方法:

public final Disposable subscribe() {}
public final Disposable subscribe(Consumer<? super T> onNext) {}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {} 
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}
public final void subscribe(Observer<? super T> observer) {}

最后一個(gè)帶有Observer參數(shù)的我們已經(jīng)使用過(guò)了,這里對(duì)其他幾個(gè)方法進(jìn)行說(shuō)明.

不帶任何參數(shù)的subscribe() 表示下游不關(guān)心任何事件,你上游盡管發(fā)你的數(shù)據(jù)去吧, 老子可不管你發(fā)什么.
帶有一個(gè)Consumer參數(shù)的方法表示下游只關(guān)心onNext事件, 其他的事件我假裝沒(méi)看見(jiàn), 因此我們?nèi)绻恍枰猳nNext事件可以這么寫:

   Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
            Log.d(TAG, "emit 1");
            emitter.onNext(1);
            Log.d(TAG, "emit 2");
            emitter.onNext(2);
            Log.d(TAG, "emit 3");
            emitter.onNext(3);
            Log.d(TAG, "emit complete");
            emitter.onComplete();
            Log.d(TAG, "emit 4");
            emitter.onNext(4);
        }
    }).subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            Log.d(TAG, "onNext: " + integer);
        }
    });

其他幾個(gè)方法同理, 這里就不一一解釋了。