@(Rxjava学习笔记)

RxJava 学习笔记（八） — Combining 结合操作

RxJava 学习笔记八 Combining 结合操作
- StartWith 在数据序列的开头插入一条指定的项
- Merge 将多个Observable合并为一个
- MergeDelayError 将多个Observable合并为一个
- Zip 使用一个函数组合多个Observable发射的数据集合然后再发射这个结果
- CombineLatest 当两个Observables中的任何一个发射了一个数据时通过一个指定的函数组合每个Observable发射的最新数据一共两个数据然后发射这个函数的结果
- Join 无论何时如果一个Observable发射了一个数据项只要在另一个Observable发射的数据项定义的时间窗口内就将两个Observable发射的数据合并发射
- GroupJoin groupJoin操作符非常类似于join操作符区别在于join操作符中第四个参数的传入函数不一致
- SwitchOnNext 将一个发射Observables的Observable转换成另一个Observable后者发射这些Observables最近发射的数据

1. `StartWith` —> 在数据序列的开头插入一条指定的项

一个Observable在发射数据之前先发射一个指定的数据序列

示例代码：

Observable.just(10,20,30).startWith(2, 3, 4).subscribe(new Subscriber<Integer>() {@Overridepublic void onCompleted() {System.out.println("Sequence complete.");}@Overridepublic void onError(Throwable e) {System.err.println("Error: " + e.getMessage());}@Overridepublic void onNext(Integer value) {System.out.println("Next:" + value);}});

输出：

Next:2
Next:3
Next:4
Next:10
Next:20
Next:30
Sequence complete.

Javadoc：startWith(T) (最多接受九个参数)
Javadoc：startWith(Iterable)

你也可以传递一个Observable给startWith，它会将那个Observable的发射物插在原始Observable发射的数据序列之前，然后把这个当做自己的发射物集合。这可以看作是Concat的反转。

示例代码：

Observable.just(1,2,3,4,5).startWith(Observable.just(9,8,5)).subscribe(new Action1<Integer>() {@Overridepublic void call(Integer integer) {System.out.println(integer+"");}});

输出：

Javadoc：startWith(Observable)

2. `Merge` —> 将多个Observable合并为一个

使用Merge操作符你可以将多个Observables的输出合并，就好像它们是一个单个的Observable一样。

Merge可能会让合并的Observables发射的数据交错（有一个类似的操作符Concat不会让数据交错，它会按顺序一个接着一个发射多个Observables的发射物）。

正如图例上展示的，任何一个原始Observable的onError通知会被立即传递给观察者，而且会终止合并后的Observable。

示例代码：

Observable<Integer> odds = Observable.just(1, 3, 5).subscribeOn(someScheduler);
Observable<Integer> evens = Observable.just(2, 4, 6);Observable.merge(odds, evens).subscribe(new Subscriber<Integer>() {@Overridepublic void onNext(Integer item) {System.out.println("Next: " + item);}@Overridepublic void onError(Throwable error) {System.err.println("Error: " + error.getMessage());}@Overridepublic void onCompleted() {System.out.println("Sequence complete.");}});

输出：

Next: 1
Next: 3
Next: 5
Next: 2
Next: 4
Next: 6
Sequence complete.

除了传递多个Observable给merge，你还可以传递一个Observable列表List，数组，甚至是一个发射Observable序列的Observable，merge将合并它们的输出作为单个Observable的输出：

如果你传递一个发射Observables序列的Observable，你可以指定merge应该同时订阅的Observable‘的最大数量。一旦达到订阅数的限制，它将不再订阅原始Observable发射的任何其它Observable，直到某个已经订阅的Observable发射了onCompleted通知。

Javadoc：merge(Observable,Observable) (接受二到九个Observable)
Javadoc：merge(Observable[])
Javadoc：merge(Iterable)
Javadoc：merge(Iterable,int)

3. `MergeDelayError` —> 将多个Observable合并为一个

它的行为有一点不同，它会保留onError通知直到合并后的Observable所有的数据发射完成，在那时它才会把onError传递给观察者。

示例代码：

        //产生0,5,10数列,最后会产生一个错误Observable<Long> errorObservable = Observable.error(new Exception("this is end!"));Observable < Long > observable1 = Observable.timer(0, 1000, TimeUnit.MILLISECONDS).map(new Func1<Long, Long>() {@Overridepublic Long call(Long aLong) {return aLong * 5;}}).take(3).mergeWith(errorObservable.delay(3500, TimeUnit.MILLISECONDS));//产生0,10,20,30,40数列Observable<Long> observable2 = Observable.timer(500, 1000, TimeUnit.MILLISECONDS).map(new Func1<Long, Long>() {@Overridepublic Long call(Long aLong) {return aLong * 10;}}).take(5);Observable.mergeDelayError(observable1, observable2).subscribe(new Subscriber<Long>() {@Overridepublic void onCompleted() {System.out.println("Sequence complete.");}@Overridepublic void onError(Throwable e) {System.err.println("Error: " + e.getMessage());}@Overridepublic void onNext(Long aLong) {System.out.println("Next:" + aLong);}});

