RxJava 2.x 使用最佳實踐

轉載請标明出處：http://blog.csdn.net/zhaoyanjun6/article/details/76443347

本文出自【趙彥軍的部落格】

以前寫過 Rxjava 系列教程, 如下所示

RxJava 和 RxAndroid 一 (基礎)
RxJava 和 RxAndroid 二（操作符的使用）
RxJava 和 RxAndroid 三（生命周期控制和記憶體優化）
RxJava 和 RxAndroid 四（RxBinding的使用）
RxJava 和 RxAndroid 五（線程排程）

上面的這些教程覆寫了 rxjava 的方方面面，很詳細。隻是當時寫的時候是基于 rxjava 1.X 的版本寫的，後來 rxjava 進入了快速疊代的時期，很快就出現了 2.x 版本。根據 Rxjava 官方的GitHub 來看，2.x 相對于 1.x 做了很多改進，删除了不少的類，同時也增加了一些新的類。基于以上背景，以前的這些文章，就顯得有些不足，為了緊跟 rxjava 的步伐，下面的這篇部落格，就是對 rxjava 的重新認識。

Rxjava、RxAndroid

Rxjava : https://github.com/ReactiveX/RxJava

RxAndroid : https://github.com/ReactiveX/RxAndroid

添加依賴

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

create() ：建立

create操作符應該是最常見的操作符了，主要用于産生一個Obserable被觀察者對象，為了友善大家的認知，以後的教程中統一把被觀察者Observable稱為發射器（上遊事件），觀察者Observer稱為接收器（下遊事件）。

Observable.create(new ObservableOnSubscribe<Integer>() {
          @Override
          public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
           e.onNext(1);
           e.onNext(2);
           e.onNext(3);
           e.onComplete(); //結束
           e.onNext( 4 );
          }
        })
          .subscribe(new Observer<Integer>() {
                  @Override
                  public void onSubscribe(@NonNull Disposable d) {
                      Log.e("zhao", "onSubscribe: " + d.isDisposed());
                  }

                  @Override
                  public void onNext(@NonNull Integer integer) {
                      Log.e("zhao", "onNext: " + integer);
                  }

                  @Override
                  public void onError(@NonNull Throwable e) {
                      Log.e("zhao", "onError: ");
                  }

                  @Override
                  public void onComplete() {
                      Log.e("zhao", "onComplete: ");
                  }
            });

結果是：

E/zhao: onSubscribe: false
E/zhao: onNext: 1
E/zhao: onNext: 2
E/zhao: onNext: 3
E/zhao: onComplete:

需要注意的幾點是：

1）在發射完 3 之後，調用 e.onComplete() 方法，結束發射資料。4 沒有發射出來。

另外一個值得注意的點是，在RxJava 2.x中，可以看到發射事件方法相比1.x多了一個throws Excetion，意味着我們做一些特定操作再也不用try-catch了。
并且2.x 中有一個Disposable概念，這個東西可以直接調用切斷，可以看到，當它的isDisposed()傳回為false的時候，接收器能正常接收事件，但當其為true的時候，接收器停止了接收。是以可以通過此參數動态控制接收事件了。

在上面接收資料的時候，我們用了 Observer 對象，需要實作 4 個方法。這顯得過于累贅，我們可以用 Consumer 對象來代替 Observer 對象，代碼如下：

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

效果如下：

E/zhao: accept: 1
 E/zhao: accept: 2
 E/zhao: accept: 3

需要注意的是：

1）、Consumer 對象完全代替了Observer ，效果是一樣的。Consumer 顧名思義是消費者的意思，是消費資料的對象。Consumer 對象是 Rxjava 2.x 才出現的，老版本沒有。

map 操作符

map基本算是 RxJava 中一個最簡單的操作符了，熟悉 RxJava 1.x 的知道，它的作用是對發射時間發送的每一個事件應用一個函數，是的每一個事件都按照指定的函數去變化，而在2.x中它的作用幾乎一緻。

Observable.create(new ObservableOnSubscribe<Integer>() {
       @Override
       public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
           e.onNext(1);
           e.onNext(2);
           e.onNext(3);
       }
   })
   .map(new Function<Integer, String>() {
       @Override
       public String apply(@NonNull Integer integer) throws Exception {
           // map 操作符，就是轉換輸入、輸出 的類型；本例中輸入是 Integer , 輸出是 String 類型
           Log.e("zhao", "apply: " + integer + "  線程：" + Thread.currentThread().getName());
           return "This is result " + integer;
       }
   })
   .subscribeOn(Schedulers.io()) //在子線程發射
   .observeOn(AndroidSchedulers.mainThread())  //在主線程接收
   .subscribe(new Consumer<String>() {
         @Override
         public void accept(@NonNull String s) throws Exception {
          Log.e("zhao", "accept: " + s + "  線程：" + Thread.currentThread().getName());
   }
});

