資訊專欄INFORMATION COLUMN
摘要：組合操作符主要包含等等。將兩個發射的事件序列組合并成一個事件序列，就像是一個發射的一樣。用于在源發射的數據前插入數據。和前面的例子一樣，我們將操作符換成了程序輸出用于將兩個最近發射的數據已經函數的規則進展組合。
轉載請注明出處：https://zhuanlan.zhihu.com/p/22039934
RxJava系列1(簡介)
RxJava系列2(基本概念及使用介紹)
RxJava系列3(轉換操作符)
RxJava系列4(過濾操作符)
RxJava系列5(組合操作符)
RxJava系列6(從微觀角度解讀RxJava源碼)
RxJava系列7(最佳實踐)
這一章我們接著介紹組合操作符，這類operators可以同時處理多個Observable來創建我們所需要的Observable。組合操作符主要包含： Merge StartWith Concat Zip CombineLatest SwitchOnNext Join等等。
Mergemerge(Observable, Observable)將兩個Observable發射的事件序列組合并成一個事件序列，就像是一個Observable發射的一樣。你可以簡單的將它理解為兩個Obsrvable合并成了一個Observable，合并后的數據是無序的。
我們看下面的例子，一共有兩個Observable：一個用來發送字母，另一個用來發送數字；現在我們需要兩連個Observable發射的數據合并。
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發射的事件序列組合并成一個事件序列，就像是一個Observable發射的一樣。
StartWithstartWith(T)用于在源Observable發射的數據前插入數據。使用startWith(Iterable
startWith(Observable
源Observable發射數據的前面）。官方示意圖：
concat(Observable, Observable) concat(Observable<？ extends Observable
merge:合并后發射的數據是無序的；
startWitch:只能在源Observable發射的數據前插入數據。
這里我們將前面Merge操作符的例子拿過來，并將操作符換成Concat，然后我們看看執行結果：
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發射的數據項，根據Func2函數生成一個新的值并發射出去。當其中一個Observable發送數據結束或者出現異常后，另一個Observable也將停在發射數據。
和前面的例子一樣，我們將操作符換成了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最近發射的數據已經Func2函數的規則進展組合。下面是官方提供的原理圖：
