前言
在RxJava中， 同时处理多个Observables是很常见的一种操作。下面我们简单分析下几个组合类的操作符。
- Merge
在异步的世界经常会创建这样的场景，我们有多个来源但是只想有一个结果：多输入，单输出。RxJava的merge()方法将帮助你把两个甚至更多的Observables合并到他们发射的数据里。下图给出了把两个序列合并在一个最终发射的Observable。
merge.png
/**
* Merge 解释：将2-9个Observables合并到一个Observable中进行发射，合并后的数据可能会是交错（无序）的（如果想要没有交错，
* 可以使用concat操作符） merge还可以传递一个Observable列表List，数组
* 甚至是一个发射Observable序列的Observable，merge将合并它们的输出作为单个Observable的输出
*/
Observable<String> letterObservable = Observable.just("A", "B", "C", "D");
Observable<Integer> numberObservable = Observable.just(1, 2, 3, 4);
Observable.merge(letterObservable, numberObservable)
.subscribe(new Action1<Serializable>() {
@Override
public void call(Serializable serializable) {
System.out.println("value =" + serializable);
}
});
}
value =A
value =B
value =C
value =D
value =1
value =2
value =3
value =4
例如：我们加载Recyclerview数据时，我们从多个地方获取，但是只有一个输出，可以用merge处理
2.Zip
我们在处理多源时可能会带来这样一种场景：多从个Observables接收数据，处理它们，然后将它们合并成一个新的可观测序列来使用。RxJava有一个特殊的方法可以完成：zip()合并两个或者多个Observables发射出的数据项，根据指定的函数Func*变换它们，并发射一个新值。下图展示了zip()方法如何处理发射的“numbers”和“letters”然后将它们合并一个新的数据项：
zip.png
如图可以看到，zip操作符并不是像merge操作符那样只合并了数据，重要的是发生了质的变化。
Observable letterObservable = Observable.just("A", "B", "C", "D");
Observable numberObservable = Observable.just(1, 2, 3, 4, 5);
Observable.zip(letterObservable, numberObservable, new Func2<String, Integer, String>() {
@Override
public String call(String o, Integer o2) {
return o + o2;
}
}).subscribe(new Action1<String>() {
@Override
public void call(String o) {
System.out.println("value :"+o);
}
});
value :A1
value :B2
value :C3
value :D4
注意：
zip操作符合并后的Observable数据类型可以发生改变。
3.combineLatest
RxJava的combineLatest()函数有点像zip()函数的特殊形式。正如我们已经学习的，zip()作用于最近未打包的两个Observables。相反，combineLatest()作用于最近发射的数据项：如果Observable1发射了A并且Observable2发射了B和C，combineLatest()将会分组处理AB和AC，如下图所示：
combineLatest.png
Observable letterObservable = Observable.just("a", "b", "c");
Observable numberObservable = Observable.just(1, 2, 3);
Observable.combineLatest(letterObservable, numberObservable, new Func2<String, Integer, String>() {
@Override
public String call(String t1, Integer t2) {
System.out.println("combine t1 = " + t1 + " | t2 = " + t2);// t1的值一开始就是letterObservable中的最后一个值
return t1 + t2;
}
}).subscribe(new Action1() {
@Override
public void call(Object o) {
System.out.println("value = " + o);
}
});
combine t1 = c | t2 = 1
value = c1
combine t1 = c | t2 = 2
value = c2
combine t1 = c | t2 = 3
value = c3
- Join
前面两个方法，zip()和merge()方法作用在发射数据的范畴内，在决定如何操作值之前有些场景我们需要考虑时间的。RxJava的join()函数基于时间窗口将两个Observables发射的数据结合在一起。
join.png
为了正确的理解上一张图，我们解释下join()需要的参数：
第二个Observable和源Observable结合。
Func1参数：在指定的由时间窗口定义时间间隔内，源Observable发射的数据和从第二个Observable发射的数据相互配合返回的Observable。
Func1参数：在指定的由时间窗口定义时间间隔内，第二个Observable发射的数据和从源Observable发射的数据相互配合返回的Observable。
Func2参数：定义已发射的数据如何与新发射的数据项相结合。
Observable<Integer> observableA=Observable.range(1,5);
List<Integer> data= Arrays.asList(6,7,8,9,10);
Observable<Integer> observableB=Observable.from(data);
observableA.join(observableB, new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer integer) {
return Observable.just(integer).delay(1, TimeUnit.SECONDS);
}
}, new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer integer) {
return Observable.just(integer);
}
},new Func2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer value1, Integer value2) {
System.out.println("left: " + value1 + " right:" + value2);
return value1 + value2;
}
}).subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer integer) {
System.out.println("onNext value = " + integer);
}
});
一句话概括：在observableA的生命周期内：observableB输出的数据项与observableA输出的数据项每个合并
left: 1 right:6
onNext value = 7
left: 2 right:6
onNext value = 8
left: 3 right:6
onNext value = 9
left: 4 right:6
onNext value = 10
left: 5 right:6
onNext value = 11
left: 1 right:7
onNext value = 8
left: 2 right:7
onNext value = 9
left: 3 right:7
onNext value = 10
left: 4 right:7
onNext value = 11
left: 5 right:7
onNext value = 12。。。。。。
5.StartWith
在源Observable输出之前插入指定数据项
startWith.png
Observable.just(7,8,9).startWith(11,12).subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer integer) {
System.out.println("onNext t = " + integer);
}
});
onNext t = 11
onNext t = 12
onNext t = 7
onNext t = 8
onNext t = 9
网友评论