RxJava2.0 操作符（4）—— Combin 组合操作

作者: DoubleThunder | 来源:发表于2017-10-16 23:07 被阅读86次

这个页面展示的操作符可用于组合多个 Observables。

And/Then/When — 通过模式和计划组合多个 Observables 发射的数据集合。
CombineLatest — 当两个 Observables 中的任何一个发射了一个数据时，通过一个指定的函数组合每个 Observable 发射的最新数据（一共两个数据），然后发射这个函数的结果。
Join — 无论何时，如果一个 Observable 发射了一个数据项，只要在另一个 Observable 发射的数据项定义的时间窗口内，就将两个 Observable 发射的数据合并发射。
Merge — 将多个 Observable 合并为一个。
StartWith — 在数据序列的开头插入一条指定的项。
Switch — 将一个发射 Observables 的 Observable 转换成另一个 Observable，后者发射这些 Observables 最近发射的数据。
Zip — 使用一个函数组合多个 Observable 发射的数据集合，然后再发射这个结果。

4.1 And/Then/When

通过模式和计划组合多个 Observables 发射的数据集合。

and,then,when

示例代码：

无

4.2 CombineLatest

当两个 Observables 中的任何一个发射了一个数据时，通过一个指定的函数组合每个 Observable 发射的最新数据（一共两个数据），然后发射这个函数的结果。

CombineLatest

CombineLatest 操作符行为类似于 zip，但是只有当原始的 Observable 中的每一个都发射了一条数据时 zip 才发射数据。CombineLatest 则在原始的 Observable 中任意一个发射了数据时发射一条数据。当原始 Observables 的任何一个发射了一条数据时，CombineLatest 使用一个函数结合它们最近发射的数据，然后发射这个函数的返回值。

实践：可用于填写表单提交，需要满足多个条件才触发“提交”按钮。

示例代码：

//2个有差值的observable
Observable<Long> ob1 = Observable.interval(100, TimeUnit.MILLISECONDS);
Observable<Long> ob2 = Observable.interval(150, TimeUnit.MILLISECONDS);

Observable.combineLatest(ob1, ob2, new BiFunction<Long, Long, String>() {
    @Override
    public String apply(@NonNull Long aLong, @NonNull Long aLong2) throws Exception {
        return  aLong + " - " + aLong2;
    }
}).take(6).subscribe(new Consumer<String>() {
    @Override
    public void accept(@NonNull String s) throws Exception {
        Log.e(TAG, "accept : " + s);
    }
});

输出结果：

accept : 0 - 0
accept : 1 - 0
accept : 1 - 1
accept : 2 - 1
accept : 3 - 1
accept : 3 - 2

4.3 Join

无论何时，如果一个 Observable 发射了一个数据项，只要在另一个 Observable 发射的数据项定义的时间窗口内，就将两个 Observable 发射的数据合并发射。

Join

Join 操作符结合两个 Observable 发射的数据，基于时间窗口（你定义的针对每条数据特定的原则）选择待集合的数据项。你将这些时间窗口实现为一些 Observables，它们的生命周期从任何一条 Observable 发射的每一条数据开始。当这个定义时间窗口的 Observable 发射了一条数据或者完成时，与这条数据关联的窗口也会关闭。只要这条数据的窗口是打开的，它将继续结合其它 Observable 发射的任何数据项。你定义一个用于结合数据的函数。
示例代码：

Observable<Integer> ob1 = Observable.just(1, 2);
Observable<String> ob2 = Observable.just("A", "B", "C");
ob1.join(ob2, new Function<Integer, ObservableSource<String>>() {//ob1产生结果声明周期控制函数
        @Override
        public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
//              return Observable.just(integer + "00");
            return Observable.just(integer + "00").delay(100, TimeUnit.MILLISECONDS);
//              return Observable.just(integer + "00").delay(200, TimeUnit.MILLISECONDS);
//              return Observable.just(integer + "00").delay(300, TimeUnit.MILLISECONDS);
        }
    }, new Function<String, ObservableSource<String>>() {//ob2产生结果声明周期控制函数
        @Override
        public ObservableSource<String> apply(@NonNull String s) throws Exception {
            return Observable.just("Function1:" + s).delay(200, TimeUnit.MILLISECONDS);
        }
    }, new BiFunction<Integer, String, Object>() {//ob1 和ob2产生结果的合并规则
        @Override
        public Object apply(@NonNull Integer integer, @NonNull String s) throws Exception {
//              Thread.sleep(1000);
            return  "string = " + s + ", integer = " + integer;
        }
    }

).subscribe(new Consumer<Object>() {
    @Override
    public void accept(@NonNull Object o) throws Exception {
        Log.e(TAG, "accept : " + o.toString());
    }
});

