摘要：组合操作符主要包含等等。将两个发射的事件序列组合并成一个事件序列，就像是一个发射的一样。用于在源发射的数据前插入数据。和前面的例子一样，我们将操作符换成了程序输出用于将两个最近发射的数据已经函数的规则进展组合。
转载请注明出处：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函数的规则进展组合。下面是官方提供的原理图：
