一区拳、RxJava操作符概述

RxJava中的操作符就是為了提供函數(shù)式的特性，函數(shù)式最大的好處就是使得數(shù)據(jù)處理簡潔易懂。

操作符實質(zhì)上就是RxJava函數(shù)式編程模式的體現(xiàn)写穴，在上篇文章中浪汪，我詳細闡述了RxJava中觀察者模式巴柿、Iterator模式、函數(shù)式編程模式概念的解讀死遭，詳情請戳→文章傳送門广恢。

Rxjava操作符的類型眾多，在本文中呀潭，我詳細解釋如圖1.1所示的9種操作符钉迷。

圖1.1 RxJava操作符的分類

本項目案例代碼已上傳Github至非，詳情戳→GitHub案例代碼。

圖1.2 RxJava操作符案例圖

二糠聪、RxJava操作符詳解

1荒椭、創(chuàng)建操作符

創(chuàng)建操作符的分類如下圖所示，關(guān)于create操作符的詳細操作可在我的上篇文章中查看舰蟆，在本文中不加以贅述趣惠，文章鏈接戳→文章傳送門。在本文中從from操作符開始介紹身害。

圖2.1? 創(chuàng)建操作符的分類

注：在這里味悄，我將請求的不完整回調(diào)在父類中進行了封裝，具體代碼可查看GitHub的代碼鏈接塌鸯。

①from操作符

在這里以發(fā)送數(shù)組為例侍瑟，from操作符的使用代碼如下所示：

//from操作符，創(chuàng)建以數(shù)組內(nèi)容發(fā)送事件的ObservableString[] observableArr =newString[]{"Alex","Payne"};//onNextAction界赔、onErrorAction提取到父類中丢习，具體代碼可查看GitHub的代碼鏈接Observable.from(observableArr).subscribe(onNextAction, onErrorAction);

首先，我們查看如下所示from操作符的結(jié)構(gòu)圖淮悼，可以看到它有多種實現(xiàn)方式咐低，但是有一個共同點，都會返回一個Observable對象袜腥。

圖2.1.1 from操作符操作結(jié)構(gòu)圖

實質(zhì)上见擦，Observable將數(shù)組中的元素逐個進行發(fā)送，在發(fā)送過程中轉(zhuǎn)換為Observable對象羹令。

進一步查看源碼鲤屡，可得知from操作符的作用：將一個Iterable、一個Future福侈、 或者一個數(shù)組酒来，內(nèi)部通過代理的方式轉(zhuǎn)換成一個Observable。

Future轉(zhuǎn)換為OnSubscribe是通過OnSubscribeToObservableFuture進行的肪凛，Iterable轉(zhuǎn)換通過OnSubscribeFromIterable進行堰汉。數(shù)組通過OnSubscribeFromArray轉(zhuǎn)換。

②just操作符

使用代碼如下所示：

//just操作符伟墙，創(chuàng)建將逐個內(nèi)容進行發(fā)送的Observable翘鸭，其內(nèi)部發(fā)送內(nèi)容在內(nèi)部以from的操作符的方式進行轉(zhuǎn)換Observable.just("Alex","Payne").subscribe(onNextAction);

圖2.1.2 just操作符結(jié)構(gòu)圖

查看just操作符的結(jié)構(gòu)圖，結(jié)合源碼得知戳葵，just操作符將單個參數(shù)發(fā)送的內(nèi)容通過ScalarSynchronousObservable轉(zhuǎn)換為一個新的Observable對象就乓，而將多個參數(shù)發(fā)送的內(nèi)容轉(zhuǎn)換為一個數(shù)組，然后將數(shù)組通過from操作符進行發(fā)送。

③interval操作符

interval操作符使用代碼如下所示：

//interval操作符,創(chuàng)建以1秒為事件間隔發(fā)送整數(shù)序列的ObservableObservable.interval(1, TimeUnit.SECONDS, AndroidSchedulers.mainThread()).subscribe(onNextAction);

圖2.1.3 interval操作符結(jié)構(gòu)圖

查看interval的結(jié)構(gòu)圖生蚁，其只能發(fā)送Long類型的數(shù)噩翠，實質(zhì)上其作用為：創(chuàng)建一個按固定時間間隔發(fā)射整數(shù)序列的Observable，這個序列為一個無限遞增的整數(shù)序列守伸。

需要注意的是：interval默認在computation調(diào)度器上執(zhí)行绎秒。你也可以傳遞一個可選的Scheduler參數(shù)來指定調(diào)度器。

④range操作符

range操作符使用的代碼如下所示：

//range操作符尼摹，創(chuàng)建以發(fā)送范圍內(nèi)的整數(shù)序列的ObservableObservable.range(0,3).subscribe(onNextAction);

range操作符發(fā)射一個范圍內(nèi)的有序整數(shù)序列见芹，并且我們可以指定范圍的起始和長度

圖2.1.4 range操作符參數(shù)釋義

⑤repeat操作符

repeat操作符使用的代碼如下所示：