E/zhao: apply: 1  線程：RxCachedThreadScheduler-1
E/zhao: apply: 2  線程：RxCachedThreadScheduler-1
E/zhao: apply: 3  線程：RxCachedThreadScheduler-1
E/zhao: accept: This is result 1  線程：main
E/zhao: accept: This is result 2  線程：main
E/zhao: accept: This is result 3  線程：main

flatMap 操作符

FlatMap 是一個很有趣的東西，我堅信你在實際開發中會經常用到。它可以把一個發射器Observable 通過某種方法轉換為多個Observables，然後再把這些分散的Observables裝進一個單一的發射器Observable。但有個需要注意的是，flatMap并不能保證事件的順序，如果需要保證，需要用到我們下面要講的ConcatMap。

Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
            e.onNext(1);
            e.onNext(2);
            e.onNext(3);
            }
        })
           .flatMap(new Function<Integer, ObservableSource<String>>() {
               @Override
               public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
                   List<String> list = new ArrayList<>();
                   for (int i = 0; i < 3; i++) {
                       list.add("I am value " + integer);
                   }
                   //随機生成一個時間
                   int delayTime = (int) (1 + Math.random() * 10);
                   return Observable.fromIterable(list).delay(delayTime, TimeUnit.MILLISECONDS);
               }
           })
           .subscribe(new Consumer<String>() {
              @Override
               public void accept(String s) throws Exception {
                   Log.e("zhao", "accept: " + s);
               }
           });

E/zhao: accept: I am value 1
E/zhao: accept: I am value 3
E/zhao: accept: I am value 3
E/zhao: accept: I am value 3
E/zhao: accept: I am value 1
E/zhao: accept: I am value 1
E/zhao: accept: I am value 2
E/zhao: accept: I am value 2
E/zhao: accept: I am value 2

一切都如我們預期中的有意思，為了區分concatMap（下一個會講），我在代碼中特意動了一點小手腳，我采用一個随機數，生成一個時間，然後通過delay（後面會講）操作符，做一個小延時操作，而檢視Log日志也确認驗證了我們上面的說法，它是無序的。

concatMap 操作符

上面其實就說了，concatMap 與 FlatMap 的唯一差別就是 concatMap 保證了順序，是以，我們就直接把 flatMap 替換為 concatMap 驗證吧。

Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
            e.onNext(1);
            e.onNext(2);
            e.onNext(3);
            }
        })
           .concatMap(new Function<Integer, ObservableSource<String>>() {
               @Override
               public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
                   List<String> list = new ArrayList<>();
                   for (int i = 0; i < 3; i++) {
                       list.add("I am value " + integer);
                   }
                   //随機生成一個時間
                   int delayTime = (int) (1 + Math.random() * 10);
                   return Observable.fromIterable(list).delay(delayTime, TimeUnit.MILLISECONDS);
               }
           })
           .subscribe(new Consumer<String>() {
              @Override
               public void accept(String s) throws Exception {
                   Log.e("zhao", "accept: " + s);
               }
           });

E/zhao: accept: I am value 1
E/zhao: accept: I am value 1
E/zhao: accept: I am value 1
E/zhao: accept: I am value 2
E/zhao: accept: I am value 2
E/zhao: accept: I am value 2
E/zhao: accept: I am value 3
E/zhao: accept: I am value 3
E/zhao: accept: I am value 3

zip 操作符

建構一個 String 發射器和 Integer 發射器

//建立 String 發射器
private Observable<String> getStringObservable() {
      return Observable.create(new ObservableOnSubscribe<String>() {
          @Override
          public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                e.onNext("A");
                e.onNext("B");
                e.onNext("C");
            }
        });
    }

//建立 String 發射器
private Observable<Integer> getIntegerObservable() {
      return Observable.create(new ObservableOnSubscribe<Integer>() {
          @Override
          public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
              e.onNext(1);
              e.onNext(2);
              e.onNext(3);
              e.onNext(4);
              e.onNext(5);
          }
      });
  }

使用 zip 操作符

