Rxjava2 操作符 - Combining Observab

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