## 前言 很快我們就迎來了第二期，上一期我們主要講解了 RxJava 1.x 到 2.x 的變化概覽，相信各位熟練掌握RxJava 1.x的老司機們隨便看一下變化概覽就可以上手RxJava 2.x了，但為了滿足更廣大的年輕一代司機（未來也是老司機），在本節中，我們將學習RxJava 2.x 強大的操作符章節。 【注】以下所有操作符標題都可直接點擊進入官方doc查看。 ## 正題 #### [Create](http://reactivex.io/documentation/operators/create.html) `create` 操作符應該是最常見的操作符了，主要用于產生一個 `Obserable` 被觀察者對象，為了方便大家的認知，以后的教程中統一把被觀察者 `Observable` 稱為發射器（上游事件），觀察者 `Observer` 稱為接收器（下游事件）。  ~~~ Observable.create(new ObservableOnSubscribe<Integer>() { @Override public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception { mRxOperatorsText.append("Observable emit 1" + "\n"); Log.e(TAG, "Observable emit 1" + "\n"); e.onNext(1); mRxOperatorsText.append("Observable emit 2" + "\n"); Log.e(TAG, "Observable emit 2" + "\n"); e.onNext(2); mRxOperatorsText.append("Observable emit 3" + "\n"); Log.e(TAG, "Observable emit 3" + "\n"); e.onNext(3); e.onComplete(); mRxOperatorsText.append("Observable emit 4" + "\n"); Log.e(TAG, "Observable emit 4" + "\n" ); e.onNext(4); } }).subscribe(new Observer<Integer>() { private int i; private Disposable mDisposable; @Override public void onSubscribe(@NonNull Disposable d) { mRxOperatorsText.append("onSubscribe : " + d.isDisposed() + "\n"); Log.e(TAG, "onSubscribe : " + d.isDisposed() + "\n" ); mDisposable = d; } @Override public void onNext(@NonNull Integer integer) { mRxOperatorsText.append("onNext : value : " + integer + "\n"); Log.e(TAG, "onNext : value : " + integer + "\n" ); i++; if (i == 2) { // 在RxJava 2.x 中，新增的Disposable可以做到切斷的操作，讓Observer觀察者不再接收上游事件 mDisposable.dispose(); mRxOperatorsText.append("onNext : isDisposable : " + mDisposable.isDisposed() + "\n"); Log.e(TAG, "onNext : isDisposable : " + mDisposable.isDisposed() + "\n"); } } @Override public void onError(@NonNull Throwable e) { mRxOperatorsText.append("onError : value : " + e.getMessage() + "\n"); Log.e(TAG, "onError : value : " + e.getMessage() + "\n" ); } @Override public void onComplete() { mRxOperatorsText.append("onComplete" + "\n"); Log.e(TAG, "onComplete" + "\n" ); } }); ~~~ 輸出：  需要注意的幾點是： * 在發射事件中，我們在發射了數值 3 之后，直接調用了 `e.onComlete()`，雖然無法接收事件，但發送事件還是繼續的。 * 另外一個值得注意的點是，在 RxJava 2.x 中，可以看到發射事件方法相比 1.x 多了一個 throws Excetion，意味著我們做一些特定操作再也不用 try-catch 了。 * 并且 2.x 中有一個 `Disposable` 概念，這個東西可以直接調用切斷，可以看到，當它的 `isDisposed()` 返回為 false 的時候，接收器能正常接收事件，但當其為 true 的時候，接收器停止了接收。所以可以通過此參數動態控制接收事件了。 #### [Map](http://reactivex.io/documentation/operators/map.html) `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 { return "This is result " + integer; } }).subscribe(new Consumer<String>() { @Override public void accept(@NonNull String s) throws Exception { mRxOperatorsText.append("accept : " + s +"\n"); Log.e(TAG, "accept : " + s +"\n" ); } }); ~~~ 輸出：  是的，`map` 基本作用就是將一個 `Observable` 通過某種函數關系，轉換為另一種 `Observable`，上面例子中就是把我們的 `Integer` 數據變成了 `String` 類型。從Log日志顯而易見。 #### [Zip](http://reactivex.io/documentation/operators/zip.html) `zip` 專用于合并事件，該合并不是連接（連接操作符后面會說），而是兩兩配對，也就意味著，最終配對出的 `Observable` 發射事件數目只和少的那個相同。 ~~~ 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(@NonNull String s) throws Exception { mRxOperatorsText.append("zip : accept : " + s + "\n"); Log.e(TAG, "zip : accept : " + s + "\n"); } }); ~~~ ~~~ private Observable<String> getStringObservable() { return Observable.create(new ObservableOnSubscribe<String>() { @Override public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception { if (!e.isDisposed()) { e.onNext("A"); mRxOperatorsText.append("String emit : A \n"); Log.e(TAG, "String emit : A \n"); e.onNext("B"); mRxOperatorsText.append("String emit : B \n"); Log.e(TAG, "String emit : B \n"); e.onNext("C"); mRxOperatorsText.append("String emit : C \n"); Log.e(TAG, "String emit : C \n"); } } }); } private Observable<Integer> getIntegerObservable() { return Observable.create(new ObservableOnSubscribe<Integer>() { @Override public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception { if (!e.isDisposed()) { e.onNext(1); mRxOperatorsText.append("Integer emit : 1 \n"); Log.e(TAG, "Integer emit : 1 \n"); e.onNext(2); mRxOperatorsText.append("Integer emit : 2 \n"); Log.e(TAG, "Integer emit : 2 \n"); e.onNext(3); mRxOperatorsText.append("Integer emit : 3 \n"); Log.e(TAG, "Integer emit : 3 \n"); e.onNext(4); mRxOperatorsText.append("Integer emit : 4 \n"); Log.e(TAG, "Integer emit : 4 \n"); e.onNext(5); mRxOperatorsText.append("Integer emit : 5 \n"); Log.e(TAG, "Integer emit : 5 \n"); } } }); } ~~~ 輸出：  需要注意的是： * `zip` 組合事件的過程就是分別從發射器 A 和發射器 B 各取出一個事件來組合，并且一個事件只能被使用一次，組合的順序是嚴格按照事件發送的順序來進行的，所以上面截圖中，可以看到，1 永遠是和 A 結合的，2 永遠是和 B 結合的。 * 最終接收器收到的事件數量是和發送器發送事件最少的那個發送器的發送事件數目相同，所以如截圖中，5 很孤單，沒有人愿意和它交往，孤獨終老的單身狗。 #### [Concat](http://reactivex.io/documentation/operators/concat.html) 對于單一的把兩個發射器連接成一個發射器，雖然 `zip` 不能完成，但我們還是可以自力更生，官方提供的 `concat` 讓我們的問題得到了完美解決。  ~~~ Observable.concat(Observable.just(1,2,3), Observable.just(4,5,6)) .subscribe(new Consumer<Integer>() { @Override public void accept(@NonNull Integer integer) throws Exception { mRxOperatorsText.append("concat : "+ integer + "\n"); Log.e(TAG, "concat : "+ integer + "\n" ); } }); ~~~ 輸出：  如圖，可以看到。發射器 B 把自己的三個孩子送給了發射器 A，讓他們組合成了一個新的發射器，非常懂事的孩子，有條不紊的排序接收。 #### [FlatMap](http://reactivex.io/documentation/operators/flatmap.html) `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); } }).subscribeOn(Schedulers.newThread()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Consumer<String>() { @Override public void accept(@NonNull String s) throws Exception { Log.e(TAG, "flatMap : accept : " + s + "\n"); mRxOperatorsText.append("flatMap : accept : " + s + "\n"); } }); ~~~ 輸出：  一切都如我們預期中的有意思，為了區分 `concatMap`（下一個會講），我在代碼中特意動了一點小手腳，我采用一個隨機數，生成一個時間，然后通過 `delay`（后面會講）操作符，做一個小延時操作，而查看 Log 日志也確認驗證了我們上面的說法，它是無序的。 #### [concatMap](http://reactivex.io/documentation/operators/flatmap.html) 上面其實就說了，`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); } }).subscribeOn(Schedulers.newThread()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Consumer<String>() { @Override public void accept(@NonNull String s) throws Exception { Log.e(TAG, "flatMap : accept : " + s + "\n"); mRxOperatorsText.append("flatMap : accept : " + s + "\n"); } }); ~~~ 輸出：  結果的確和我們預想的一樣。 ## 寫在最后 好了，這一節就先介紹到這里，下一節我們將學習其它的一些操作符，在操作符講完后再帶大家進入實際情景，希望持續關注，[代碼傳送門](https://github.com/nanchen2251/RxJava2Examples)。 作者：南塵2251 鏈接：http://www.jianshu.com/p/b39afa92807e 來源：簡書 著作權歸作者所有。商業轉載請聯系作者獲得授權，非商業轉載請注明出處。