Observable.zip(getStringObservable(), getIntegerObservable(), new BiFunction<String, Integer, String>() {
       @Override
       public String apply(@NonNull String s, @NonNull Integer integer) throws Exception {
           return s + integer;
          }
      })
            .subscribe(new Consumer<String>() {
                @Override
                public void accept(String s) throws Exception {
                    Log.e("zhao", "accept: " + s);
                }
            });

E/zhao: accept: A1
E/zhao: accept: B2
E/zhao: accept: C3

zip 組合事件的過程就是分别從發射器A和發射器B各取出一個事件來組合，并且一個事件隻能被使用一次，組合的順序是嚴格按照事件發送的順序來進行的，是以上面截圖中，可以看到，1永遠是和A 結合的，2永遠是和B結合的。
最終接收器收到的事件數量是和發送器發送事件最少的那個發送器的發送事件數目相同，是以如截圖中，5很孤單，沒有人願意和它交往，孤獨終老的單身狗。

interval 操作符

interval操作符是每隔一段時間就産生一個數字，這些數字從0開始，一次遞增1直至無窮大

//方法1 
Flowable.interval(1, TimeUnit.SECONDS)
        .subscribe(new Consumer<Long>() {
             @Override
             public void accept(@NonNull Long aLong) throws Exception {
                 Log.e("zhao", "accept11>: " + aLong);
              }
      });

//方法2 
Observable.interval(1, TimeUnit.SECONDS)
          .subscribe(new Consumer<Long>() {
              @Override
              public void accept(Long aLong) throws Exception {
                  Log.e("zhao", "accept:22> " + aLong);
              }
      });

E/zhao: accept11>: 0
E/zhao: accept11>: 1
E/zhao: accept11>: 2
E/zhao: accept11>: 3
E/zhao: accept11>: 4

倒計時

既然 interval 操作符會産生從 0 到無窮大的序列，那麼我們我們會傳回來思考一下，如果倒過來想，就會發現可以用 interval 方法，實作一個倒計時的功能。

建立一個倒計時的 Observable

/**
  * 産生一個倒計時的 Observable
  * @param time
  * @return
  */
  
public Observable<Long> countdown(final long time) {
      return Observable.interval(1, TimeUnit.SECONDS)
             .map(new Function<Long, Long>() {
                 @Override
                 public Long apply(@NonNull Long aLong) throws Exception {
                     return time - aLong;
                 }
             }).take( time + 1 );
  }

實作倒計時的功能

countdown(4).subscribe(new Consumer<Long>() {
       @Override
       public void accept(Long aLong) throws Exception {
            Log.e("zhao", "accept: 倒計時： " + aLong);
        }
    });

E/zhao: accept: 倒計時： 4
E/zhao: accept: 倒計時： 3
E/zhao: accept: 倒計時： 2
E/zhao: accept: 倒計時： 1
E/zhao: accept: 倒計時： 0

repeat 操作符：重複的發射資料

repeat 重複地發射資料

repeat( ) //無限重複
repeat( int time ) //設定重複的次數

Observable
         .just(1, 2)
         .repeat( 3 ) //重複3次
         .subscribe(new Consumer<Integer>() {
               @Override
               public void accept(Integer integer) throws Exception {
                   Log.e("zhao", "accept: " + integer);
               }
          });

效果：

E/zhao: accept: 1
E/zhao: accept: 2
E/zhao: accept: 1
E/zhao: accept: 2
E/zhao: accept: 1
E/zhao: accept: 2

range ：發射特定的整數序列

range 發射特定整數序列的 Observable

range( int start , int end ) //start :開始的值， end ：結束的值

要求： end >= start

Observable
           .range( 1 , 5 )
           .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    Log.e("zhao", "accept: " + integer);
                }
           });

E/zhao: accept: 1
E/zhao: accept: 2
E/zhao: accept: 3
E/zhao: accept: 4
E/zhao: accept: 5

fromArray ：周遊數組

Integer[] items = {0, 1, 2, 3, 4, 5};

Observable
        .fromArray(items)
        .subscribe(new Consumer<Integer>() {
             @Override
             public void accept(Integer integer) throws Exception {
                 Log.e("zhao", "accept: " + integer
                );
              }
          });

效果是：

E/zhao: accept: 0
E/zhao: accept: 1
E/zhao: accept: 2
E/zhao: accept: 3
E/zhao: accept: 4
E/zhao: accept: 5

fromIterable : 周遊集合

List<String> list = new ArrayList<>();
list.add("a");
list.add("b");
list.add("c");

