Rxjava2 组合 / 合并操作符
- 组合 多个被观察者（
Observable） & 合并需要发送的事件
1 组合多个被观察者
concat（） / concatArray（）
- 组合多个被观察者一起发送数据，合并后 按发送顺序串行执行
二者区别：组合被观察者的数量，即concat（）组合被观察者数量≤4个，而concatArray（）则>0个
merge（） / mergeArray（）
- 组合多个被观察者一起发送数据，合并后 按时间线并行执行
concatDelayError（） / mergeDelayError（）
concatArrayDelayError（）/ mergeArrayDelayError()
2 合并多个事件
Zip（）
- 合并 多个被观察者（
Observable）发送的事件，生成一个新的事件序列（即组合过后的事件序列），并最终发送
- 特别注意：
- 事件组合方式 = 严格按照原先事件序列 进行对位合并
- 最终合并的事件数量 = 多个被观察者（
Observable）中数量最少的数量
<-- 创建第1个被观察者 -->
Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "被观察者1发送了事件1");
emitter.onNext(1);
// 为了方便展示效果，所以在发送事件后加入2s的延迟
Thread.sleep(1000);
Log.d(TAG, "被观察者1发送了事件2");
emitter.onNext(2);
Thread.sleep(1000);
Log.d(TAG, "被观察者1发送了事件3");
emitter.onNext(3);
Thread.sleep(1000);
emitter.onComplete();
}
}).subscribeOn(Schedulers.io()); // 设置被观察者1在工作线程1中工作
<-- 创建第2个被观察者 -->
Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(TAG, "被观察者2发送了事件A");
emitter.onNext("A");
Thread.sleep(1000);
Log.d(TAG, "被观察者2发送了事件B");
emitter.onNext("B");
Thread.sleep(1000);
Log.d(TAG, "被观察者2发送了事件C");
emitter.onNext("C");
Thread.sleep(1000);
Log.d(TAG, "被观察者2发送了事件D");
emitter.onNext("D");
Thread.sleep(1000);
emitter.onComplete();
}
}).subscribeOn(Schedulers.newThread());// 设置被观察者2在工作线程2中工作
// 假设不作线程控制，则该两个被观察者会在同一个线程中工作，即发送事件存在先后顺序，而不是同时发送
<-- 使用zip变换操作符进行事件合并 -->
// 注：创建BiFunction对象传入的第3个参数 = 合并后数据的数据类型
Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
@Override
public String apply(Integer integer, String string) throws Exception {
return integer + string;
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe");
}
@Override
public void onNext(String value) {
Log.d(TAG, "最终接收到的事件 = " + value);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError");
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
特别注意：
- 尽管被观察者2的事件
D没有事件与其合并，但还是会继续发送 - 若在被观察者1 & 被观察者2的事件序列最后发送
onComplete()事件，则被观察者2的事件D也不会发送，测试结果如下
combineLatest（）
当两个Observables中的任何一个发送了数据后，将先发送了数据的Observables 的最新（最后）一个数据 与 另外一个Observable发送的每个数据结合，最终基于该函数的结果发送数据
与Zip（）的区别：Zip（） = 按个数合并，即1对1合并；
CombineLatest（） = 按时间合并，即在同一个时间点上合并
Observable.combineLatest(
Observable.just(1L, 2L, 3L), // 第1个发送数据事件的Observable
Observable.intervalRange(0, 3, 1, 1, TimeUnit.SECONDS), // 第2个发送数据事件的Observable：从0开始发送、共发送3个数据、第1次事件延迟发送时间 = 1s、间隔时间 = 1s
new BiFunction<Long, Long, Long>() {
@Override
public Long apply(Long o1, Long o2) throws Exception {
// o1 = 第1个Observable发送的最新（最后）1个数据
// o2 = 第2个Observable发送的每1个数据
Log.e(TAG, "数据是： "+ o1 + " "+ o2);
return o1 + o2;
// 合并的逻辑 = 相加
// 即第1个Observable发送的最后1个数据 与 第2个Observable发送的每1个数据进行相加
}
}).subscribe(new Consumer<Long>() {
@Override
public void accept(Long s) throws Exception {
Log.e(TAG, "结果是： "+s);
}
});
combineLatestDelayError（）
作用类似于concatDelayError（） / mergeDelayError（） ，即错误处理
reduce（）
把被观察者需要发送的事件聚合成1个事件 & 发送
本质都是前2个数据聚合，然后与后1个数据继续进行聚合，依次类推
Observable.just(1,2,3,4)
.reduce(new BiFunction<Integer, Integer, Integer>() {
// 在该复写方法中复写聚合的逻辑
@Override
public Integer apply(@NonNull Integer s1, @NonNull Integer s2) throws Exception {
Log.e(TAG, "本次计算的数据是： "+s1 +" 乘 "+ s2);
return s1 * s2;
// 原理：第1次取前2个数据相乘，之后每次获取到的数据 = 返回的数据x原始下1个数据每
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer s) throws Exception {
Log.e(TAG, "最终计算的结果是： "+s);
}
});
collect（）
将被观察者Observable发送的数据事件收集到一个数据结构里
3 发送事件前追加发送事件
startWith（） / startWithArray（）
在一个被观察者发送事件前，追加发送一个数据 / 一些新的被观察者
4 统计发送事件数量
count（）
统计被观察者发送事件的数量
Observable.just(1, 2, 3, 4)
.count()
.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
Log.e(TAG, "发送的事件数量 = "+aLong);
}
});
发送的事件数量 = 4