输出结果：

accept : string = A, integer = 1
accept : string = A, integer = 2
accept : string = B, integer = 1
accept : string = B, integer = 2
accept : string = C, integer = 1
accept : string = C, integer = 2

Join 的效果类似于排列组合，把第一个数据源 ob1 作为基座窗口，它根据自己的节奏不断发射数据元素。第二个数据源 ob2 ，每发射一个数据，我们都把它和第一个数据源 ob1 中已经发射的数据进行一对一匹配；

如果某一时刻 ob2 发射了一个数据 B ,此时 ob1 已经发射了 1，2 共两个数据，那么我们的合并操作就会把 B 依次与 1，2 配对，得到两组数据： (1，B)、(2，B)

4.3.1 GroupJoin

GroupJoin

GroupJoin 运算符与 Join 类似。将第两个 Observable 的发射放在了 GroupJoin 第三个参数中。具体看代码吧。

示例代码：

Observable<Integer> ob1 = Observable.just(1, 2);
Observable<String> ob2 = Observable.just("A", "B", "C");
ob1.groupJoin(ob2, new Function<Integer, ObservableSource<String>>() {//ob1产生结果声明周期控制函数
    @Override
    public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
        return Observable.just(integer + "00").delay(300, TimeUnit.MILLISECONDS);
    }
}, new Function<String, ObservableSource<String>>() {//ob2产生结果声明周期控制函数
    @Override
    public ObservableSource<String> apply(@NonNull String s) throws Exception {
        return Observable.just("Function1:" + s).delay(200, TimeUnit.MILLISECONDS);
    }
    //与 join 运算符的差别主要是在这里。
}, new BiFunction<Integer, Observable<String>, Observable<String>>() {
    @Override
    public Observable<String> apply(@NonNull final Integer i, @NonNull Observable<String> sob) throws Exception {
        return sob.map(new Function<String, String>() {
            @Override
            public String apply(@NonNull String s) throws Exception {
                return  "string = " + s + ", integer = " + i;
            }
        });
    }
}).subscribe(new Consumer<Observable<String>>() {
    @Override
    public void accept(@NonNull Observable<String> sob) throws Exception {
        sob.subscribe(new Consumer<String>() {
            @Override
            public void accept(@NonNull String s) throws Exception {
                Log.e(TAG, "accept : " + s);
            }
        });
    }
});

输出结果：

accept : string = A, integer = 2
accept : string = B, integer = 1
accept : string = B, integer = 2
accept : string = C, integer = 1
accept : string = C, integer = 2

4.4 Merge

将多个 Observable 合并为一个 Observable。

merge

Merge 可能会让合并的 Observables 发射的数据交错（有一个类似的操作符 Concat 不会让数据交错，它会按顺序一个接着一个发射多个 Observable）。

示例代码：

Observable<String> ob1 = Observable.interval(100, TimeUnit.MILLISECONDS)
    .map(new Function<Long, String>() {
        @Override
        public String apply(@NonNull Long aLong) throws Exception {
            //强迫症与上图一致
            return "" + ((aLong + 1) * 20);
        }
    }).take(5);
Observable<String> ob2 = Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Exception {
        Thread.sleep(350);
        emitter.onNext("1");
        Thread.sleep(250);
        emitter.onNext("1");
        emitter.onComplete();
    }
});
Observable.merge(ob1, ob2).subscribe(new Consumer<String>() {
    @Override
    public void accept(@NonNull String s) throws Exception {
        Log.e(TAG, "accept : " + s);
    }
});

输出结果：

accept : 20
accept : 40
accept : 60
accept : 1
accept : 80
accept : 100
accept : 1

4.5 StartWith

在数据序列的开头插入一条指定的项。

StartWith
如果你想要一个 Observable 在发射数据之前先发射一个指定的数据序列，可以使用 StartWith 操作符。（如果你想一个 Observable 发射的数据末尾追加一个数据序列可以使用 Concat 操作符）。
示例代码：

