回到頂部

一、前言

       很快我們就迎來了第二期，上一期我們主要講解了 RxJava 1.x 到 2.x 的變化概覽，相信各位熟練掌握RxJava 1.x的老司機(jī)們隨便看一下變化概覽就可以上手RxJava 2.x了，但為了滿足更廣大的年輕一代司機(jī)（未來也是老司機(jī)），在本節(jié)中，我們將學(xué)習(xí)RxJava 2.x 強(qiáng)大的操作符章節(jié)。

     【注】以下所有操作符標(biāo)題都可直接點(diǎn)擊進(jìn)入官方doc查看。

回到頂部

二、正題

1、Create

      create操作符應(yīng)該是最常見的操作符了，主要用于產(chǎn)生一個Obserable被觀察者對象，為了方便大家的認(rèn)知，以后的教程中統(tǒng)一把被觀察者Observable稱為發(fā)射器（上游事件），觀察者Observer稱為接收器（下游事件）。

 1 Observable.create(new ObservableOnSubscribe<Integer>() { 2             @Override 3             public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception { 4                 mRxOperatorsText.append("Observable emit 1" + "\n"); 5                 Log.e(TAG, "Observable emit 1" + "\n"); 6                 e.onNext(1); 7                 mRxOperatorsText.append("Observable emit 2" + "\n"); 8                 Log.e(TAG, "Observable emit 2" + "\n"); 9                 e.onNext(2);10                 mRxOperatorsText.append("Observable emit 3" + "\n");11                 Log.e(TAG, "Observable emit 3" + "\n");12                 e.onNext(3);13                 e.onComplete();14                 mRxOperatorsText.append("Observable emit 4" + "\n");15                 Log.e(TAG, "Observable emit 4" + "\n" );16                 e.onNext(4);17             }18         }).subscribe(new Observer<Integer>() {19             private int i;20             private Disposable mDisposable;21 22             @Override23             public void onSubscribe(@NonNull Disposable d) {24                 mRxOperatorsText.append("onSubscribe : " + d.isDisposed() + "\n");25                 Log.e(TAG, "onSubscribe : " + d.isDisposed() + "\n" );26                 mDisposable = d;27             }28 29             @Override30             public void onNext(@NonNull Integer integer) {31                 mRxOperatorsText.append("onNext : value : " + integer + "\n");32                 Log.e(TAG, "onNext : value : " + integer + "\n" );33                 i++;34                 if (i == 2) {35                     // 在RxJava 2.x 中，新增的Disposable可以做到切斷的操作，讓Observer觀察者不再接收上游事件36                     mDisposable.dispose();37                     mRxOperatorsText.append("onNext : isDisposable : " + mDisposable.isDisposed() + "\n");38                     Log.e(TAG, "onNext : isDisposable : " + mDisposable.isDisposed() + "\n");39                 }40             }41 42             @Override43             public void onError(@NonNull Throwable e) {44                 mRxOperatorsText.append("onError : value : " + e.getMessage() + "\n");45                 Log.e(TAG, "onError : value : " + e.getMessage() + "\n" );46             }47 48             @Override49             public void onComplete() {50                 mRxOperatorsText.append("onComplete" + "\n");51                 Log.e(TAG, "onComplete" + "\n" );52             }53         });

輸出：

需要注意的幾點(diǎn)是：

1）在發(fā)射事件中，我們在發(fā)射了數(shù)值3之后，直接調(diào)用了e.onComlete()，雖然無法接收事件，但發(fā)送事件還是繼續(xù)的。

2) 另外一個值得注意的點(diǎn)是，在RxJava 2.x中，可以看到發(fā)射事件方法相比1.x多了一個throws Excetion，意味著我們做一些特定操作再也不用try-catch了。

3) 并且2.x 中有一個Disposable概念，這個東西可以直接調(diào)用切斷，可以看到，當(dāng)它的isDisposed()返回為false的時候，接收器能正常接收事件，但當(dāng)其為true的時候，接收器停止了接收。所以可以通過此參數(shù)動態(tài)控制接收事件了。

2、Map

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

 1 Observable.create(new ObservableOnSubscribe<Integer>() { 2             @Override 3             public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception { 4                 e.onNext(1); 5                 e.onNext(2); 6                 e.onNext(3); 7             } 8         }).map(new Function<Integer, String>() { 9             @Override10             public String apply(@NonNull Integer integer) throws Exception {11                 return "This is result " + integer;12             }13         }).subscribe(new Consumer<String>() {14             @Override15             public void accept(@NonNull String s) throws Exception {16                 mRxOperatorsText.append("accept : " + s +"\n");17                 Log.e(TAG, "accept : " + s +"\n" );18             }19         });

輸出：

是的，map基本作用就是將一個Observable通過某種函數(shù)關(guān)系，轉(zhuǎn)換為另一種Observable，上面例子中就是把我們的Integer數(shù)據(jù)變成了String類型。從Log日志顯而易見。

3、Zip

zip專用于合并事件，該合并不是連接（連接操作符后面會說），而是兩兩配對，也就意味著，最終配對出的Observable發(fā)射事件數(shù)目只和少的那個相同。

 1 Observable.zip(getStringObservable(), getIntegerObservable(), new BiFunction<String, Integer, String>() { 2             @Override 3             public String apply(@NonNull String s, @NonNull Integer integer) throws Exception { 4                 return s + integer; 5             } 6         }).subscribe(new Consumer<String>() { 7             @Override 8             public void accept(@NonNull String s) throws Exception { 9                 mRxOperatorsText.append("zip : accept : " + s + "\n");10                 Log.e(TAG, "zip : accept : " + s + "\n");11             }12         });

 1 private Observable<String> getStringObservable() { 2         return Observable.create(new ObservableOnSubscribe<String>() { 3             @Override 4             public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception { 5                 if (!e.isDisposed()) { 6                     e.onNext("A"); 7                     mRxOperatorsText.append("String emit : A \n"); 8                     Log.e(TAG, "String emit : A \n"); 9                     e.onNext("B");10                     mRxOperatorsText.append("String emit : B \n");11                     Log.e(TAG, "String emit : B \n");12                     e.onNext("C");13                     mRxOperatorsText.append("String emit : C \n");14                     Log.e(TAG, "String emit : C \n");15                 }16             }17         });18     }19 20     private Observable<Integer> getIntegerObservable() {21         return Observable.create(new ObservableOnSubscribe<Integer>() {22             @Override23             public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {24                 if (!e.isDisposed()) {25                     e.onNext(1);26                     mRxOperatorsText.append("Integer emit : 1 \n");27                     Log.e(TAG, "Integer emit : 1 \n");28                     e.onNext(2);29                     mRxOperatorsText.append("Integer emit : 2 \n");30                     Log.e(TAG, "Integer emit : 2 \n");31                     e.onNext(3);32                     mRxOperatorsText.append("Integer emit : 3 \n");33                     Log.e(TAG, "Integer emit : 3 \n");34                     e.onNext(4);35                     mRxOperatorsText.append("Integer emit : 4 \n");36                     Log.e(TAG, "Integer emit : 4 \n");37                     e.onNext(5);38                     mRxOperatorsText.append("Integer emit : 5 \n");39                     Log.e(TAG, "Integer emit : 5 \n");40                 }41             }42         });43     }

輸出：

需要注意的是：

1) zip 組合事件的過程就是分別從發(fā)射器A和發(fā)射器B各取出一個事件來組合，并且一個事件只能被使用一次，組合的順序是嚴(yán)格按照事件發(fā)送的順序來進(jìn)行的，所以上面截圖中，可以看到，1永遠(yuǎn)是和A 結(jié)合的，2永遠(yuǎn)是和B結(jié)合的。

2) 最終接收器收到的事件數(shù)量是和發(fā)送器發(fā)送事件最少的那個發(fā)送器的發(fā)送事件數(shù)目相同，所以如截圖中，5很孤單，沒有人愿意和它交往，孤獨(dú)終老的單身狗。

