資訊專欄INFORMATION COLUMN
摘要：組合操作符主要包含等等。將兩個發(fā)射的事件序列組合并成一個事件序列，就像是一個發(fā)射的一樣。用于在源發(fā)射的數(shù)據(jù)前插入數(shù)據(jù)。和前面的例子一樣，我們將操作符換成了程序輸出用于將兩個最近發(fā)射的數(shù)據(jù)已經(jīng)函數(shù)的規(guī)則進(jìn)展組合。
轉(zhuǎn)載請注明出處：https://zhuanlan.zhihu.com/p/22039934
RxJava系列1(簡介)
RxJava系列2(基本概念及使用介紹)
RxJava系列3(轉(zhuǎn)換操作符)
RxJava系列4(過濾操作符)
RxJava系列5(組合操作符)
RxJava系列6(從微觀角度解讀RxJava源碼)
RxJava系列7(最佳實(shí)踐)
這一章我們接著介紹組合操作符，這類operators可以同時處理多個Observable來創(chuàng)建我們所需要的Observable。組合操作符主要包含： Merge StartWith Concat Zip CombineLatest SwitchOnNext Join等等。
Mergemerge(Observable, Observable)將兩個Observable發(fā)射的事件序列組合并成一個事件序列，就像是一個Observable發(fā)射的一樣。你可以簡單的將它理解為兩個Obsrvable合并成了一個Observable，合并后的數(shù)據(jù)是無序的。
我們看下面的例子，一共有兩個Observable：一個用來發(fā)送字母，另一個用來發(fā)送數(shù)字；現(xiàn)在我們需要兩連個Observable發(fā)射的數(shù)據(jù)合并。
String[] letters = new String[]{"A", "B", "C", "D", "E", "F", "G", "H"};
Observable letterSequence = Observable.interval(300, TimeUnit.MILLISECONDS)
.map(new Func1() {
@Override
public String call(Long position) {
return letters[position.intValue()];
}
}).take(letters.length);
Observable numberSequence = Observable.interval(500, TimeUnit.MILLISECONDS).take(5);
Observable.merge(letterSequence, numberSequence)
.subscribe(new Observer() {
@Override
public void onCompleted() {
System.exit(0);
}
@Override
public void onError(Throwable e) {
System.out.println("Error:" + e.getMessage());
}
@Override
public void onNext(Serializable serializable) {
System.out.print(serializable.toString()+" ");
}
});
程序輸出：
A 0 B C 1 D E 2 F 3 G H 4
merge(Observable[])將多個Observable發(fā)射的事件序列組合并成一個事件序列，就像是一個Observable發(fā)射的一樣。
StartWithstartWith(T)用于在源Observable發(fā)射的數(shù)據(jù)前插入數(shù)據(jù)。使用startWith(Iterable
startWith(Observable
源Observable發(fā)射數(shù)據(jù)的前面）。官方示意圖：
concat(Observable, Observable) concat(Observable<？ extends Observable
merge:合并后發(fā)射的數(shù)據(jù)是無序的；
startWitch:只能在源Observable發(fā)射的數(shù)據(jù)前插入數(shù)據(jù)。
這里我們將前面Merge操作符的例子拿過來，并將操作符換成Concat，然后我們看看執(zhí)行結(jié)果：
String[] letters = new String[]{"A", "B", "C", "D", "E", "F", "G", "H"};
Observable letterSequence = Observable.interval(300, TimeUnit.MILLISECONDS)
.map(new Func1() {
@Override
public String call(Long position) {
return letters[position.intValue()];
}
}).take(letters.length);
Observable numberSequence = Observable.interval(500, TimeUnit.MILLISECONDS).take(5);
Observable.concat(letterSequence, numberSequence)
.subscribe(new Observer() {
@Override
public void onCompleted() {
System.exit(0);
}
@Override
public void onError(Throwable e) {
System.out.println("Error:" + e.getMessage());
}
@Override
public void onNext(Serializable serializable) {
System.out.print(serializable.toString() + " ");
}
});
程序輸出：
A B C D E F G H 0 1 2 3 4Zip
zip(Observable, Observable, Func2)用來合并兩個Observable發(fā)射的數(shù)據(jù)項(xiàng)，根據(jù)Func2函數(shù)生成一個新的值并發(fā)射出去。當(dāng)其中一個Observable發(fā)送數(shù)據(jù)結(jié)束或者出現(xiàn)異常后，另一個Observable也將停在發(fā)射數(shù)據(jù)。
和前面的例子一樣，我們將操作符換成了zip:
String[] letters = new String[]{"A", "B", "C", "D", "E", "F", "G", "H"};
Observable letterSequence = Observable.interval(120, TimeUnit.MILLISECONDS)
.map(new Func1() {
@Override
public String call(Long position) {
return letters[position.intValue()];
}
}).take(letters.length);
Observable numberSequence = Observable.interval(200, TimeUnit.MILLISECONDS).take(5);
Observable.zip(letterSequence, numberSequence, new Func2() {
@Override
public String call(String letter, Long number) {
return letter + number;
}
}).subscribe(new Observer() {
@Override
public void onCompleted() {
System.exit(0);
}
@Override
public void onError(Throwable e) {
System.out.println("Error:" + e.getMessage());
}
@Override
public void onNext(String result) {
System.out.print(result + " ");
}
});
程序輸出：
A0 B1 C2 D3 E4CombineLatest
comnineLatest(Observable, Observable, Func2)用于將兩個Observale最近發(fā)射的數(shù)據(jù)已經(jīng)Func2函數(shù)的規(guī)則進(jìn)展組合。下面是官方提供的原理圖：