//repeat操作符,創(chuàng)建一個以N次重復發(fā)送數(shù)據(jù)的ObservableObservable.range(0,3).repeat(2).subscribe(onNextAction);

在這里需要強調(diào)一下，它不是創(chuàng)建一個Observable蠢涝，而是重復發(fā)射原始Observable的數(shù)據(jù)序列玄呛，這個序列或者是無限的，或者通過repeat(n)指定重復次數(shù)和二。

2徘铝、變換操作符

在這里我介紹如下圖所示7種變換操作符，變換操作符的作用是將源Observable發(fā)送的數(shù)據(jù)進行變換惯吕。

圖2.1 變換操作符的分類

①map操作符

map操作符使用的代碼如下所示：

//map操作符惕它，通過指定一個Func，將Observable轉(zhuǎn)換為另一個Observable對象并發(fā)送Observable.just("Alex_Payne")? ? ? ? ? ? ? ? ? .map(newFunc1() {? ? ? ? ? ? ? ? ? ? ? @Override? ? ? ? ? ? ? ? ? ? ? publicStringcall(Strings) {return"My Name is"+ s;? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? }).subscribe(onNextAction);

map操作符將源Observable發(fā)送的數(shù)據(jù)轉(zhuǎn)換為一個新的Observable對象废登。

在這里淹魄，F(xiàn)unc1和Action的區(qū)別在于，F(xiàn)unc1包裝的是有返回值的方法堡距。另外甲锡，和ActionX 一樣，F(xiàn)uncX 也有多個羽戒，用于不同參數(shù)個數(shù)的方法缤沦。

FuncX 和 ActionX 的區(qū)別在 FuncX 包裝的是有返回值的方法。

②flatMap操作符

flatMap操作符使用的代碼如下所示：

//flatMap操作符易稠，將Observable發(fā)送的數(shù)據(jù)集合轉(zhuǎn)換為Observable集合//flatMap的合并運行允許交叉缸废，允許交錯的發(fā)送事件String[] observableArr = {"Alex","Max","Bruce","Frank","Tom"};? ? ? ? Observable.from(observableArr).flatMap(newFunc1>() {? ? ? ? ? ? ? @Override? ? ? ? ? ? ? public Observable call(Strings) {returnObservable.just("My Name is:"+ s);? ? ? ? ? ? ? }? ? ? ? }).subscribe(onNextAction);

源Observable通過flatMap操作符轉(zhuǎn)換為包含源Observable發(fā)送的所有子條目的Observable集合，可見下圖的示意圖驶社，然后從Observable集合中逐個取出轉(zhuǎn)化為單個Observable對象進行發(fā)送企量。不同于map操作符的一點就是一對多的轉(zhuǎn)化。

圖2.2.2 flatMap轉(zhuǎn)換示意圖

它的應用場景可以體現(xiàn)在源Observable發(fā)送的內(nèi)容為一個復雜的數(shù)據(jù)集衬吆，例如一個Bean對象，而該外層Bean對象中一個成員變量為另一個內(nèi)層Bean對象绳泉，我們想要拆解外層Bean對象獲取內(nèi)層Bean對象逊抡，就可以用flatMap操作符。

注意：FlatMap對這些Observables發(fā)射的數(shù)據(jù)做的是合并(merge)操作，因此它們可能是交錯的冒嫡。

③concatMap操作符

concatMap操作符使用的代碼如下所示：

//concatMap操作符拇勃，將Observable發(fā)送的數(shù)據(jù)集合轉(zhuǎn)換為Observable集合//解決了flatMap的交叉問題，將發(fā)送的數(shù)據(jù)連接發(fā)送String[] observableArr = {"Alex","Max","Bruce","Frank","Tom"};? ? ? ? Observable.from(observableArr).concatMap(newFunc1>() {? ? ? ? ? ? ? @Override? ? ? ? ? ? ? public Observable call(Strings) {returnObservable.just("My Name is:"+ s);? ? ? ? ? ? ? }? ? ? ? }).subscribe(onNextAction);

concatMap操作符類似于flatMap操作符孝凌，不同的一點是它按次序連接方咆。

④cast操作符

cast操作符使用的代碼如下所示：

//cast操作符，將類對象進行轉(zhuǎn)換Object[] objectsArr = {"1","2","3"};? ? ? ? Observable.from(objectsArr).cast(String.class).subscribe(onNextAction);

cast操作符將源Observable發(fā)送的數(shù)據(jù)都強制轉(zhuǎn)換為一個指定的類型蟀架，然后再發(fā)射數(shù)據(jù)瓣赂。

需強調(diào)的一點是只能由父類對象轉(zhuǎn)換為子類對象，否則會報錯片拍。

⑤flatMapIterable操作符

flatMapIterable操作符使用的代碼如下所示：

//將數(shù)據(jù)集合轉(zhuǎn)換為Iterable煌集，在Iterable中對數(shù)據(jù)進行處理Observable.just(1,2,3).flatMapIterable(newFunc1>() {@OverridepublicIterablecall(Integer number){? ? ? ? ? ? ? ? ? ArrayList mList =newArrayList<>();? ? ? ? ? ? ? ? ? mList.add(1000+ number);returnmList;? ? ? ? ? ? ? }? ? ? ? ? }).subscribe(onNextAction);

