RxJava的學習與實現

時間 2019-11-29

標籤 rxjava 學習實現欄目 Java 简体版

原文原文鏈接

RxJava

要在Android中使用RxJava2, 先添加Gradle配置:java

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

原理

在開始學習以前，先來介紹點原理性的東西。react

網上也有不少介紹RxJava原理的文章，一般這些文章都從觀察者模式開始，先講觀察者，被觀察者，訂閱關係巴拉巴拉一大堆，說實話，當我第一次看到這些文章的時候已經被這些名詞給繞暈了，用了很長的時間才理清楚它們之間的關係。多是我太蠢了，境界不夠，領會不到那麼多高大上的名詞.android

今天我用兩根水管代替觀察者和被觀察者, 試圖用通俗易懂的話把它們的關係解釋清楚, 在這裏我將從事件流這個角度來講明RxJava的基本工做原理。ide

先假設有兩根水管：學習

上面一根水管爲事件產生的水管，叫它上游吧，下面一根水管爲事件接收的水管叫它下游吧。線程

兩根水管經過必定的方式鏈接起來，使得上游每產生一個事件，下游就能收到該事件。注意這裏和官網的事件圖是反過來的, 這裏的事件發送的順序是先1,後2,後3這樣的順序, 事件接收的順序也是先1,後2,後3的順序, 我以爲這樣更符合咱們普通人的思惟, 簡單明瞭.code

這裏的上游和下游就分別對應着RxJava中的Observable和Observer，它們之間的鏈接就對應着subscribe()，所以這個關係用RxJava來表示就是：server

//建立一個上游 Observable：
        Observable<Integer> observable = 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();
            }
        });
        //建立一個下游 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");
            }
        };
        //創建鏈接
        observable.subscribe(observer);

這個運行的結果就是:事件

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

注意: 只有當上遊和下游創建鏈接以後, 上游纔會開始發送事件. 也就是調用了subscribe() 方法以後纔開始發送事件.it

把這段代碼連起來寫就成了RxJava引覺得傲的鏈式操做：

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和Disposable.

ObservableEmitter： Emitter是發射器的意思，那就很好猜了，這個就是用來發出事件的，它能夠發出三種類型的事件，經過調用emitter的onNext(T value)、onComplete()和onError(Throwable error)就能夠分別發出next事件、complete事件和error事件。

可是，請注意，並不意味着你能夠隨意亂七八糟發射事件，須要知足必定的規則：

上游能夠發送無限個onNext, 下游也能夠接收無限個onNext.
當上遊發送了一個onComplete後, 上游onComplete以後的事件將會繼續發送, 而下游收到onComplete事件以後將再也不繼續接收事件.
當上遊發送了一個onError後, 上游onError以後的事件將繼續發送, 而下游收到onError事件以後將再也不繼續接收事件.
上游能夠不發送onComplete或onError.
最爲關鍵的是onComplete和onError必須惟一而且互斥, 即不能發多個onComplete, 也不能發多個onError, 也不能先發一個onComplete, 而後再發一個onError, 反之亦然

注: 關於onComplete和onError惟一而且互斥這一點, 是須要自行在代碼中進行控制, 若是你的代碼邏輯中違背了這個規則, 並不必定會致使程序崩潰. 好比發送多個onComplete是能夠正常運行的, 依然是收到第一個onComplete就再也不接收了, 但如果發送多個onError, 則收到第二個onError事件會致使程序會崩潰.

介紹了ObservableEmitter, 接下來介紹Disposable, 這個單詞的字面意思是一次性用品,用完便可丟棄的. 那麼在RxJava中怎麼去理解它呢, 對應於上面的水管的例子, 咱們能夠把它理解成兩根管道之間的一個機關, 當調用它的dispose()方法時, 它就會將兩根管道切斷, 從而致使下游收不到事件.

注意: 調用dispose()並不會致使上游再也不繼續發送事件, 上游會繼續發送剩餘的事件.

來看個例子, 咱們讓上游依次發送1,2,3,complete,4,在下游收到第二個事件以後, 切斷水管, 看看運行結果:

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");
            }
        });

運行結果爲:

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

從運行結果咱們看到, 在收到onNext 2這個事件後, 切斷了水管, 可是上游仍然發送了3, complete, 4這幾個事件, 並且上游並無由於發送了onComplete而中止. 同時能夠看到下游的onSubscribe()方法是最早調用的.

Disposable的用處不止這些, 後面講解到了線程的調度以後, 咱們會發現它的重要性. 隨着後續深刻的講解, 咱們會在更多的地方發現它的身影.

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

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) {}

最後一個帶有Observer參數的咱們已經使用過了,這裏對其餘幾個方法進行說明.

不帶任何參數的subscribe() 表示下游不關心任何事件,你上游儘管發你的數據去吧, 老子可無論你發什麼.
帶有一個Consumer參數的方法表示下游只關心onNext事件, 其餘的事件我僞裝沒看見, 所以咱們若是隻須要onNext事件能夠這麼寫:

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);
            }
        });