Observable<Integer> ob1 = Observable.range(2, 3);
ob1.startWith(4).subscribe(new Consumer<Integer>() {
    @Override
    public void accept(@NonNull Integer integer) throws Exception {
        Log.e(TAG, "accept : " + integer);
    }
});

输出结果：

accept : 4
accept : 2
accept : 3
accept : 4

4.6 Switch

将一个发射 Observables 的 Observable 转换成另一个 Observable，后者发射这些 Observables 最近发射的数据。

Switch

Switch 订阅一个发射多个 Observables 的 Observable。它每次观察那些 Observables 中的一个，Switch 返回的这个 Observable 取消订阅前一个发射数据的 Observable，开始发射最近的 Observable 发射的数据。注意：当原始 Observable 发射了一个新的 Observable 时（不是这个新的 Observable 发射了一条数据时），它将取消订阅之前的那个 Observable。这意味着，在后来那个 Observable 产生之后到它开始发射数据之前的这段时间里，前一个 Observable 发射的数据将被丢弃（就像图例上的那个黄色圆圈一样）。
示例代码：

Observable.switchOnNext(Observable.interval(100, TimeUnit.MILLISECONDS)
        .map(new Function<Long, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(@NonNull final Long aL2) throws Exception {
    //          Log.e(TAG, "al2 =" + aL2);
                return Observable.interval(30, TimeUnit.MILLISECONDS)
                .map(new Function<Long, String>() {
                    @Override
                    public String apply(@NonNull Long aL3) throws Exception {
    //                  Log.e(TAG, "al3 =" + aL3);
                        return "al2 =" + aL2 + " , al3 = " + aL3;
                    }
                });
            }
    }))
    .take(6)
    .subscribe(new Consumer<String>() {
        @Override
        public void accept(@NonNull String s) throws Exception {
            Log.e(TAG, "accept : " + s);
        }
    });

输出结果：

accept : al2 =0 , al3 = 0
accept : al2 =0 , al3 = 1
accept : al2 =0 , al3 = 2
accept : al2 =1 , al3 = 0
accept : al2 =1 , al3 = 1
accept : al2 =2 , al3 = 0

4.7 Zip

使用一个函数组合多个 Observable 发射的数据集合，然后再发射这个结果。

Zip

Zip 操作符返回一个 Obversable，它使用这个函数按顺序结合两个或多个 Observables 发射的数据项，然后它发射这个函数返回的结果。它按照严格的顺序应用这个函数。它只发射与发射数据项最少的那个 Observable 一样多的数据。

RxJava 将这个操作符实现为 zip 和 zipWith。

4.7.1 Zip

zip

示例代码：

Observable<Integer> ob1 = Observable.just(1, 2, 3, 4, 5);
Observable<String> ob2 = Observable.just("A", "B", "C", "D");
Observable.zip(ob1, ob2, new BiFunction<Integer, String, String>() {
    @Override
    public String apply(@NonNull Integer i, @NonNull String s) throws Exception {
        return i + " + " + s;
    }
}).subscribe(new Consumer<String>() {
    @Override
    public void accept(@NonNull String s) throws Exception {
        Log.e(TAG, "accept:" + s);
    }
});

输出结果：

accept:1 + A
accept:2 + B
accept:3 + C
accept:4 + D

3.7.2 ZipWith

zipWith

示例代码：

Observable<Integer> ob1 = Observable.just(1, 2, 3, 4, 5);
Observable<String> ob2 = Observable.just("A", "B", "C", "D");
ob1.zipWith(ob2, new BiFunction<Integer, String, String>() {
    @Override
    public String apply(@NonNull Integer i, @NonNull String s) throws Exception {
        return i + "-" + s;
    }
}).subscribe(new Consumer<String>() {
    @Override
    public void accept(@NonNull String s) throws Exception {
        Log.e(TAG, "accept:" + s);
    }
});

输出结果：

accept:1-A
accept:2-B
accept:3-C
accept:4-D

RxJava2.0 操作符（4）—— Combin 组合操作

4.1 And/Then/When

4.2 CombineLatest

4.3 Join

4.3.1 GroupJoin

4.4 Merge

4.5 StartWith

4.6 Switch

4.7 Zip

4.7.1 Zip

3.7.2 ZipWith

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读