flatMapIterable相當于是flatMap的變體，直接在內(nèi)部以Iterable接口將集合數(shù)據(jù)進行接收捌省，示意圖如下所示：

圖2.2.5 flatMapIterable示意圖

⑥buffer操作符

buffer操作符使用的代碼如下所示：

//buffer操作符苫纤，將原有Observable轉(zhuǎn)換為一個新的Observable，這個新的Observable每次發(fā)送一組值纲缓，而不是一個個進行發(fā)送Observable.just(1,2,3,4,5,6)? ? ? ? ? ? ? .buffer(3).subscribe(newAction1>() {@Overridepublicvoidcall(List mList){for(Integer i : mList) {? ? ? ? ? ? ? ? ? ? ? ? Toast.makeText(getActivity(),"new Number i is:"+ i, Toast.LENGTH_SHORT).show();? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? Toast.makeText(getActivity(),"Another request is called", Toast.LENGTH_SHORT).show();? ? ? ? ? ? ? ? }? ? ? ? ? ? });

buffer操作符將原有Observable轉(zhuǎn)換為一個新的Observable卷拘，這個新的Observable每次發(fā)送一組值，而不是一個個進行發(fā)送祝高，我們可以定義這個新的Observable存放幾個原有的Observable對象栗弟。

圖2.2.6 buffer操作符示意圖

⑦groupBy操作符

groupBy操作符使用的代碼如下所示：

//groupBy操作符，可以做分組操作Observable.range(0,10).groupBy(newFunc1() {@OverridepublicIntegercall(Integer num){returnnum %3;? ? ? ? ? ? ? ? ? }? ? ? ? ? ? }).subscribe(newAction1>() {@Overridepublicvoidcall(finalGroupedObservable groupedObservable){? ? ? ? ? ? ? ? ? groupedObservable.subscribe(newAction1() {@Overridepublicvoidcall(Integer num){? ? ? ? ? ? ? ? ? ? ? ? ? Toast.makeText(getActivity(),"當前的組別是："+ groupedObservable.getKey() +"組別內(nèi)的數(shù)字是："+ num, Toast.LENGTH_SHORT).show();? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? });? ? ? ? ? ? ? }? ? ? ? ? });

groupBy操作符褂策，將原有的Observable對象轉(zhuǎn)換為發(fā)送一組Observable集合的GroupedObservable對象横腿，可以做分組操作，GroupedObservable將分組完畢的Observable對象可以繼續(xù)發(fā)送斤寂。

注意：groupBy將原始Observable分解為一個發(fā)射多個GroupedObservable的Observable耿焊，一旦有訂閱，每個GroupedObservable就開始緩存數(shù)據(jù)遍搞。因此罗侯，如果你忽略這些GroupedObservable中的任何一個，這個緩存可能形成一個潛在的內(nèi)存泄露溪猿。因此钩杰，如果你不想觀察，也不要忽略GroupedObservable诊县。你應該使用像take(0)這樣會丟棄自己的緩存的操作符讲弄。

3、過濾操作符

過濾操作符用于從Observable發(fā)射的數(shù)據(jù)中進行選擇依痊，在這里介紹如下圖所示的8種避除。

圖3.1 過濾操作符的分類

①filter操作符

filter操作符使用的代碼如下所示：

//filter過濾操作符，對Observable發(fā)送的內(nèi)容根據(jù)自定義的規(guī)則進行過濾Observable.range(0,5).filter(newFunc1() {@OverridepublicBooleancall(Integer num){returnnum >2;//自定義的條件,只有符合條件的結(jié)果才會提交給觀察者}? ? ? ? ? }).subscribe(onNextAction);

filter默認不在任何特定的調(diào)度器上執(zhí)行。

②elementAt操作符

elementAt操作符使用的代碼如下所示：

//elementAt操作符瓶摆，用于返回指定位置后一位的數(shù)據(jù)凉逛，即腳標+1的數(shù)據(jù)//在這里發(fā)送0、1群井、2状飞、3、4书斜，腳標為3的數(shù)據(jù)為2诬辈，發(fā)送其后一位數(shù)據(jù)3Observable.range(0,5).elementAt(3).subscribe(onNextAction);

elementAt操作符獲取原始Observable發(fā)射的數(shù)據(jù)序列指定索引位置的數(shù)據(jù)項，然后當做自己的唯一數(shù)據(jù)發(fā)射菩佑。對應示意圖如下：

圖2.3.2 elementAt操作符示意圖

③distinct操作符

distinct操作符使用的代碼如下所示：

//distinct操作符自晰，用于Observable發(fā)送的元素的去重Observable.just(1,1,2,2,2,3).distinct().subscribe(onNextAction);

在這里需要強調(diào)一點：distinct操作符只允許還沒有發(fā)射過的數(shù)據(jù)項通過。

④skip操作符

skip操作符使用的代碼如下所示：

