下面展示了可用于多個(gè)Observables的各種操作符
- startWith()---在數(shù)據(jù)序列的開(kāi)頭增加一項(xiàng)數(shù)據(jù)
- merge()---將多個(gè)Observable合并為一個(gè)
- mergeDelayError()---合并多個(gè)Observables,讓沒(méi)有錯(cuò)誤的Observable都完成后再發(fā)射錯(cuò)誤通知
- zip()---使用一個(gè)函數(shù)組合多個(gè)Observable發(fā)射的數(shù)據(jù)集合,然后在發(fā)射這個(gè)結(jié)果
- and(),then(),and when()---(rxjava-joins)通過(guò)模式和計(jì)劃組合多個(gè)Observables發(fā)射的數(shù)據(jù)集合
- combineLatest()---當(dāng)兩個(gè)Observables中的任何一個(gè)發(fā)射了一個(gè)數(shù)據(jù)時(shí),通過(guò)一個(gè)指定的函數(shù)組合每個(gè)Observable發(fā)射的最新數(shù)據(jù)(一共兩個(gè)數(shù)據(jù)),然后發(fā)射這個(gè)函數(shù)的結(jié)果。
- join() and groupJoin()---無(wú)論何時(shí),如果一個(gè)Observable發(fā)射了一個(gè)數(shù)據(jù)項(xiàng),只要在另一個(gè)Observable發(fā)射的數(shù)據(jù)項(xiàng)定義的時(shí)間窗口內(nèi),就將兩個(gè)Observable發(fā)射的數(shù)據(jù)合并發(fā)射。
- switchOnNext()---將一個(gè)發(fā)射Observables的Observable轉(zhuǎn)換成另一個(gè)Observable,后者發(fā)射這些Observables最近發(fā)射的數(shù)據(jù)磺平。
StartWith
在數(shù)據(jù)序列的開(kāi)頭插入一條指定的項(xiàng)
如果你想要一個(gè)Observable在發(fā)射數(shù)據(jù)之前先發(fā)射一個(gè)指定的數(shù)據(jù)序列，可以使用StartWith操作符部脚。（如果你想一個(gè)Observable發(fā)射的數(shù)據(jù)末尾追加一個(gè)數(shù)據(jù)序列可以使用Concat操作符姻采。）
你也可以傳遞一個(gè)Observable給startWith做祝，它會(huì)將那個(gè)Observable的發(fā)射物插在原始Observable發(fā)射的數(shù)據(jù)序列之前近尚，然后把這個(gè)當(dāng)做自己的發(fā)射物集合蠕啄。這可以看作是Concat的反轉(zhuǎn)。
/**
* startWith操作符 解釋：在源Observable輸出之前插入指定數(shù)據(jù)項(xiàng)
*/
private static void test() {
Observable.just(7, 8, 9).startWith(11, 12)
.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
System.out.println("onCompleted ");
}
@Override
public void onError(Throwable e) {
System.out.println("onError " + e.toString());
}
@Override
public void onNext(Integer t) {
System.out.println("onNext t = " + t);
}
});
}
Merge操作符
合并多個(gè)Observables的發(fā)射物
使用Merge操作符你可以將多個(gè)Observables的輸出合并戈锻，就好像它們是一個(gè)單個(gè)的Observable一樣歼跟。
Merge可能會(huì)讓合并的Observables發(fā)射的數(shù)據(jù)交錯(cuò)（有一個(gè)類似的操作符Concat不會(huì)讓數(shù)據(jù)交錯(cuò)，它會(huì)按順序一個(gè)接著一個(gè)發(fā)射多個(gè)Observables的發(fā)射物）格遭。
正如圖例上展示的哈街，任何一個(gè)原始Observable的onError通知會(huì)被立即傳遞給觀察者，而且會(huì)終止合并后的Observable拒迅。
在很多ReactiveX實(shí)現(xiàn)中還有一個(gè)叫MergeDelayError的操作符骚秦，它的行為有一點(diǎn)不同，它會(huì)保留onError通知直到合并后的Observable所有的數(shù)據(jù)發(fā)射完成坪它，在那時(shí)它才會(huì)把onError傳遞給觀察者骤竹。
RxJava將它實(shí)現(xiàn)為merge, mergeWith和mergeDelayError。
/**
* Merge 解釋：將2-9個(gè)Observables合并到一個(gè)Observable中進(jìn)行發(fā)射往毡，合并后的數(shù)據(jù)可能會(huì)是交錯(cuò)（無(wú)序）的（如果想要沒(méi)有交錯(cuò)，
* 可以使用concat操作符） merge還可以傳遞一個(gè)Observable列表List靶溜，數(shù)組
* 甚至是一個(gè)發(fā)射Observable序列的Observable开瞭，merge將合并它們的輸出作為單個(gè)Observable的輸出
*/
private static void test1() {
Observable<String> letterObservable = Observable.just("A", "B", "C", "D", "E", "F", "G", "H");
Observable<Integer> numberObservable = Observable.just(1, 2, 3, 4, 5);
Observable.merge(letterObservable, numberObservable)
.subscribe(new Action1<Serializable>() {
@Override
public void call(Serializable value) {
System.out.println("value = " + value);
}
});
}
Zip操作符
通過(guò)一個(gè)函數(shù)將多個(gè)Observables的發(fā)射物結(jié)合到一起懒震，基于這個(gè)函數(shù)的結(jié)果為每個(gè)結(jié)合體發(fā)射單個(gè)數(shù)據(jù)項(xiàng)。
Zip操作符返回一個(gè)Obversable嗤详，它使用這個(gè)函數(shù)按順序結(jié)合兩個(gè)或多個(gè)Observables發(fā)射的數(shù)據(jù)項(xiàng)个扰，然后它發(fā)射這個(gè)函數(shù)返回的結(jié)果。它按照嚴(yán)格的順序應(yīng)用這個(gè)函數(shù)葱色。它只發(fā)射與發(fā)射數(shù)據(jù)項(xiàng)最少的那個(gè)Observable一樣多的數(shù)據(jù)递宅。
RxJava將這個(gè)操作符實(shí)現(xiàn)為zip和zipWith。
zip的最后一個(gè)參數(shù)接受每個(gè)Observable發(fā)射的一項(xiàng)數(shù)據(jù)苍狰，返回被壓縮后的數(shù)據(jù)办龄，它可以接受一到九個(gè)參數(shù)：一個(gè)Observable序列，或者一些發(fā)射Observable的Observables淋昭。
zipWith
zipWith操作符總是接受兩個(gè)參數(shù)俐填，第一個(gè)參數(shù)是一個(gè)Observable或者一個(gè)Iterable。
zip和zipWith默認(rèn)不在任何特定的操作符上執(zhí)行翔忽。
/**
* Zip
* 結(jié)合兩個(gè)或多個(gè)Observables發(fā)射的數(shù)據(jù)項(xiàng)英融，每個(gè)數(shù)據(jù)只能組合一次，而且都是有序的歇式。
* 它只發(fā)射與發(fā)射數(shù)據(jù)項(xiàng)最少的那個(gè)Observable一樣多的數(shù)據(jù)驶悟。
*
* 應(yīng)用場(chǎng)景參考：http://blog.csdn.net/jdsjlzx/article/details/51724087
*/
private static void test2() {
Observable<String> letterObservable = Observable.just("A", "B", "C", "D", "E", "F", "G", "H");
Observable<Integer> numberObservable = Observable.just(1, 2, 3, 4, 5);
Observable.zip(letterObservable, numberObservable, new Func2<String, Integer, String>() {
@Override
public String call(String t1, Integer t2) {
return t1 + t2;
}
}).subscribe(new Action1<String>() {
@Override
public void call(String value) {
System.out.println("value = " + value);
}
});
}
CombineLatest操作符
當(dāng)兩個(gè)Observables中的任何一個(gè)發(fā)射了數(shù)據(jù)時(shí)，使用一個(gè)函數(shù)結(jié)合每個(gè)Observable發(fā)射的最近數(shù)據(jù)項(xiàng)材失，并且基于這個(gè)函數(shù)的結(jié)果發(fā)射數(shù)據(jù)痕鳍。
CombineLatest操作符行為類似于zip，但是只有當(dāng)原始的Observable中的每一個(gè)都發(fā)射了一條數(shù)據(jù)時(shí)zip才發(fā)射數(shù)據(jù)豺憔。CombineLatest則在原始的Observable中任意一個(gè)發(fā)射了數(shù)據(jù)時(shí)發(fā)射一條數(shù)據(jù)额获。當(dāng)原始Observables的任何一個(gè)發(fā)射了一條數(shù)據(jù)時(shí)，CombineLatest使用一個(gè)函數(shù)結(jié)合它們最近發(fā)射的數(shù)據(jù)恭应，然后發(fā)射這個(gè)函數(shù)的返回值抄邀。
RxJava將這個(gè)操作符實(shí)現(xiàn)為combineLatest，它接受二到九個(gè)Observable作為參數(shù)昼榛，或者單個(gè)Observables列表作為參數(shù)境肾。它默認(rèn)不在任何特定的調(diào)度器上執(zhí)行。
/**
* CombineLatest
* 解釋：combineLatest操作符把兩個(gè)Observable產(chǎn)生的結(jié)果進(jìn)行合并胆屿，合并的結(jié)果組成一個(gè)新的Observable奥喻。
* 這兩個(gè)Observable中任意一個(gè)Observable產(chǎn)生的結(jié)果，都和另一個(gè)Observable最后產(chǎn)生的結(jié)果非迹，按照一定的規(guī)則進(jìn)行合并环鲤。
*/
private static void test3() {
Observable<String> letterObservable = Observable.just("A", "B", "C");
Observable<Integer> numberObservable = Observable.just(4, 5);
//letterObservable numberObservable誰(shuí)在前誰(shuí)在后都會(huì)對(duì)執(zhí)行結(jié)果會(huì)有影響
Observable.combineLatest( numberObservable, letterObservable,new Func2<Integer, String, String>() {
@Override
public String call(Integer t1, String t2) {
System.out.println("combine t1 = " + t1 + " | t2 = " + t2);// t1的值一開(kāi)始就是letterObservable中的最后一個(gè)值
return t1 + t2;
}
}).subscribe(new Action1<String>() {
@Override
public void call(String value) {
System.out.println("value = " + value);
}
});
}
withLatestFrom操作符
withLatestFrom操作符還在開(kāi)發(fā)中，不是1.0版本的一部分憎兽。類似于combineLatest冷离，但是只在單個(gè)原始Observable發(fā)射了一條數(shù)據(jù)時(shí)才發(fā)射數(shù)據(jù)吵冒。
Join操作符
任何時(shí)候，只要在另一個(gè)Observable發(fā)射的數(shù)據(jù)定義的時(shí)間窗口內(nèi)西剥，這個(gè)Observable發(fā)射了一條數(shù)據(jù)痹栖，就結(jié)合兩個(gè)Observable發(fā)射的數(shù)據(jù)。
Join操作符結(jié)合兩個(gè)Observable發(fā)射的數(shù)據(jù)瞭空，基于時(shí)間窗口（你定義的針對(duì)每條數(shù)據(jù)特定的原則）選擇待集合的數(shù)據(jù)項(xiàng)揪阿。你將這些時(shí)間窗口實(shí)現(xiàn)為一些Observables，它們的生命周期從任何一條Observable發(fā)射的每一條數(shù)據(jù)開(kāi)始咆畏。當(dāng)這個(gè)定義時(shí)間窗口的Observable發(fā)射了一條數(shù)據(jù)或者完成時(shí)南捂，與這條數(shù)據(jù)關(guān)聯(lián)的窗口也會(huì)關(guān)閉。只要這條數(shù)據(jù)的窗口是打開(kāi)的鳖眼，它將繼續(xù)結(jié)合其它Observable發(fā)射的任何數(shù)據(jù)項(xiàng)黑毅。你定義一個(gè)用于結(jié)合數(shù)據(jù)的函數(shù)。
/**
* Join 類似于combineLatest操作符钦讳，但是join操作符可以控制每個(gè)Observable產(chǎn)生結(jié)果的生命周期矿瘦，在每個(gè)結(jié)果的生命周期內(nèi)，
* 可以與另一個(gè)Observable產(chǎn) 生的結(jié)果按照一定的規(guī)則進(jìn)行合并
*
* Join(Observable,Func1,Func1,Func2) 需要傳遞四個(gè)參數(shù)
*
* join操作符的用法如下： observableA.join(observableB, observableA產(chǎn)生結(jié)果生命周期控制函數(shù)愿卒，
* observableB產(chǎn)生結(jié)果生命周期控制函數(shù)缚去， observableA產(chǎn)生的結(jié)果與observableB產(chǎn)生的結(jié)果的合并規(guī)則）
*
* 一句話概括：在observableA的生命周期內(nèi)：observableB輸出的數(shù)據(jù)項(xiàng)與observableA輸出的數(shù)據(jù)項(xiàng)每個(gè)合并
*
* test4() 沒(méi)有任何合并結(jié)果輸出 分析:同一線程:observableA的生命周期已經(jīng)執(zhí)行完了,observableB還沒(méi)出來(lái),所以合并不了
*
*/
private static void test4() {
Observable<Integer> observableA = Observable.range(1, 5);
List<Integer> data = Arrays.asList(6, 7, 8, 9, 10);
Observable<Integer> observableB = Observable.from(data);
observableA.join(observableB, new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer value) {
//return Observable.just(value);
return Observable.just(value).delay(1, TimeUnit.SECONDS);
}
}, new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer value) {
return Observable.just(value);
}
}, new Func2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer value1, Integer value2) {
System.out.println("left: " + value1 + " right:" + value2);
return value1 + value2;
}
}).subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
System.out.println("onCompleted");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer value) {
System.out.println("onNext value = " + value);
}
});
}