Observable
        .fromIterable(list)
        .subscribe(new Consumer<String>() {
             @Override
             public void accept(String s) throws Exception {
                  Log.e("zhao", "accept: " + s);
         }
   });

效果

E/zhao: accept: a
E/zhao: accept: b
E/zhao: accept: c

toList : 把資料轉換成 List 集合

Observable
          .just(1, 2, 3, 4)
          .toList()
          .subscribe(new Consumer<List<Integer>>() {
               @Override
               public void accept(List<Integer> integers) throws Exception {
                 Log.e("zhao", "accept: " + integers);
           }
     });

效果是

accept: [1, 2, 3, 4]

**把數組轉化成 List 集合 **

Integer[] items = {0, 1, 2, 3, 4, 5};
        
Observable
         .fromArray( items )  //周遊數組
         .toList()  //把周遊後的數組轉化成 List 
         .subscribe(new Consumer<List<Integer>>() {
               @Override
               public void accept(List<Integer> integers) throws Exception {
                  Log.e("zhao", "accept: " + integers);
            }
      });

accept: [0, 1, 2, 3, 4, 5]

delay : 延遲發射資料

Observable
          .just(1, 2, 3)
          .delay(3, TimeUnit.SECONDS)  //延遲3秒鐘，然後在發射資料
          .subscribe(new Consumer<Integer>() {
               @Override
               public void accept(Integer integer) throws Exception {
                  Log.e("zhao", "accept: " + integer);
              }
      });

E/zhao: accept: 1
E/zhao: accept: 2
E/zhao: accept: 3

背壓 BackPressure

背壓産生的原因：被觀察者發送消息太快以至于它的操作符或者訂閱者不能及時處理相關的消息。在 Rxjava 1.x 版本很容易就會報錯，使程式發生崩潰。

...
    Caused by: rx.exceptions.MissingBackpressureException
...
...

為了解決這個問題，在RxJava2裡，引入了Flowable這個類：Observable不包含 backpressure 處理，而 Flowable 包含。

下面我們來模拟一個觸發背壓的執行個體 , 發射器每1毫秒發射一個資料，接收器每一秒處理一個資料。資料産生是資料處理的1000 倍。

首先用 RxJava 2.x 版本的 Observable 來實作。

Observable.interval(1, TimeUnit.MILLISECONDS)
          .subscribeOn(Schedulers.io())
          .observeOn(Schedulers.newThread())
          .subscribe(new Consumer<Long>() {
                @Override
                public void accept(Long aLong) throws Exception {
                    Thread.sleep(1000);
                    Log.e("zhao", "onNext: " + aLong);
                }
       });

經過測試，app 很健壯，沒有發生崩潰，日志每1秒列印一次。在上面我們說到 2.x 版本中 Observable 不再支援背壓，發神器生成的資料全部緩存在記憶體中。

Observable ：

不支援 backpressure 處理，不會發生 MissingBackpressureException 異常。
所有沒有處理的資料都緩存在記憶體中，等待被訂閱者處理。
壞處是：當産生的資料過快，記憶體中緩存的資料越來越多，占用大量記憶體。

然後用 RxJava 2.x 版本的 Flowable 來實作。

Flowable.interval(1, TimeUnit.MILLISECONDS)
        .subscribeOn(Schedulers.io())
        .observeOn(Schedulers.newThread())
        .subscribe(new Consumer<Long>() {
              @Override
               public void accept(Long aLong) throws Exception {
                    Thread.sleep(1000);
                    Log.e("zhao", "onNext: " + aLong);
                }
         });

運作起來發生崩潰，崩潰日志如下：

io.reactivex.exceptions.OnErrorNotImplementedException: Can't deliver value 128 due to lack of requests
...
...
  Caused by: io.reactivex.exceptions.MissingBackpressureException: Can't deliver value 128 due to lack of requests

很明顯發生了 MissingBackpressureException 異常 , 128 代表是 Flowable 最多緩存 128 個資料，緩存次超過 128 個資料，就會報錯。可喜的是，Rxjava 已經給我們提供了解決背壓的政策。

onBackpressureDrop

onBackpressureDrop() ：當緩沖區資料滿 128 個時候，再新來的資料就會被丢棄，如果此時有資料被消費了，那麼就會把目前最新産生的資料，放到緩沖區。簡單來說 Drop 就是直接把存不下的事件丢棄。

onBackpressureDrop 測試