//skip操作符稍坯，用于Observable發(fā)送的元素前N項去除掉Observable.range(0,5).skip(2).subscribe(onNextAction);

skip操作符抑制Observable發(fā)射的前N項數(shù)據(jù)酬荞，只發(fā)送后N項數(shù)據(jù)

圖2.3.4 skip操作符示意圖

⑤take操作符

//take操作符，用于Observable發(fā)送的元素只取前N項Observable.range(0,5).take(2).subscribe(onNextAction);

圖2.3.5 take操作符示意圖

⑥ignoreElements操作符

//ignoreElements操作符瞧哟，忽略掉源Observable發(fā)送的結(jié)果混巧，只把Observable的onCompleted或onError發(fā)送Observable.range(0,5).ignoreElements().subscribe(onNextAction, onErrorAction, onCompletedAction);

IgnoreElements操作符抑制原始Observable發(fā)射的所有數(shù)據(jù)，只允許它的終止通知（onError或onCompleted）進行發(fā)送勤揩。

⑦throttleFirst操作符

//throttleFirst操作符咧党，會定期發(fā)送這個時間段里源Observable發(fā)送的第一個數(shù)據(jù)//throttleFirst操作符默認在computaioin調(diào)度器上執(zhí)行，其他的數(shù)據(jù)都會被過濾掉Observable.create(newObservable.OnSubscribe() {@Overridepublicvoidcall(Subscriber subscriber){for(inti =0; i <10; i++) {? ? ? ? ? ? ? ? ? ? ? ? ? ? subscriber.onNext(i);//線程休眠100毫秒try{? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? Thread.sleep(100);? ? ? ? ? ? ? ? ? ? ? ? ? ? }catch(InterruptedException e) {? ? ? ? ? ? ? ? ? ? ? ? ? ? e.printStackTrace();? ? ? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? }? ? ? ? ? })? ? ? ? ? .throttleFirst(200, TimeUnit.MILLISECONDS)? ? ? ? ? .subscribe(onNextAction);

throttleFirst操作符會按照固定的時間間隔將信息進行發(fā)送陨亡。在這里我設置的事件間隔為200毫秒傍衡，其中每發(fā)送一個數(shù)據(jù)線程休眠100毫秒，所以最后會顯示的數(shù)據(jù)為0,示意圖如下：

圖2..3.7 throttleFirst操作符示意圖

注：throttleFirst操作符默認在computation調(diào)度器上執(zhí)行负蠕，但是你可以使用第三個參數(shù)指定其它的調(diào)度器蛙埂。

⑧throttleWithTimeOut操作符

//throttleWithTimeout操作符//源發(fā)射數(shù)據(jù)時，如果兩次數(shù)據(jù)的發(fā)射間隔小于指定時間遮糖，就會丟棄前一次的數(shù)據(jù),直到指定時間內(nèi)都沒有新數(shù)據(jù)發(fā)射時才進行發(fā)射? ? ? ? ? Observable.create(newObservable.OnSubscribe() {@Overridepublicvoidcall(Subscriber subscriber){? ? ? ? ? ? ? ? ? ? ? ? subscriber.onNext(1);try{? ? ? ? ? ? ? ? ? ? ? ? ? ? ? Thread.sleep(500);? ? ? ? ? ? ? ? ? ? ? ? ? }catch(InterruptedException e) {throwExceptions.propagate(e);? ? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? ? ? subscriber.onNext(2);try{? ? ? ? ? ? ? ? ? ? ? ? ? ? ? Thread.sleep(500);? ? ? ? ? ? ? ? ? ? ? ? ? }catch(InterruptedException e) {throwExceptions.propagate(e);? ? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? ? ? ? subscriber.onNext(3);try{? ? ? ? ? ? ? ? ? ? ? ? ? ? Thread.sleep(1000);? ? ? ? ? ? ? ? ? ? ? ? ? }catch(InterruptedException e) {throwExceptions.propagate(e);? ? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? ? ? ? subscriber.onNext(4);? ? ? ? ? ? ? ? ? ? ? ? ? subscriber.onNext(5);? ? ? ? ? ? ? ? ? ? ? ? ? subscriber.onCompleted();? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? })? ? ? ? ? ? ? ? .throttleWithTimeout(800, TimeUnit.MILLISECONDS)? ? ? ? ? ? ? ? .subscribeOn(Schedulers.newThread())? ? ? ? ? ? ? ? .observeOn(AndroidSchedulers.mainThread())? ? ? ? ? ? ? ? .subscribe(onNextAction);

在這里绣的，我設置的時間間隔指定為800毫秒，所以最后顯示的數(shù)據(jù)是有3欲账、4屡江、5。

4赛不、組合操作符

組合操作符用于將多個Observable組合成一個單一的Observable惩嘉，在這里我介紹如下圖所示5種操作符：

圖4.1 組合操作符分類

①startWith組合操作符

startWith組合操作符使用的代碼如下所示：

//startWith操作符，會在發(fā)送的數(shù)據(jù)之前插入數(shù)據(jù)Observable.range(3,5).startWith(0,10086).subscribe(onNextAction);

