RxJava（三）-合并操作符

1.concat和concatArray
  是将多个被观察者合并后，发射出去。

Observable o1 = Observable.just(1,2,3);
Observable o2 = Observable.just("a","b","c");
Observable.concat(o1,o2).subscribe(new Observer() {
    @Override
    public void onSubscribe(@NonNull Disposable d) {
 
    }
 
    @Override
    public void onNext(@NonNull Object o) {
         Log.e("rxjava","onNext "+o.toString());
    }
 
    @Override
    public void onError(@NonNull Throwable e) {
 
    }
 
    @Override
    public void onComplete() {
 
    }
});
执行结果：
rxjava: onNext 1
rxjava: onNext 2
rxjava: onNext 3
rxjava: onNext a
rxjava: onNext b
rxjava: onNext c

concat和concatArray都是返回的new ObservableConcatMap(fromArray(sources), ...)对象，接收sources对象是一个ObservableSource数组。
fromArray(sources),就是传递进来的被观察者集合
在调用concat时，在RxJava内部调用的还是concatArray
2.merge和mergeArray
  将多个被观察者合并后发射出去

List list = new ArrayList();
list.add(1);
list.add(2);
list.add(3);
Observable o1 = Observable.fromIterable(list);
Observable o2 = Observable.just("a","b","c");
Observable.merge(o1,o2).subscribe(new Observer() {
    @Override
    public void onSubscribe(@NonNull Disposable d) {
 
    }
 
    @Override
    public void onNext(@NonNull Object o) {
        Log.e("rxjava","onNext "+o.toString());
    }
 
    @Override
    public void onError(@NonNull Throwable e) {
 
    }
 
    @Override
    public void onComplete() {
 
    }
});
执行结果：
rxjava: onNext 1
rxjava: onNext 2
rxjava: onNext 3
rxjava: onNext a
rxjava: onNext b
rxjava: onNext c

merge和mergeArray调用的都是：
fromArray(sources).flatMap((Function)Functions.identity(), sources.length);
它的合并是通过flatMap实现的。
3.zip
  是将多个被观察者合并后，再转换新的对象返回给链上下游的观察者和被观察者。

bservable o1 = Observable.just(1,2,3);
Observable o2 = Observable.just("a","b","c");
Observable.zip(o1, o2, new BiFunction<Integer, String, String>() {
    @NonNull
    @Override
    public String apply(@NonNull Integer integer, @NonNull String string) throws Exception {
        return integer+"_"+string;
    }
}).subscribe(new Observer<String>() {
    @Override
    public void onSubscribe(@NonNull Disposable d) {
 
    }
 
    @Override
    public void onNext(@NonNull String o) {
         Log.e("rxjava","onNext "+o);
    }
 
    @Override
    public void onError(@NonNull Throwable e) {
 
    }
 
    @Override
    public void onComplete() {
 
    }
});
执行结果：
rxjava: onNext 1_a
rxjava: onNext 2_b
rxjava: onNext 3_c

zip返回的是ObservableZip(ObservableSource[] sources,Function zipper)对象，sources就是合并后的Observable。

zipper 就是传递进来的BiFunction类，通过apply函数，返回合并后的结果，并将结果传递给下游的对象。