4、Concat

對于單一的把兩個發(fā)射器連接成一個發(fā)射器，雖然 zip 不能完成，但我們還是可以自力更生，官方提供的 concat 讓我們的問題得到了完美解決。

1 Observable.concat(Observable.just(1,2,3), Observable.just(4,5,6))2                 .subscribe(new Consumer<Integer>() {3                     @Override4                     public void accept(@NonNull Integer integer) throws Exception {5                         mRxOperatorsText.append("concat : "+ integer + "\n");6                         Log.e(TAG, "concat : "+ integer + "\n" );7                     }8                 });

輸出：

如圖，可以看到。發(fā)射器B把自己的三個孩子送給了發(fā)射器A，讓他們組合成了一個新的發(fā)射器，非常懂事的孩子，有條不紊的排序接收。

5、FlatMap

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

 1  Observable.create(new ObservableOnSubscribe<Integer>() { 2             @Override 3             public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception { 4                 e.onNext(1); 5                 e.onNext(2); 6                 e.onNext(3); 7             } 8         }).flatMap(new Function<Integer, ObservableSource<String>>() { 9             @Override10             public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {11                 List<String> list = new ArrayList<>();12                 for (int i = 0; i < 3; i++) {13                     list.add("I am value " + integer);14                 }15                 int delayTime = (int) (1 + Math.random() * 10);16                 return Observable.fromIterable(list).delay(delayTime, TimeUnit.MILLISECONDS);17             }18         }).subscribeOn(Schedulers.newThread())19                 .observeOn(AndroidSchedu

作　　者： 南 塵 　　 
出　　處： http://www.cnblogs.com/liushilin/ 
關(guān)于作者：專注于移動前端的項(xiàng)目開發(fā)。如有問題或建議，請多多賜教！歡迎加入Android交流群：118116509 
版權(quán)聲明：本文版權(quán)歸作者和博客園共有，歡迎轉(zhuǎn)載，但未經(jīng)作者同意必須保留此段聲明，且在文章頁面明顯位置給出原文鏈接。 
特此聲明：所有評論和私信都會在第一時間回復(fù)。也歡迎園子的大大們指正錯誤，共同進(jìn)步?；蛘?a style="margin: 0px; padding: 0px; color: rgb(0, 0, 0); text-decoration-line: none;">直接私信我 
聲援博主：如果您覺得文章對您有幫助，可以點(diǎn)擊文章下部【推薦】或側(cè)邊【關(guān)注】。您的鼓勵是作者堅(jiān)持原創(chuàng)和持續(xù)寫作的最大動力！ 

http://www.cnblogs.com/liushilin/p/7066074.html

延伸閱讀

• 【知識整理】這可能是最好的RxJava 2.x 教程（完結(jié)版） 2017-07-21
• 【知識整理】這可能是最好的RxJava 2.x 入門教程（三） 2017-07-20
• 【知識整理】這可能是最好的RxJava 2.x 入門教程（五） 2017-07-03
• 【知識整理】這可能是最好的RxJava 2.x 入門教程（一） 2017-06-21

【知識整理】這可能是最好的RxJava 2.x 入門教程（三）