很簡單踢故，會在發(fā)送的數(shù)據(jù)序列前插入數(shù)據(jù)序列文黎，并且會發(fā)送插入的數(shù)據(jù)序列奏路。

②merge組合操作符

merge組合操作符使用的代碼如下所示：

//merge操作符，會將多個Observable對象合并到一個Observable對象中進行發(fā)送? ? ? ? ? ? ? ? ObservablefirstObservable = Observable.just(0, 1, 2).subscribeOn(Schedulers.io());? ? ? ? ? ? ? ? ObservablesecondObservable = Observable.just(3, 4, 5);? ? ? ? ? ? ? ? Observable.merge(firstObservable, secondObservable).subscribe(onNextAction, onErrorAction);

如下圖所示臊诊，merge操作符會將多個Observable對象進行合并。

圖2.4.2 merge操作符示意圖

需要注意的是：merge可能會讓合并的Observables發(fā)射的數(shù)據(jù)交錯斜脂。

在這里我將firstObservable指定在IO線程中進行發(fā)送抓艳，secondObservable沒有指定線程，兩者合并然后發(fā)送數(shù)據(jù)時便會產(chǎn)生數(shù)據(jù)交錯的現(xiàn)象帚戳。

③concat組合操作符

concat組合操作符使用的代碼如下所示：

//concat操作符玷或，會將多個Observable對象合并到一個Observable對象中進行發(fā)送，嚴格按照順序進行發(fā)送? ? ? ? ? ? ? ? ObservablefirstObservable = Observable.just(0, 1, 2).subscribeOn(Schedulers.io());? ? ? ? ? ? ? ? ObservablesecondObservable = Observable.just(3, 4, 5);? ? ? ? ? ? ? ? Observable.concat(firstObservable, secondObservable)? ? ? ? ? ? ? ? ? ? ? ? .subscribeOn(Schedulers.io())? ? ? ? ? ? ? ? ? ? ? ? .observeOn(AndroidSchedulers.mainThread())? ? ? ? ? ? ? ? ? ? ? ? .subscribe(onNextAction);

concat操作符不同于merge操作符的區(qū)別就是：會將多個Observable對象合并到一個Observable對象中進行發(fā)送片任，嚴格按照順序進行發(fā)送偏友。如下圖所示，直到第一個Observable發(fā)送完畢數(shù)據(jù)后对供，第二個Observable才會進行數(shù)據(jù)的發(fā)送位他。

圖2.4.3 concat組合操作符

④zip組合操作符

zip組合操作符使用的代碼如下所示：

//zip操作符，會將多個Observable對象轉(zhuǎn)換成一個Observable對象然后進行發(fā)送产场，轉(zhuǎn)換關(guān)系可根據(jù)需求自定義Observable integerObservable = Observable.range(0,4);? ? ? ? ? ? ? ? Observable stringObservable = Observable.just("a","b","c","d");? ? ? ? ? ? ? ? Observable.zip(integerObservable, stringObservable,newFunc2() {? ? ? ? ? ? ? ? ? ? @Override? ? ? ? ? ? ? ? ? ? publicStringcall(Integer num,Stringinfo) {//在這里的轉(zhuǎn)換關(guān)系為將數(shù)字與字串內(nèi)容進行拼接return"數(shù)字為:"+ num +"……字符為："+ info;? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? }).subscribe(onNextAction);

image.png

zip操作符返回一個Obversable鹅髓，它使用這個函數(shù)按順序結(jié)合兩個或多個Observables發(fā)射的數(shù)據(jù)項，然后它發(fā)射這個函數(shù)返回的結(jié)果京景。

它按照嚴格的順序進行數(shù)據(jù)發(fā)送窿冯。它只發(fā)射與發(fā)射數(shù)據(jù)項最少的那個Observable一樣多的數(shù)據(jù)。

⑤combineLastest組合操作符

combineLastest組合操作符使用的代碼如下所示：

//combineLastest操作符确徙，會將多個Observable對象轉(zhuǎn)換一個Observable對象然后進行發(fā)送醒串，轉(zhuǎn)換關(guān)系可以根據(jù)需求自定義//不同于zip操作符的是，會將最新發(fā)送的數(shù)據(jù)組合到一起integerObservable = Observable.just(1,2,3);? ? ? ? ? ? ? ? stringObservable = Observable.just("a","b","c");? ? ? ? ? ? ? ? Observable.combineLatest(integerObservable, stringObservable,newFunc2() {? ? ? ? ? ? ? ? ? ? @Override? ? ? ? ? ? ? ? ? ? publicStringcall(Integer num,Stringinfo) {//在這里的轉(zhuǎn)換關(guān)系為將數(shù)字與字串內(nèi)容進行拼接return"數(shù)字為:"+ num +"……字符為："+ info;? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? }).subscribe(onNextAction);