Flowable.interval( 1 , TimeUnit.MILLISECONDS)
        .onBackpressureDrop() //onBackpressureDrop 一定要放在 interval 後面否則不會生效
        .subscribeOn(Schedulers.io())
        .observeOn(Schedulers.newThread())
        .subscribe(new Consumer<Long>() {
              @Override
               public void accept(Long aLong) throws Exception {
                   Thread.sleep(1000);
                   Log.e("zhao", "onNext: " + aLong);
               }
       });

E/zhao: onNext: 0
E/zhao: onNext: 1
...
E/zhao: onNext: 126
E/zhao: onNext: 127
E/zhao: onNext: 96129
E/zhao: onNext: 96130
E/zhao: onNext: 96131

從日志上分析來看，發射器發射的 0 ~ 127 總共 128 個資料是連續的，下一個資料就是 96129 ， 128 ~ 96128 的資料被丢棄了。

注意事項

1、onBackpressureDrop 一定要放在 interval 後面否則不會生效

onBackpressureLatest

onBackpressureLatest 就是隻保留最新的事件。

onBackpressureBuffer

onBackpressureBuffer：預設情況下緩存所有的資料，不會丢棄資料，這個方法可以解決背壓問題，但是它有像 Observable 一樣的缺點，緩存資料太多，占用太多記憶體。
onBackpressureBuffer(int capacity) ：設定緩存隊列大小，但是如果緩沖資料超過了設定的值，就會報錯，發生崩潰。

onBackpressureBuffer（int capacity）測試

Flowable.interval( 1 , TimeUnit.MILLISECONDS)
        .onBackpressureBuffer( 1000 ) //設定緩沖隊列大小為 1000
        .subscribeOn(Schedulers.io())
        .observeOn(Schedulers.newThread())
        .subscribe(new Consumer<Long>() {
              @Override
               public void accept(Long aLong) throws Exception {
                  Thread.sleep(1000);
                  Log.e("zhao", "onNext: " + aLong);
               }
          });

運作起來後，過了幾秒鐘，發生崩潰，日志如下：

io.reactivex.exceptions.OnErrorNotImplementedException: Buffer is full
···
Caused by: io.reactivex.exceptions.MissingBackpressureException: Buffer is full

通過日志可以看出，緩沖區已經滿了。

1、onBackpressureBuffer 一定要放在 interval 後面否則不會生效

參考資料

RxJava2 源碼分析

如何形象地描述 RxJava 中的背壓和流控機制？

給初學者的RxJava2.0教程（八）: Flowable緩存

個人微信号：

zhaoyanjun125

，歡迎關注

RxJava 2.x 使用最佳實踐

Rxjava、RxAndroid

create() ：建立

map 操作符

flatMap 操作符

concatMap 操作符

zip 操作符

interval 操作符

repeat 操作符：重複的發射資料

range ：發射特定的整數序列

fromArray ：周遊數組

fromIterable : 周遊集合

toList : 把資料轉換成 List 集合

delay : 延遲發射資料

背壓 BackPressure

onBackpressureDrop

onBackpressureLatest

onBackpressureBuffer

參考資料

繼續閱讀

Rxjava2 Retrofit2 和 mvp再封裝

RxJava2+Retrofit 簡單封裝

RxJava2和Retrofit2配合使用詳解

Retrofit結合RxJava2使用

RxJava(二：Observable 建立，組合與過濾)

RxJava學習筆記之Subscription

RxJava+Retrofit+OkHttp深入淺出-mvp（使用篇）RxJava+Retrofit+OkHttp深入淺出-mvp（使用篇）

retrofit+RxJava+okhttp簡便封裝實作網絡請求（詳解）寫在開頭寫在後面

項目中關于Retrofit2.0+RxJava＋OkHttp的封裝和使用

手寫Rxjava+Retrofit+Mvp手寫Rxjava+Retrofit+Mvp

RxJava學習筆記Rxjava 的原理

RxJava 學習筆記

Rxjava學習筆記

Rxjava的學習筆記Rxjava的學習筆記

RXJava學習筆記（3）

RxJava 學習筆記二

RxJava 2.x 使用最佳實踐

Rxjava、RxAndroid

create() ：建立

map 操作符

flatMap 操作符

concatMap 操作符

zip 操作符

interval 操作符

repeat 操作符：重複的發射資料

range ：發射特定的整數序列

fromArray ： 周遊數組

fromIterable : 周遊集合

toList : 把資料轉換成 List 集合

delay : 延遲發射資料

背壓 BackPressure

**onBackpressureDrop **

onBackpressureLatest

onBackpressureBuffer

參考資料

繼續閱讀

fromArray ：周遊數組

onBackpressureDrop