RxJava是什麼、優點

RxJava (本文所有RxJava全都代指RxJava2)在 GitHub 主頁上的自我介紹是:RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM. 
核心就是一個基於事件的非同步的程式庫！ 
好處就是解決原生非同步回撥，造成的程式碼冗長、可讀性可維護性差的問題，將其以一種鏈式的平鋪的寫法呈現，更方便維護！（有點類似前端Promise的作用）

implementation 'io.reactivex.rxjava2:rxjava:2.1.12'
implementation 'io.reactivex.rxjava2:rxandroid:2.0.2'

觀察者與被觀察者 
簡單的說, 被觀察者用於傳送事件，觀察者用於接收事件，兩者建立訂閱關係後，就可以實現，被觀察者在需要的時候傳送事件來通知觀察者！ 
關於觀察者模式不用過於糾結，只需簡單的理解，被觀察者屬於上游事件來源，用來發送事件，觀察者屬於下游，用於接收事件就ok了！ 
RxJava中有兩種觀察者模式：

• Observable ( 被觀察者 ) / Observer ( 觀察者 )
• Flowable （被觀察者）/ Subscriber （觀察者）

前者不支援背壓（Backpressure），後者支援！ 
背壓是指在非同步場景中，被觀察者傳送事件速度遠快於觀察者的處理速度的情況下，一種告訴上游的被觀察者降低傳送速度的策略。官方對於背壓的解釋

來一個例子

    /*
         RxJava 的三部曲：
         初始化 Observable（被觀察者 事件來源）
         初始化 Observer（觀察者 事件接收）
         建立訂閱關係（將兩者建立聯絡 被觀察者可以通過發射事件通知觀察者）

    */
    private void testRxJava1() {
       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);
                Log.e("TAG", "emitter on...");
                emitter.onComplete();
                Log.e("TAG", "emitter complete and on...");
                emitter.onNext(4);
            }
        });
       Observer<Integer> observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                //Disposable可以做到切斷的操作，讓Observer觀察者不再接收上游事件
                Log.d("TAG", "subscribe");
            }

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

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

            @Override
            public void onComplete() {
                Log.d("TAG", "onComplete");
            }
        };
        observable.subscribe(consumer);//建立訂閱關係
    }
 

05-19 11:50:57.050 2321-2321/com.example.rxjava.rxjavademo E/TAG: subscribe
05-19 11:50:57.050 2321-2321/com.example.rxjava.rxjavademo E/TAG: integer = 1
05-19 11:50:57.050 2321-2321/com.example.rxjava.rxjavademo E/TAG: integer = 2
05-19 11:50:57.050 2321-2321/com.example.rxjava.rxjavademo E/TAG: integer = 3
05-19 11:50:57.050 2321-2321/com.example.rxjava.rxjavademo E/TAG: emitter on...
05-19 11:50:57.050 2321-2321/com.example.rxjava.rxjavademo E/TAG: onComplete
05-19 11:50:57.050 2321-2321/com.example.rxjava.rxjavademo E/TAG: emitter complete and on...

從上面的結果可以看出，事件發射執行的一些順序，規則！如被觀察者呼叫emitter.onComplete();之後，雖然下面的事件繼續發射，但是觀察者並沒有接收！

ObservableEmitter 事件發射器 它可以通過呼叫emitter的onNext(T value)、onComplete()和onError(Throwable error)分別發出next事件、complete事件和error事件，它有以下幾點需要注意的地方：

• 官方推薦當呼叫onComplete或者onError，作為結束標誌後，不應再發射事件
• onComplete或者onError應當只調用一次
• 其實你違背了上述原則，也沒事，如呼叫onComplete之後，可以繼續發射事件，但是下游觀察者無法接收

Disposable，可以通過呼叫他的dispose()方法，讓下游觀察者無法繼續接收事件！

    //將上面那個例子鏈起來 就是RxJava經典寫法鏈式操作
    private void testRxJava1() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                Log.e("TAG", "emitter on...");
                emitter.onComplete();
                Log.e("TAG", "emitter complete and on...");
                emitter.onNext(4);
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d("TAG", "subscribe");
            }

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

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

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

另外， 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) {}

帶有一個Consumer引數的subscribe

上面這個例子，可以用java8的箭頭函式，進一步簡化

05-19 15:25:11.290 7170-7170/com.example.rxjava.rxjavademo E/TAG: emmiter 1
05-19 15:25:11.290 7170-7170/com.example.rxjava.rxjavademo E/TAG: accept integer = 1
05-19 15:25:11.290 7170-7170/com.example.rxjava.rxjavademo E/TAG: emmiter 2
05-19 15:25:11.290 7170-7170/com.example.rxjava.rxjavademo E/TAG: accept integer = 2
05-19 15:25:11.290 7170-7170/com.example.rxjava.rxjavademo E/TAG: emmiter 3
05-19 15:25:11.290 7170-7170/com.example.rxjava.rxjavademo E/TAG: accept integer = 3
05-19 15:25:11.290 7170-7170/com.example.rxjava.rxjavademo E/TAG: emmiter 4
05-19 15:25:11.290 7170-7170/com.example.rxjava.rxjavademo E/TAG: emmiter 5

帶有兩個引數的subscribe() ：

   Observable.just(1, "2")
             .cast(Integer.class)//將被觀察者傳送的事件資料型別強轉為Integer
             .subscribe(new Consumer<Integer>() {//只處理onNext
                 @Override
                 public void accept(Integer integer) throws Exception {
                     Log.e("TAG", "" + integer);
                 }
             }, new Consumer<Throwable>() {//捕獲異常
                 @Override
                 public void accept(Throwable throwable) throws Exception {
                     Log.e("TAG", throwable.getMessage());
                 }
             });

當然它也可以用java8的箭頭函式改寫：

    Observable.just(1, "2")
              .cast(Integer.class)
              .subscribe(integer -> Log.e("TAG", "" + integer), 
                         throwable -> Log.e("TAG", throwable.getMessage()));

執行結果

05-22 15:15:54.405 6034-6034/com.example.rxjava.rxjavademo E/TAG: 1
05-22 15:15:54.405 6034-6034/com.example.rxjava.rxjavademo E/TAG: java.lang.String cannot be cast to java.lang.Integer

這裡只是對基本概念的介紹，對RxJava有一個基本瞭解，後續會進行詳細的用法解讀！ 

                                                                  
                                                                           全棧開發者微信公眾號