當兩個Observables中的任何一個發(fā)射了數(shù)據(jù)時鄙皇，使用一個函數(shù)結(jié)合每個Observable發(fā)射的最近數(shù)據(jù)項芜赌，并且基于這個函數(shù)的結(jié)果發(fā)射數(shù)據(jù)∮叮可在本案例代碼中進行驗證较鼓。

5、輔助操作符

輔助操作符就是處理Observable的幫助動作违柏，在這里介紹如下5種輔助操作符博烂。

圖2.5 輔助操作符分類

①delay操作符

delay操作符使用的代碼如下所示：

//delay操作符可以讓源Observable對象發(fā)送數(shù)據(jù)之前暫停一段制定的時間Observable.just(1,2,3)? ? ? ? ? ? ? ? ? ? ? ? ? .delay(2, TimeUnit.SECONDS)? ? ? ? ? ? ? ? ? ? ? ? ? .observeOn(AndroidSchedulers.mainThread())? ? ? ? ? ? ? ? ? ? ? ? ? .subscribe(onNextAction);

在這里我將延時時間設置為2秒，延遲指定的時間后發(fā)射源Observable中的數(shù)據(jù)漱竖。

②do操作符

do操作符使用的代碼如下所示：

//doOnNext是do操作符中的一種? ? ? ? ? ? ? ? Observable.range(0, 3).doOnNext(onNextAction).subscribe(onNextAction);

do操作符,其下細分有很多內(nèi)容禽篱，以doOnNext為例，其作用就是為源Observable對象發(fā)送數(shù)據(jù)后馍惹，當Subscriber接收到數(shù)據(jù)時躺率，即當Subscriber的onNext方法被調(diào)用時玛界，提供回調(diào)相應數(shù)據(jù)。

③subscribeOn輔助操作符

④observeOn輔助操作符

subscribeOn悼吱、observeOn操作符使用的代碼如下所示：

Observable.just("當前的線程ID為" +Thread.currentThread().getName()).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(onNextAction);

subscribeOn操作符慎框，指定subscribe()所發(fā)生的線程，即Observable.OnSubscribe被激活時所處的線程后添”靠荩或者叫做事件產(chǎn)生的線程。

observeOn操作符,指定Subscriber所運行的線程遇西∠诰或者叫做事件消費的線程。

上篇文章中我提到Schedulers可以使得RxJava實現(xiàn)線程切換粱檀，實質(zhì)上就是借助于lift變換方法進行轉(zhuǎn)換洲敢，subscribeOn發(fā)生在下圖的通知過程，observeOn發(fā)生在下圖中的發(fā)送過程茄蚯。

圖2.5.4 Observable與Subscriber的轉(zhuǎn)換關(guān)系圖

⑤timeout輔助操作符

timeout操作符使用的代碼如下所示：

//timeout操作符压彭，如果源Observable對象過了一段時間沒有發(fā)送數(shù)據(jù),timeout會以onError通知終止這個ObservableObservable.create(newObservable.OnSubscribe() {@Overridepublicvoidcall(Subscriber subscriber){for(inti =0; i <5; i++) {try{? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? Thread.sleep(i *100);? ? ? ? ? ? ? ? ? ? ? ? ? ? }catch(InterruptedException e) {? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? e.printStackTrace();? ? ? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? ? ? ? ? subscriber.onNext(i);? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? }).timeout(200, TimeUnit.MILLISECONDS, Observable.just(100,200))? ? ? ? ? ? ? ? ? ? ? ? .observeOn(AndroidSchedulers.mainThread())? ? ? ? ? ? ? ? ? ? ? ? .subscribe(onNextAction);

需要強調(diào)的一點是，在這里timeout(long timeout, TimeUnit timeUnit, Observable other)是timeout其中的一種渗常，它在超時的時候會將源Observable轉(zhuǎn)換為備用的Observable對象進行發(fā)送哮塞。

6、錯誤操作符

圖2.6 錯誤操作符的分類

①catch操作符

實質(zhì)上在這里catch操作符細分有三種實現(xiàn)方案：onErrorReturn凳谦、onErrorResumeNext忆畅、onExceptionResumeNext。

首先分析onErrorReturn的代碼：