输出：

Next:0
Next:0
Next:10
Next:20
Next:30
Next:40
Error: this is end!

Javadoc：mergeDelayError(Observable)
Javadoc：mergeDelayError(Observable,Observable)

4. `Zip` —> 使用一个函数组合多个Observable发射的数据集合，然后再发射这个结果

Zip
通过一个函数将多个Observables的发射物结合到一起，基于这个函数的结果为每个结合体发射单个数据项。

Zip操作符返回一个Obversable，它使用这个函数按顺序结合两个或多个Observables发射的数据项，然后它发射这个函数返回的结果。它按照严格的顺序应用这个函数。它只发射与发射数据项最少的那个Observable一样多的数据。

zip的最后一个参数接受每个Observable发射的一项数据，返回被压缩后的数据，它可以接受一到九个参数：一个Observable序列，或者一些发射Observable的Observables。

Observable<Integer> observable1 = Observable.just(10,20,30);Observable<Integer> observable2 = Observable.just(4, 8, 12, 16);Observable.zip(observable1, observable2, new Func2<Integer, Integer, Integer>() {@Overridepublic Integer call(Integer integer, Integer integer2) {return integer + integer2;}}).subscribe(new Subscriber<Integer>() {@Overridepublic void onCompleted() {System.out.println("Sequence complete.");}@Overridepublic void onError(Throwable e) {System.err.println("Error: " + e.getMessage());}@Overridepublic void onNext(Integer value) {System.out.println("Next:" + value);}});

输出：

Next:14
Next:28
Next:42
Sequence complete.

Javadoc：zip(Iterable,FuncN)
Javadoc：zip(Observable,FuncN)
Javadoc：zip(Observable,Observable,Func2) (最多可以有九个Observables参数)

ZipWith

zipWith操作符总是接受两个参数，第一个参数是一个Observable或者一个Iterable。

zip和zipWith默认不在任何特定的操作符上执行。

Javadoc：zipWith(Observable,Func2)
Javadoc：zipWith(Iterable,Func2)

5. `CombineLatest` —> 当两个Observables中的任何一个发射了一个数据时，通过一个指定的函数组合每个Observable发射的最新数据（一共两个数据），然后发射这个函数的结果

CombineLatest操作符行为类似于zip，但是只有当原始的Observable中的每一个都发射了一条数据时zip才发射数据。CombineLatest则在原始的Observable中任意一个发射了数据时发射一条数据。当原始Observables的任何一个发射了一条数据时，CombineLatest使用一个函数结合它们最近发射的数据，然后发射这个函数的返回值。

示例代码:

//产生0,5,10,15,20数列Observable<Long> observable1 = Observable.timer(0, 1000, TimeUnit.MILLISECONDS).map(new Func1<Long, Long>() {@Overridepublic Long call(Long aLong) {return aLong * 5;}}).take(5);//产生0,10,20,30,40数列Observable<Long> observable2 = Observable.timer(500, 1000, TimeUnit.MILLISECONDS).map(new Func1<Long, Long>() {@Overridepublic Long call(Long aLong) {return aLong * 10;}}).take(5);Observable.combineLatest(observable1, observable2, new Func2<Long, Long, Long>() {@Overridepublic Long call(Long aLong, Long aLong2) {return aLong+aLong2;}}).subscribe(new Subscriber<Long>() {@Overridepublic void onCompleted() {System.out.println("Sequence complete.");}@Overridepublic void onError(Throwable e) {System.err.println("Error: " + e.getMessage());}@Overridepublic void onNext(Long aLong) {System.out.println("Next: " + aLong);}});

输出：

Next: 0
Next: 5
Next: 15
Next: 20
Next: 30
Next: 35
Next: 45
Next: 50
Next: 60
Sequence complete.

Javadoc：combineLatest(List,FuncN)
Javadoc：combineLatest(Observable,Observable,Func2)

6. `Join` —> 无论何时，如果一个Observable发射了一个数据项，只要在另一个Observable发射的数据项定义的时间窗口内，就将两个Observable发射的数据合并发射

zip()和merge()方法作用在发射数据的范畴内，在决定如何操作值之前有些场景我们需要考虑时间的。RxJava的join()函数基于时间窗口将两个Observables发射的数据结合在一起。+

