CombineLatest — 把两个Observable产生的结果进行合并，合并的结果组成一个新的Observable。最后一个Observable中的每一个数据项，都与前面Observable中的最后一项进行相应运算，将生成的结果组成一个新的Observable对象.。
private static void andThenWhen() {
Observable<Integer> o1 = Observable.just(1, 2, 3);
Observable<Integer> o2 = Observable.just(4, 5, 6);
Observable<Integer> o3 = Observable.just(7, 8, 9);
Observable.combineLatest(o1, o2, o3, new Function3<Integer, Integer, Integer, String>() {
@Override
public String apply(@NonNull Integer integer, @NonNull Integer integer2, @NonNull Integer integer3) throws Exception {
return integer + " - " + integer2 + " - " + integer3;
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String string) throws Exception {
System.out.println("onNext: " + string);
}
});
}
}
输出结果：
onNext: 3 - 6 - 7
onNext: 3 - 6 - 8
onNext: 3 - 6 - 9
Join —在另一个Observable发射的数据定义的时间窗口内，这个Observable发射了一条数据，就结合两个Observable发射的数据。
private static void join() {
Observable<Integer> o1 = Observable.just(1, 2, 3);
Observable<Integer> o2 = Observable.just(4, 5, 6);
Observable<Integer> o3 = Observable.just(7, 8, 9);
o1.join(o2, new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
return Observable.just(integer + "-o1").delay(200, TimeUnit.MILLISECONDS);
}
}, new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
return Observable.just(integer + "-o2").delay(200, TimeUnit.MILLISECONDS);
}
}, new BiFunction<Integer, Integer, String>() {
@Override
public String apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {
return integer + " - " + integer2;
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println("onNext" + s);
}
});
}
输出结果：
onNext1 - 4
onNext2 - 4
onNext3 - 4
onNext1 - 5
onNext2 - 5
onNext3 - 5
onNext1 - 6
onNext2 - 6
onNext3 - 6
Merge — 合并多个Observable为一个，可能会让合并的Observables发射的数据交错。
private static void merge() {
Observable<Integer> o1 = Observable.just(1, 2, 3);
Observable<Integer> o2 = Observable.just(4, 5, 6);
Observable.merge(o1, o2)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("onNext: " + integer);
}
});
}
输出结果：
onNext: 1
onNext: 2
onNext: 3
onNext: 4
onNext: 5
onNext: 6
StartWith —一个Observable在发射源数据之前先发射一个指定的数据或数据序列
private static void startWith() {
Observable
.just(1, 2, 3)
.startWith(0)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("onNext: " + integer);
}
});
}
输出结果：
onNext: 0
onNext: 1
onNext: 2
onNext: 3
Switch — switchOnNext操作符是把一组Observable转换成一个Observable,在同一个时间内存在两个或多个Observable提交的结果，只取最后一个Observable提交的结果给订阅者
private static void rxSwitch() {
Observable<Observable<Long>> o1 = Observable.interval(500, TimeUnit.MILLISECONDS)
.map(new Function<Long, Observable<Long>>() {
@Override
public Observable<Long> apply(@NonNull Long aLong) throws Exception {
return Observable.interval(200, TimeUnit.MILLISECONDS).map(new Function<Long, Long>() {
@Override
public Long apply(@NonNull Long aLong) throws Exception {
return aLong * 10;
}
}).take(5);
}
})
.take(2);
Observable
.switchOnNext(o1)
.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
System.out.println("onNext:" + aLong);
}
});
while (true);
}
输出结果：
onNext:0
onNext:10
onNext:0
onNext:10
onNext:20
onNext:30
onNext:40
Zip —把两个observable提交的结果，严格按照顺序进行合并
private static void rxZip() {
Observable
.zip(Observable.just(0, 1, 2), Observable.just("a", "b", "c"), new BiFunction<Integer, String, String>() {
@Override
public String apply(@NonNull Integer integer, @NonNull String s) throws Exception {
return "zip: " + 0 + " - " + s;
}
})
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println(s);
}
});
}
输出结果：
zip: 0 - a
zip: 0 - b
zip: 0 - c
