注：本文摘自《RxJava Essentials》翻译中文版电子书
Merge
我们往往回想得到一个结果：多输入，单输出。merge()方法将帮助你把两个甚至更多的Observables合并到他们发射的数据项里。注意如果你同步的合并Observable，它们将连接在一起并且不会交叉。例子：
private void loadList(List<AppInfo> apps) {
List reversedApps = Lists.reverse(apps);
Observable<AppInfo> observableApps =Observable.from(apps);
Observable<AppInfo> observableReversedApps =Observable.from(reversedApps);
//合并
Observable<AppInfo> mergedObserbable = Observable.merge(observableApps,observableReversedApps);
mergedObserbable.subscribe(...);
}
如果发生错误，你可以认为每个Observable抛出的错误都将会打断合并。
如果你需要避免这种情况，RxJava提供了mergeDelayError()，它能从一个Observable中继续发射数据即便是其中有一个抛出了错误。当所有的Observables都完成时，mergeDelayError()将会发射onError()
ZIP
在一种场景中，如果要处理多个数据来源时会带从来多个Observables接收数据，然后处理后将它们合并成一个新的可观测序列来使用。
zip()方法可以完成。
zip()合并两个或者多个Observables发射出的数据项，根据指定的函数Func*变换它们，并发射一个新值。
例子：
private void loadList(List<AppInfo> apps) {
//得到app列表
Observable<AppInfo> observableApp = Observable.from(apps);
//每秒生成一个Long类型的数据
Observable<Long> tictoc = Observable.interval(1, TimeUnit.SECONDS);
//使用zip合并
Observable.zip(observableApp, tictoc,
(AppInfo appInfo, Long time) -> updateTitle(appInfo, time))
.observeOn(AndroidSchedulers.mainThread())
.subscribe(...);
}
zip()函数有三个参数：两个Observables和一个Func2。
Join
zip()和merge()方法作用在发射数据的范畴内，在决定如何操作值之前有些场景我们需要考虑时间的。RxJava的join()函数基于时间窗口将两个Observables发射的数据结合在一起。
（PS：Join有点难理解，请仔细阅读下面例子代码体会）
解释下join()需要的4个参数：
- 第二个Observable和源Observable结合。
- Func1参数：在指定的由时间窗口定义时间间隔内，源Observable发射的数据和从第二个Observable发射的数据相互配合返回的Observable。
- Func1参数：在指定的由时间窗口定义时间间隔内，第二个Observable发射的数据和从源Observable发射的数据相互配合返回的Observable。
- Func2参数：定义已发射的数据如何与新发射的数据项相结合。
例子：
private void loadList(List<AppInfo> apps) {
Observable<AppInfo> appsSequence = Observable.interval(1000, TimeUnit.MILLISECONDS)
.map(position -> {
return apps.get(position.intValue());
});
Observable<Long> tictoc = Observable.interval(1000,TimeUnit.MILLISECONDS);
appsSequence.join(
tictoc,
appInfo -> Observable.timer(2,TimeUnit.SECONDS),
time -> Observable.timer(0, TimeUnit.SECONDS),
this::updateTitle)
.observeOn(AndroidSchedulers.mainThread())
.take(10)
.subscribe(...);
}
我们有一个新的对象appsSequence，它是一个每秒从我们已安装的app列表发射app数据的可观测序列。tictoc这个Observable数据每秒只发射一个新的Long型整数。为了合并它们，我们需要指定两个Func1变量：
appInfo -> Observable.timer(2, TimeUnit.SECONDS)
time -> Observable.timer(0, TimeUnit.SECONDS)
上面描述了两个时间窗口。下面一行描述我们如何使用Func2将两个发射的数据结合在一起。
this::updateTitle
combineLatest
combineLatest()函数有点像zip()函数的特殊形式。正如我们已经学习的，zip()作用于最近未打包的两个Observables。相反，combineLatest()作用于最近发射的数据项：如果Observable1发射了A并且Observable2发射了B和C，combineLatest()将会分组处理AB和AC。
combineLatest()函数接受二到九个Observable作为参数，如果有需要的话或者单个Observables列表作为参数。
例子：
private void loadList(List<AppInfo> apps) {
Observable<AppInfo> appsSequence = Observable.interval(1000, TimeUnit.MILLISECONDS)
.map(position ->apps.get(position.intValue()));
Observable<Long> tictoc = Observable.interval(1500, TimeUnit.MILLISECONDS);
Observable.combineLatest(appsSequence, tictoc,
this::updateTitle)
.observeOn(AndroidSchedulers.mainThread())
.subscribe(...);
}
这我们使用了两个Observables：一个是每秒钟从我们已安装的应用列表发射一个App数据，第二个是每隔1.5秒发射一个Long型整数。我们将他们结合起来并执行updateTitle()函数。
由于不同的时间间隔，AppInfo对象如我们所预料的那样有时候会重复。
And,Then和When
在将来还有一些zip()满足不了的场景。如复杂的架构，或者是仅仅为了个人爱好，你可以使用And/Then/When解决方案。它们在RxJava的joins包下，使用Pattern和Plan作为中介，将发射的数据集合并到一起。例子：
private void loadList(List<AppInfo> apps) {
//发射我们安装的应用列表数据
Observable<AppInfo> observableApp = Observable.from(apps);
//每秒发射一个Long型整数
Observable<Long> tictoc = Observable.interval(1, TimeUnit.SECONDS);
//用and()连接源Observable和第二个Observable。
Pattern2<AppInfo, Long> pattern = JoinObservable.from(observableApp).and(tictoc);
Plan0<AppInfo> plan = pattern.then(this::updateTitle);
JoinObservable
.when(plan)
.toObservable()
.observeOn(AndroidSchedulers.mainThread())
.subscribe(...);
}
上述代码中，有两个发射的序列，observableApp，发射我们安装的应用列表数据，tictoc每秒发射一个Long型整数。
然后用and()连接源Observable和第二个Observable。
JoinObservable.from(observableApp).and(tictoc);
然后创建一个pattern对象，使用这个对象我们可以创建一个Plan对象。
现在我们有了一个Plan对象并且当plan发生时我们可以决定接下来发生的事情。
.when(plan).toObservable()
然后就
pattern.then(this::updateTitle);
这时候，我们可以订阅新的Observable。
（这段想一想就会理解是什么意思了）
Switch
有这样一个复杂的场景就是在一个subscribe-unsubscribe的序列里我们能够从一个Observable自动取消订阅来订阅一个新的Observable。
RxJava的switch()，正如定义的，将一个发射多个Observables的Observable转换成另一个单独的Observable，后者发射那些Observables最近发射的数据项。
给出一个发射多个Observables序列的源Observable，switch()订阅到源Observable然后开始发射由第一个发射的Observable发射的一样的数据。当源Observable发射一个新的Observable时，switch()立即取消订阅前一个发射数据的Observable（因此打断了从它那里发射的数据流）然后订阅一个新的Observable，并开始发射它的数据。
（这个好难懂）
StartWith
startWith()是concat()的对应部分。
正如concat()向发射数据的Observable追加数据那样，在Observable开始发射他们的数据之前， startWith()通过传递一个参数来先发射一个数据序列。