join方法的用法如下：
observableA.join(observableB,
observableA产生结果生命周期控制函数，
observableB产生结果生命周期控制函数，
observableA产生的结果与observableB产生的结果的合并规则）

示例代码：

//产生0,5,10,15,20数列Observable<Long> observable1 = Observable.timer(0, 1000, TimeUnit.MILLISECONDS).map(new Func1<Long, Long>() {@Overridepublic Long call(Long aLong) {return aLong * 5;}}).take(5);//产生0,10,20,30,40数列Observable<Long> observable2 = Observable.timer(500, 1000, TimeUnit.MILLISECONDS).map(new Func1<Long, Long>() {@Overridepublic Long call(Long aLong) {return aLong * 10;}}).take(5);observable1.join(observable2, new Func1<Long, Observable<Long>>() {@Overridepublic Observable<Long> call(Long aLong) {//使Observable延迟600毫秒执行return Observable.just(aLong).delay(600, TimeUnit.MILLISECONDS);}}, new Func1<Long, Observable<Long>>() {@Overridepublic Observable<Long> call(Long aLong) {//使Observable延迟600毫秒执行return Observable.just(aLong).delay(600, TimeUnit.MILLISECONDS);}}, new Func2<Long, Long, Long>() {@Overridepublic Long call(Long aLong, Long aLong2) {return aLong + aLong2;}}).subscribe(new Subscriber<Long>() {@Overridepublic void onCompleted() {System.out.println("Sequence complete.");}@Overridepublic void onError(Throwable e) {System.err.println("Error: " + e.getMessage());}@Overridepublic void onNext(Long aLong) {System.out.println("Next: " + aLong);}});

输出：

Next: 0
Next: 5
Next: 15
Next: 20
Next: 30
Next: 35
Next: 45
Next: 50
Next: 60
Sequence complete.

Javadoc：Join(Observable,Func1,Func1,Func2)

7. `GroupJoin` —> groupJoin操作符非常类似于join操作符，区别在于join操作符中第四个参数的传入函数不一致

示例代码：

        //产生0,5,10,15,20数列Observable<Long> observable1 = Observable.timer(0, 1000, TimeUnit.MILLISECONDS).map(new Func1<Long, Long>() {@Overridepublic Long call(Long aLong) {return aLong * 5;}}).take(5);//产生0,10,20,30,40数列Observable<Long> observable2 = Observable.timer(500, 1000, TimeUnit.MILLISECONDS).map(new Func1<Long, Long>() {@Overridepublic Long call(Long aLong) {return aLong * 10;}}).take(5);observable1.groupJoin(observable2, new Func1<Long, Observable<Long>>() {@Overridepublic Observable<Long> call(Long aLong) {return Observable.just(aLong).delay(1600, TimeUnit.MILLISECONDS);}}, new Func1<Long, Observable<Long>>() {@Overridepublic Observable<Long> call(Long aLong) {return Observable.just(aLong).delay(600, TimeUnit.MILLISECONDS);}}, new Func2<Long, Observable<Long>, Observable<Long>>() {@Overridepublic Observable<Long> call(Long aLong, Observable<Long> observable) {return observable.map(new Func1<Long, Long>() {@Overridepublic Long call(Long aLong2) {return aLong + aLong2;}});}}).subscribe(new Subscriber<Observable<Long>>() {@Overridepublic void onCompleted() {System.out.println("Sequence complete.");}@Overridepublic void onError(Throwable e) {System.err.println("Error: " + e.getMessage());}@Overridepublic void onNext(Observable<Long> observable) {observable.subscribe(new Subscriber<Long>() {@Overridepublic void onCompleted() {}@Overridepublic void onError(Throwable e) {}@Overridepublic void onNext(Long aLong) {System.out.println("Next: " + aLong);}});}});

输出：

Next: 5
Next: 15
Next: 10
Next: 20
Next: 25
Next: 30
Next: 35
Next: 45
Next: 40
Next: 50
Next: 60
Next: 55
Sequence complete.

Javadoc：groupJoin(Observable,Func1,Func1,Func2)

8. `SwitchOnNext` —> 将一个发射Observables的Observable转换成另一个Observable，后者发射这些Observables最近发射的数据

switchOnNext操作符是把一组Observable转换成一个Observable，转换规则为：对于这组Observable中的每一个Observable所产生的结果，如果在同一个时间内存在两个或多个Observable提交的结果，只取最后一个Observable提交的结果给订阅者

示例代码：

        //每隔500毫秒产生一个observableObservable<Observable<Long>> observable = Observable.timer(0, 500, TimeUnit.MILLISECONDS).map(new Func1<Long, Observable<Long>>() {@Overridepublic Observable<Long> call(Long aLong) {//每隔200毫秒产生一组数据（0,10,20,30,40)return Observable.timer(0, 200, TimeUnit.MILLISECONDS).map(new Func1<Long, Long>() {@Overridepublic Long call(Long aLong) {return aLong * 10;}}).take(5);}}).take(2);Observable.switchOnNext(observable).subscribe(new Subscriber<Long>() {@Overridepublic void onCompleted() {System.out.println("Sequence complete.");}@Overridepublic void onError(Throwable e) {System.err.println("Error: " + e.getMessage());}@Overridepublic void onNext(Long aLong) {System.out.println("Next:" + aLong);}});

输出：

Next:0
Next:10
Next:20
Next:0
Next:10
Next:20
Next:30
Next:40
Sequence complete.

Javadoc：switchOnNext(Observable)