Observable.create(newObservable.OnSubscribe() {@Overridepublicvoidcall(Subscriber subscriber){for(inti =0; i <5; i++) {if(i >3) {? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? subscriber.onError(newThrowable("User Alex Defined Error"));? ? ? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? ? ? ? ? subscriber.onNext(i);? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? }).onErrorReturn(newFunc1() {@OverridepublicIntegercall(Throwable throwable){return404;? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? }).subscribe(onNextAction, onErrorAction, onCompletedAction);

onErrorReturn操作符尸执，會在遇到錯誤時家凯，停止源Observable的，并調(diào)用用戶自定義的返回請求如失，實質(zhì)上就是調(diào)用一次OnNext方法進行內(nèi)容發(fā)送后绊诲，停止消息發(fā)送。

然后分析onErrorResumeNext的代碼：

Observable.create(newObservable.OnSubscribe() {@Overridepublicvoidcall(Subscriber subscriber){for(inti =0; i <5; i++) {if(i >3) {? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? subscriber.onError(newThrowable("User Alex Defined Error"));? ? ? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? ? ? ? ? subscriber.onNext(i);? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? }).onErrorResumeNext(newFunc1>() {@OverridepublicObservable call(Throwable throwable) {returnObservable.just(100,101,102);? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? }).subscribe(onNextAction,onErrorAction,onCompletedAction);

onErrorResumeNext操作符褪贵，會在源Observable遇到錯誤時掂之，立即停止源Observable的數(shù)據(jù)發(fā)送，并取用新的Observable對象進行新的數(shù)據(jù)發(fā)送脆丁。

最后世舰，分析onExceptionResumeNext的代碼：

Observable.create(newObservable.OnSubscribe() {@Overridepublicvoidcall(Subscriber subscriber){for(inti =0; i <5; i++) {if(i >3) {? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? subscriber.onError(newThrowable("User Alex Defined Error"));? ? ? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? ? ? ? ? subscriber.onNext(i);? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? }).onExceptionResumeNext(Observable.just(100,101,102)).subscribe(onNextAction,onErrorAction,onCompletedAction);

onExceptionResumeNext，會將錯誤發(fā)給Observer槽卫，而不會調(diào)用備用的Observable

②retry操作符

retry操作符實現(xiàn)的代碼如下所示：

//retry操作符跟压，當遇到exception時會進行重試，重試次數(shù)可以由用戶進行定義Observable.create(newObservable.OnSubscribe() {@Overridepublicvoidcall(Subscriber subscriber){for(inti =0; i <5; i++) {if(i >1) {? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? subscriber.onError(newThrowable("User Alex Defined Error"));? ? ? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? ? ? ? ? subscriber.onNext(i);? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? }).retry(2).subscribe(onNextAction,onErrorAction,onCompletedAction);

retry操作符不會將原始Observable的onError通知傳遞給觀察者歼培，它會重新訂閱這個Observable震蒋。

圖2.6.2 retry操作符示意圖

7茸塞、布爾操作符

布爾操作符根據(jù)給定規(guī)則進行判斷，是否符合規(guī)則然后返回布爾值查剖。布爾操作符意義簡單操作簡便在這里介紹如下5種：

圖2.7 布爾操作符分類

①all操作符

all操作符實現(xiàn)的代碼如下所示：

Observable.just(1,2,3,4).all(newFunc1() {@OverridepublicBooleancall(Integer num){returnnum >3;? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? }).subscribe(onNextAction, onErrorAction, onCompletedAction);

all操作符钾虐，對源Observable發(fā)送的每一個數(shù)據(jù)根據(jù)給定的條件進行判斷。如果全部符合條件笋庄，返回true禾唁，否則返回false。

②contains操作符

contains操作符實現(xiàn)的代碼如下所示：

Observable.just(1, 2, 3, 4).contains(2).subscribe(onNextAction,onErrorAction,onCompletedAction);

contains操作符无切，對源Observable發(fā)送的數(shù)據(jù)是否包含定義的選項進行判斷。如果包含返回true丐枉，否則返回false哆键。

③isEmpty操作符

isEmpty操作符實現(xiàn)的代碼如下所示：

Observable.just(1, 2, 3, 4).isEmpty().subscribe(onNextAction,onErrorAction,onCompletedAction);

isEmpty操作符，對源Observable發(fā)送的數(shù)據(jù)是否為空進行判斷瘦锹。如果源Observable發(fā)送的數(shù)據(jù)為空返回true籍嘹，否則返回false。

④exists操作符

exists操作符實現(xiàn)的代碼如下所示：

Observable.just(1,2,3,4).exists(newFunc1() {@OverridepublicBooleancall(Integer num){returnnum >3;? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? }).subscribe(onNextAction, onErrorAction, onCompletedAction);

exists操作符弯院，對源Observable發(fā)送的單獨一個數(shù)據(jù)根據(jù)給定的條件進行判斷辱士。如果有一個數(shù)據(jù)符合條件，返回true听绳，否則返回false颂碘。

⑤sequenceEqual操作符

sequenceEqual操作符實現(xiàn)的代碼如下所示：

Observable.sequenceEqual(Observable.just(1, 2, 3, 4),Observable.just(1)).subscribe(onNextAction,onErrorAction,onCompletedAction);

sequenceEqual操作符，對兩個Observable進行判斷椅挣，兩個Observable相同時返回true头岔，否則返回false。這里包含兩個Observable的數(shù)據(jù)鼠证，發(fā)射順序峡竣，終止狀態(tài)是否相同。

8量九、條件操作符

圖2.8 條件操作符

①amb操作符

amb操作符實現(xiàn)的代碼如下所示：

//給定多個Observable适掰，只讓第一個發(fā)送數(shù)據(jù)的Observable發(fā)送數(shù)據(jù)Observable? ? ? ? ? ? ? ? ? ? ? ? .amb(Observable.range(0,3).delay(2000, TimeUnit.MILLISECONDS),Observable.range(100,3))? ? ? ? ? ? ? ? ? ? ? ? .subscribe(onNextAction);

如下圖所示，首先發(fā)送通知給Amb的那個荠列，不管發(fā)射的是一項數(shù)據(jù)還是一個onError或onCompleted通知类浪。Amb將忽略和丟棄其它所有Observables的發(fā)射物。

圖2.8.1 amb操作符示意圖

②defaultIfEmpty操作符

amb操作符實現(xiàn)的代碼如下所示：

//如果源Observable沒有發(fā)送數(shù)據(jù)肌似，則發(fā)送一個默認數(shù)據(jù)Observable.create(newObservable.OnSubscribe() {@Overridepublicvoidcall(Subscriber subscriber){? ? ? ? ? ? ? ? ? ? ? ? subscriber.onCompleted();? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? }).defaultIfEmpty(404).subscribe(onNextAction,onErrorAction,onCompletedAction);

9戚宦、轉(zhuǎn)換操作符

轉(zhuǎn)換操作符可以將Observable轉(zhuǎn)換為其它的對象或數(shù)據(jù)結(jié)構(gòu)。在這里介紹如下所示三種轉(zhuǎn)換操作符：

圖2.9 轉(zhuǎn)換操作符的分類

①toList操作符

toList操作符實現(xiàn)的代碼如下所示：

//toList操作符锈嫩，將源Observable發(fā)送的數(shù)據(jù)組合為一個List集合//然后再次在onNext方法中將轉(zhuǎn)換完的List集合進行傳遞Observable.just(1,2,3).toList().subscribe(newAction1>() {@Overridepublicvoidcall(List numList){for(Integer i : numList) {? ? ? ? ? ? ? ? ? ? ? ? ? ? Toast.makeText(getActivity(),"i:"+ i, Toast.LENGTH_SHORT).show();? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? });

通常受楼，發(fā)射多項數(shù)據(jù)的Observable會為每一項數(shù)據(jù)調(diào)用onNext方法垦搬。你可以用toList操作符改變這個行為，讓Observable將多項數(shù)據(jù)組合成一個List艳汽，然后調(diào)用一次onNext方法傳遞整個列表猴贰。

②toSortedList操作符

toSortedList操作符實現(xiàn)的代碼如下所示：

//toSortedList操作符,會將源Observable發(fā)送的數(shù)據(jù)組合為一個List集合,并會按照升序的方式進行排序//然后再次在onNext方法中將轉(zhuǎn)換完的List集合進行傳遞Observable.just(40,10,80,30).toSortedList().subscribe(newAction1>() {@Overridepublicvoidcall(List numList){for(Integer i : numList) {? ? ? ? ? ? ? ? ? ? ? ? ? ? Toast.makeText(getActivity(),"i:"+ i, Toast.LENGTH_SHORT).show();? ? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? });

不同于toList操作符的是，它會對產(chǎn)生的列表排序河狐，默認是自然升序米绕。

③toMap操作符

toMap操作符實現(xiàn)的代碼如下所示：

//toMap操作符，將源Observable發(fā)送的數(shù)據(jù)作為Map集合中的值馋艺，需要值進行鍵的定義//將轉(zhuǎn)換完畢的Map集合在onNext方法中進行發(fā)送Observable.just("Alex","Payne").toMap(newFunc1() {@OverridepublicIntegercall(String s){returns.equals("Alex")?0:1;? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? }).subscribe(newAction1>() {@Overridepublicvoidcall(Map convertMap){for(inti =0; i < convertMap.size(); i++) {? ? ? ? ? ? ? ? ? ? ? ? ? Toast.makeText(getActivity(), convertMap.get(i), Toast.LENGTH_SHORT).show();? ? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? });

源Observable發(fā)送的數(shù)據(jù)作為鍵值對中的值艺普，我們可以提供一個用于生成Map的Key的函數(shù)，然后不同的鍵存儲源Observable發(fā)送的不同的值治力。

toMap默認不在任何特定的調(diào)度器上執(zhí)行侵佃。

圖2.9.3 toMap操作符示意圖

三、總結(jié)

1踱蛀、在本文中窿给，我結(jié)合項目代碼詳細介紹了部分RxJava的操作符，局部參照：RxJava中文翻譯文檔率拒。

2崩泡、本文中的案例代碼已上傳Github，歡迎大家star猬膨、fork角撞。詳情戳→GitHub案例代碼。

3勃痴、操作符實質(zhì)上就是RxJava函數(shù)式編程模式的體現(xiàn)靴寂，Lambda表達式并且可以進一步優(yōu)化RxJava。

4召耘、在下篇文章中我會對于RxJava進行深層次的剖析百炬，還有RxJava結(jié)合例如Retrofit、RxBus等開源框架的內(nèi)容污它，希望本文對你在學習RxJava的路上有所啟發(fā)剖踊。

小禮物走一走，來簡書關(guān)注我

作者：Alex_Payne

鏈接：http://www.reibang.com/p/d997805b37d4

來源：簡書

簡書著作權(quán)歸作者所有衫贬，任何形式的轉(zhuǎn)載都請聯(lián)系作者獲得授權(quán)并注明出處德澈。