这个页面展示的操作符可用于组合多个 Observables。
And/Then/When — 通过模式和计划组合多个 Observables 发射的数据集合。
CombineLatest — 当两个 Observables 中的任何一个发射了一个数据时，通过一个指定的函数组合每个 Observable 发射的最新数据（一共两个数据），然后发射这个函数的结果。
Join — 无论何时，如果一个 Observable 发射了一个数据项，只要在另一个 Observable 发射的数据项定义的时间窗口内，就将两个 Observable 发射的数据合并发射。
Merge — 将多个 Observable 合并为一个。
StartWith — 在数据序列的开头插入一条指定的项。
Switch — 将一个发射 Observables 的 Observable 转换成另一个 Observable，后者发射这些 Observables 最近发射的数据。
Zip — 使用一个函数组合多个 Observable 发射的数据集合，然后再发射这个结果。
4.1 And/Then/When
通过模式和计划组合多个 Observables 发射的数据集合。
示例代码：
无
4.2 CombineLatest
当两个 Observables 中的任何一个发射了一个数据时，通过一个指定的函数组合每个 Observable 发射的最新数据（一共两个数据），然后发射这个函数的结果。
CombineLatest 操作符行为类似于 zip，但是只有当原始的 Observable 中的每一个都发射了一条数据时 zip 才发射数据。CombineLatest 则在原始的 Observable 中任意一个发射了数据时发射一条数据。当原始 Observables 的任何一个发射了一条数据时，CombineLatest 使用一个函数结合它们最近发射的数据，然后发射这个函数的返回值。
实践：可用于填写表单提交，需要满足多个条件才触发“提交”按钮。
示例代码：
//2个有差值的observable
Observable<Long> ob1 = Observable.interval(100, TimeUnit.MILLISECONDS);
Observable<Long> ob2 = Observable.interval(150, TimeUnit.MILLISECONDS);
Observable.combineLatest(ob1, ob2, new BiFunction<Long, Long, String>() {
@Override
public String apply(@NonNull Long aLong, @NonNull Long aLong2) throws Exception {
return aLong + " - " + aLong2;
}
}).take(6).subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
Log.e(TAG, "accept : " + s);
}
});
输出结果：
accept : 0 - 0
accept : 1 - 0
accept : 1 - 1
accept : 2 - 1
accept : 3 - 1
accept : 3 - 2
4.3 Join
无论何时，如果一个 Observable 发射了一个数据项，只要在另一个 Observable 发射的数据项定义的时间窗口内，就将两个 Observable 发射的数据合并发射。
Join 操作符结合两个 Observable 发射的数据，基于时间窗口（你定义的针对每条数据特定的原则）选择待集合的数据项。你将这些时间窗口实现为一些 Observables，它们的生命周期从任何一条 Observable 发射的每一条数据开始。当这个定义时间窗口的 Observable 发射了一条数据或者完成时，与这条数据关联的窗口也会关闭。只要这条数据的窗口是打开的，它将继续结合其它 Observable 发射的任何数据项。你定义一个用于结合数据的函数。
示例代码：
Observable<Integer> ob1 = Observable.just(1, 2);
Observable<String> ob2 = Observable.just("A", "B", "C");
ob1.join(ob2, new Function<Integer, ObservableSource<String>>() {//ob1产生结果声明周期控制函数
@Override
public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
// return Observable.just(integer + "00");
return Observable.just(integer + "00").delay(100, TimeUnit.MILLISECONDS);
// return Observable.just(integer + "00").delay(200, TimeUnit.MILLISECONDS);
// return Observable.just(integer + "00").delay(300, TimeUnit.MILLISECONDS);
}
}, new Function<String, ObservableSource<String>>() {//ob2产生结果声明周期控制函数
@Override
public ObservableSource<String> apply(@NonNull String s) throws Exception {
return Observable.just("Function1:" + s).delay(200, TimeUnit.MILLISECONDS);
}
}, new BiFunction<Integer, String, Object>() {//ob1 和ob2产生结果的合并规则
@Override
public Object apply(@NonNull Integer integer, @NonNull String s) throws Exception {
// Thread.sleep(1000);
return "string = " + s + ", integer = " + integer;
}
}
).subscribe(new Consumer<Object>() {
@Override
public void accept(@NonNull Object o) throws Exception {
Log.e(TAG, "accept : " + o.toString());
}
});
输出结果：
accept : string = A, integer = 1
accept : string = A, integer = 2
accept : string = B, integer = 1
accept : string = B, integer = 2
accept : string = C, integer = 1
accept : string = C, integer = 2
Join 的效果类似于排列组合，把第一个数据源 ob1 作为基座窗口，它根据自己的节奏不断发射数据元素。第二个数据源 ob2 ，每发射一个数据，我们都把它和第一个数据源 ob1 中已经发射的数据进行一对一匹配；
如果某一时刻 ob2 发射了一个数据 B ,此时 ob1 已经发射了 1，2 共两个数据，那么我们的合并操作就会把 B 依次与 1，2 配对，得到两组数据： (1，B)、(2，B)
4.3.1 GroupJoin
GroupJoin 运算符与 Join 类似。将第两个 Observable 的发射放在了 GroupJoin 第三个参数中。具体看代码吧。
示例代码：
Observable<Integer> ob1 = Observable.just(1, 2);
Observable<String> ob2 = Observable.just("A", "B", "C");
ob1.groupJoin(ob2, new Function<Integer, ObservableSource<String>>() {//ob1产生结果声明周期控制函数
@Override
public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
return Observable.just(integer + "00").delay(300, TimeUnit.MILLISECONDS);
}
}, new Function<String, ObservableSource<String>>() {//ob2产生结果声明周期控制函数
@Override
public ObservableSource<String> apply(@NonNull String s) throws Exception {
return Observable.just("Function1:" + s).delay(200, TimeUnit.MILLISECONDS);
}
//与 join 运算符的差别主要是在这里。
}, new BiFunction<Integer, Observable<String>, Observable<String>>() {
@Override
public Observable<String> apply(@NonNull final Integer i, @NonNull Observable<String> sob) throws Exception {
return sob.map(new Function<String, String>() {
@Override
public String apply(@NonNull String s) throws Exception {
return "string = " + s + ", integer = " + i;
}
});
}
}).subscribe(new Consumer<Observable<String>>() {
@Override
public void accept(@NonNull Observable<String> sob) throws Exception {
sob.subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
Log.e(TAG, "accept : " + s);
}
});
}
});
输出结果：
accept : string = A, integer = 2
accept : string = B, integer = 1
accept : string = B, integer = 2
accept : string = C, integer = 1
accept : string = C, integer = 2
4.4 Merge
将多个 Observable 合并为一个 Observable。
Merge 可能会让合并的 Observables 发射的数据交错（有一个类似的操作符 Concat 不会让数据交错，它会按顺序一个接着一个发射多个 Observable）。
示例代码：
Observable<String> ob1 = Observable.interval(100, TimeUnit.MILLISECONDS)
.map(new Function<Long, String>() {
@Override
public String apply(@NonNull Long aLong) throws Exception {
//强迫症与上图一致
return "" + ((aLong + 1) * 20);
}
}).take(5);
Observable<String> ob2 = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Exception {
Thread.sleep(350);
emitter.onNext("1");
Thread.sleep(250);
emitter.onNext("1");
emitter.onComplete();
}
});
Observable.merge(ob1, ob2).subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
Log.e(TAG, "accept : " + s);
}
});
输出结果：
accept : 20
accept : 40
accept : 60
accept : 1
accept : 80
accept : 100
accept : 1
4.5 StartWith
在数据序列的开头插入一条指定的项。
如果你想要一个 Observable 在发射数据之前先发射一个指定的数据序列，可以使用 StartWith 操作符。（如果你想一个 Observable 发射的数据末尾追加一个数据序列可以使用 Concat 操作符）。
示例代码：
Observable<Integer> ob1 = Observable.range(2, 3);
ob1.startWith(4).subscribe(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer integer) throws Exception {
Log.e(TAG, "accept : " + integer);
}
});
输出结果：
accept : 4
accept : 2
accept : 3
accept : 4
4.6 Switch
将一个发射 Observables 的 Observable 转换成另一个 Observable，后者发射这些 Observables 最近发射的数据。
Switch 订阅一个发射多个 Observables 的 Observable。它每次观察那些 Observables 中的一个，Switch 返回的这个 Observable 取消订阅前一个发射数据的 Observable，开始发射最近的 Observable 发射的数据。注意：当原始 Observable 发射了一个新的 Observable 时（不是这个新的 Observable 发射了一条数据时），它将取消订阅之前的那个 Observable。这意味着，在后来那个 Observable 产生之后到它开始发射数据之前的这段时间里，前一个 Observable 发射的数据将被丢弃（就像图例上的那个黄色圆圈一样）。
示例代码：
Observable.switchOnNext(Observable.interval(100, TimeUnit.MILLISECONDS)
.map(new Function<Long, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(@NonNull final Long aL2) throws Exception {
// Log.e(TAG, "al2 =" + aL2);
return Observable.interval(30, TimeUnit.MILLISECONDS)
.map(new Function<Long, String>() {
@Override
public String apply(@NonNull Long aL3) throws Exception {
// Log.e(TAG, "al3 =" + aL3);
return "al2 =" + aL2 + " , al3 = " + aL3;
}
});
}
}))
.take(6)
.subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
Log.e(TAG, "accept : " + s);
}
});
输出结果：
accept : al2 =0 , al3 = 0
accept : al2 =0 , al3 = 1
accept : al2 =0 , al3 = 2
accept : al2 =1 , al3 = 0
accept : al2 =1 , al3 = 1
accept : al2 =2 , al3 = 0
4.7 Zip
使用一个函数组合多个 Observable 发射的数据集合，然后再发射这个结果。
Zip 操作符返回一个 Obversable，它使用这个函数按顺序结合两个或多个 Observables 发射的数据项，然后它发射这个函数返回的结果。它按照严格的顺序应用这个函数。它只发射与发射数据项最少的那个 Observable 一样多的数据。
RxJava 将这个操作符实现为 zip 和 zipWith。
4.7.1 Zip
示例代码：
Observable<Integer> ob1 = Observable.just(1, 2, 3, 4, 5);
Observable<String> ob2 = Observable.just("A", "B", "C", "D");
Observable.zip(ob1, ob2, new BiFunction<Integer, String, String>() {
@Override
public String apply(@NonNull Integer i, @NonNull String s) throws Exception {
return i + " + " + s;
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
Log.e(TAG, "accept:" + s);
}
});
输出结果：
accept:1 + A
accept:2 + B
accept:3 + C
accept:4 + D
3.7.2 ZipWith
示例代码：
Observable<Integer> ob1 = Observable.just(1, 2, 3, 4, 5);
Observable<String> ob2 = Observable.just("A", "B", "C", "D");
ob1.zipWith(ob2, new BiFunction<Integer, String, String>() {
@Override
public String apply(@NonNull Integer i, @NonNull String s) throws Exception {
return i + "-" + s;
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
Log.e(TAG, "accept:" + s);
}
});
输出结果：
accept:1-A
accept:2-B
accept:3-C
accept:4-D
