RxJava进阶四（组合类操作符）

发表于 | 分类于 ， | 阅读次数
本篇文章带着大家熟悉一下RxJava的组合类操作符，本系列文章仅是带大家认识一下这些操作符的用法，并没有对操作符进行多种形态的使用，具体的还需要大家在使用时注意~

RxJava进阶

RxJava进阶一（创建类操作符）
RxJava进阶二（转换类操作符）
RxJava进阶三（过滤类操作符）
RxJava进阶四（组合类操作符）

前言

本篇文章带着大家熟悉一下RxJava的组合类操作符，本系列文章仅是带大家认识一下这些操作符的用法，并没有对操作符进行多种形态的使用，具体的还需要大家在使用时注意~

操作符总览

CombineLatest、Join、Merge、StartWith、Switch、Zip…

具体使用介绍

####CombineLatest ##

当两个Observables中的任何一个发射了一个数据时，将两个Observables数据通过指定的规则进行处理，将结果进行发射~

代码示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
Observable<Long> observable1 = Observable.interval(0, 500, TimeUnit.MILLISECONDS).take(3);
Observable<Long> observable2 = Observable.interval(500, 500, TimeUnit.MILLISECONDS).take(3);
Observable.combineLatest(observable1, observable2, new Func2<Long, Long, Long>() {
    @Override
    public Long call(Long along1, Long along2) {
        System.out.println("along1 --> " + along1);
        System.out.println("along2 --> " + along2);
        return along1 + along2;
    }
}).subscribe(new Action1<Long>() {
    @Override
    public void call(Long aLong) {
        System.out.println("result = " + aLong);
        System.out.println("--------------");
    }
});

运行结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
10-08 17:00:12.319 6650-6698/com.shenghan.haobaobei I/System.out: along1 --> 1
10-08 17:00:12.319 6650-6698/com.shenghan.haobaobei I/System.out: along2 --> 0
10-08 17:00:12.319 6650-6698/com.shenghan.haobaobei I/System.out: result = 1
10-08 17:00:12.319 6650-6698/com.shenghan.haobaobei I/System.out: --------------
10-08 17:00:12.816 6650-6728/com.shenghan.haobaobei I/System.out: along1 --> 2
10-08 17:00:12.816 6650-6728/com.shenghan.haobaobei I/System.out: along2 --> 0
10-08 17:00:12.816 6650-6728/com.shenghan.haobaobei I/System.out: result = 2
10-08 17:00:12.817 6650-6728/com.shenghan.haobaobei I/System.out: --------------
10-08 17:00:12.818 6650-6698/com.shenghan.haobaobei I/System.out: along1 --> 2
10-08 17:00:12.820 6650-6698/com.shenghan.haobaobei I/System.out: along2 --> 1
10-08 17:00:12.821 6650-6698/com.shenghan.haobaobei I/System.out: result = 3
10-08 17:00:12.821 6650-6698/com.shenghan.haobaobei I/System.out: --------------
10-08 17:00:13.315 6650-6698/com.shenghan.haobaobei I/System.out: along1 --> 2
10-08 17:00:13.315 6650-6698/com.shenghan.haobaobei I/System.out: along2 --> 2
10-08 17:00:13.315 6650-6698/com.shenghan.haobaobei I/System.out: result = 4
10-08 17:00:13.315 6650-6698/com.shenghan.haobaobei I/System.out: --------------

结论：

1.只有当两个observable都发射过第一项数据时，才能进行组合发射

从log日志来看，observable1打印的第一项数据为1，所以可以推断observable1发射的0数据时，observable2并没有发射过数据，所以并没有进行组合。

2.当observable1与observable2都发射首个数据后，在此发射任何数据都会找相应的另外一个observable的最新数据进行组合发射

Join

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
Observable<Long> observable1 = Observable.interval(2000, 1000, TimeUnit.MILLISECONDS).take(3);
Observable<Long> observable2 = Observable.interval(2500, 1000, TimeUnit.MILLISECONDS).take(3);
observable1
        .join(
                observable2,
                new Func1<Long, Observable<Long>>() {
                    @Override
                    public Observable<Long> call(Long aLong) {
                        //使Observable延迟500毫秒执行
                        System.out.println("observable1 -- >" + aLong);
                        return Observable.just(aLong).delay(500, TimeUnit.MILLISECONDS);
                    }
                },
                new Func1<Long, Observable<Long>>() {
                    @Override
                    public Observable<Long> call(Long aLong) {
                        System.out.println("observable2 -- >" + aLong);
                        return Observable.just(aLong).delay(500, TimeUnit.MILLISECONDS);
                    }
                },
                new Func2<Long, Long, Long>() {
                    @Override
                    public Long call(Long aLong1, Long aLong2) {
                        System.out.println("aLong1 = " + aLong1);
                        System.out.println("aLong2 = " + aLong2);
                        return aLong1 + aLong2;
                    }
                })
        .subscribe(new Action1<Long>() {
            @Override
            public void call(Long aLong) {
                System.out.println("result = " + aLong);
                System.out.println("--------------");
            }
        });

运行结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
10-08 16:47:56.988 20959-21011/com.shenghan.haobaobei I/System.out: observable1 -- >0
10-08 16:47:57.488 20959-21041/com.shenghan.haobaobei I/System.out: observable2 -- >0
10-08 16:47:57.488 20959-21041/com.shenghan.haobaobei I/System.out: aLong1 = 0
10-08 16:47:57.488 20959-21041/com.shenghan.haobaobei I/System.out: aLong2 = 0
10-08 16:47:57.488 20959-21041/com.shenghan.haobaobei I/System.out: result = 0
10-08 16:47:57.488 20959-21041/com.shenghan.haobaobei I/System.out: --------------
10-08 16:47:57.991 20959-21011/com.shenghan.haobaobei I/System.out: observable1 -- >1
10-08 16:47:57.991 20959-21011/com.shenghan.haobaobei I/System.out: aLong1 = 1
10-08 16:47:57.991 20959-21011/com.shenghan.haobaobei I/System.out: aLong2 = 0
10-08 16:47:57.991 20959-21011/com.shenghan.haobaobei I/System.out: result = 1
10-08 16:47:57.991 20959-21011/com.shenghan.haobaobei I/System.out: --------------
10-08 16:47:58.490 20959-21041/com.shenghan.haobaobei I/System.out: observable2 -- >1
10-08 16:47:58.490 20959-21041/com.shenghan.haobaobei I/System.out: aLong1 = 1
10-08 16:47:58.490 20959-21041/com.shenghan.haobaobei I/System.out: aLong2 = 1
10-08 16:47:58.490 20959-21041/com.shenghan.haobaobei I/System.out: result = 2
10-08 16:47:58.490 20959-21041/com.shenghan.haobaobei I/System.out: --------------
10-08 16:47:58.990 20959-21011/com.shenghan.haobaobei I/System.out: observable1 -- >2
10-08 16:47:58.990 20959-21011/com.shenghan.haobaobei I/System.out: aLong1 = 2
10-08 16:47:58.990 20959-21011/com.shenghan.haobaobei I/System.out: aLong2 = 1
10-08 16:47:58.990 20959-21011/com.shenghan.haobaobei I/System.out: result = 3
10-08 16:47:58.990 20959-21011/com.shenghan.haobaobei I/System.out: --------------
10-08 16:47:59.489 20959-21041/com.shenghan.haobaobei I/System.out: observable2 -- >2
10-08 16:47:59.489 20959-21041/com.shenghan.haobaobei I/System.out: aLong1 = 2
10-08 16:47:59.489 20959-21041/com.shenghan.haobaobei I/System.out: aLong2 = 2
10-08 16:47:59.489 20959-21041/com.shenghan.haobaobei I/System.out: result = 4
10-08 16:47:59.489 20959-21041/com.shenghan.haobaobei I/System.out: --------------

结论：

1.只有当两个observable都发射过第一项数据时，才能进行组合发射

从log日志来看，observable1发射数据0时，组合的call方法并没有被执行，而当observable2发射数据0后，组合call方法被回调了。

2.当observable1与observable2都发射首个数据后，在此发射任何数据都会找相应的另外一个observable的最新数据进行组合发射

####Merge##

将两个Observable发射的数据按照时间顺序进行组合，合并成一个Observable进行发射~

代码示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
Observable<String> observable1 = Observable.interval(0, 500, TimeUnit.MILLISECONDS).take(3).flatMap(new Func1<Long, Observable<String>>() {
    @Override
    public Observable<String> call(Long aLong) {
        return Observable.just("observable1 -- >" + aLong);
    }
});
Observable<String> observable2 = Observable.interval(500, 500, TimeUnit.MILLISECONDS).take(3).flatMap(new Func1<Long, Observable<String>>() {
    @Override
    public Observable<String> call(Long aLong) {
        return Observable.just("observable2 -- >" + aLong);
    }
});
Observable
        .merge(observable1, observable2)
        .subscribe(new Action1<String>() {
            @Override
            public void call(String aString) {
                System.out.println("result = " + aString);
            }
        });

运行结果：

1
2
3
4
5
6
10-08 17:11:33.696 16661-16737/com.shenghan.haobaobei I/System.out: result = observable1 -- >0
10-08 17:11:34.194 16661-16737/com.shenghan.haobaobei I/System.out: result = observable1 -- >1
10-08 17:11:34.198 16661-16711/com.shenghan.haobaobei I/System.out: result = observable2 -- >0
10-08 17:11:34.695 16661-16737/com.shenghan.haobaobei I/System.out: result = observable1 -- >2
10-08 17:11:34.698 16661-16711/com.shenghan.haobaobei I/System.out: result = observable2 -- >1
10-08 17:11:35.196 16661-16711/com.shenghan.haobaobei I/System.out: result = observable2 -- >2

####StartWith##

在源Observable发射数据之前，先发射一个指定的数据序列或数据项~

代码示例：

1
2
3
4
5
6
7
8
9
Observable
        .just(4, 5, 6)
        .startWith(1, 2, 3)
        .subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer integer) {
                System.out.println("result = " + integer);
            }
        });

运行结果：

1
2
3
4
5
6
10-08 17:16:43.077 21830-21830/com.shenghan.haobaobei I/System.out: result = 1
10-08 17:16:43.078 21830-21830/com.shenghan.haobaobei I/System.out: result = 2
10-08 17:16:43.078 21830-21830/com.shenghan.haobaobei I/System.out: result = 3
10-08 17:16:43.078 21830-21830/com.shenghan.haobaobei I/System.out: result = 4
10-08 17:16:43.078 21830-21830/com.shenghan.haobaobei I/System.out: result = 5
10-08 17:16:43.078 21830-21830/com.shenghan.haobaobei I/System.out: result = 6

####switchOnNext##

把一组Observable转换成一个Observable，如果在同一个时间内产生两个或多个Observable产生的数据，只发射最后一个Observable产生的数据~

示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
Observable<Observable<String>> observable = Observable.interval(2000, 500, TimeUnit.MILLISECONDS).map(new Func1<Long, Observable<String>>() {
    @Override
    public Observable<String> call(Long aLongOutside) {
        //每隔250毫秒产生一组数据（0,1,2,3,4)
        return Observable.interval(0, 250, TimeUnit.MILLISECONDS).map(new Func1<Long, String>() {
            @Override
            public String call(Long aLongInside) {
                return "aLongOutside = " + aLongOutside + "| aLongInside = " + aLongInside;
            }
        }).take(5);
    }
}).take(2);
Observable.switchOnNext(observable).subscribe(new Action1<String>() {
    @Override
    public void call(String s) {
        System.out.println("result = " + s);
    }
});

运行结果：

1
2
3
4
5
6
7
10-08 18:20:14.605 2921-2993/com.shenghan.haobaobei I/System.out: result = aLongOutside = 0| aLongInside = 0
10-08 18:20:14.857 2921-2993/com.shenghan.haobaobei I/System.out: result = aLongOutside = 0| aLongInside = 1
10-08 18:20:15.103 2921-3055/com.shenghan.haobaobei I/System.out: result = aLongOutside = 1| aLongInside = 0
10-08 18:20:15.353 2921-3055/com.shenghan.haobaobei I/System.out: result = aLongOutside = 1| aLongInside = 1
10-08 18:20:15.604 2921-3055/com.shenghan.haobaobei I/System.out: result = aLongOutside = 1| aLongInside = 2
10-08 18:20:15.853 2921-3055/com.shenghan.haobaobei I/System.out: result = aLongOutside = 1| aLongInside = 3
10-08 18:20:16.104 2921-3055/com.shenghan.haobaobei I/System.out: result = aLongOutside = 1| aLongInside = 4

####Zip##

使用一个指定的函数将多个Observable发射的数据组合在一起，然后将这个函数的结果作为单项数据发射，严格周期顺序进行合并，不能单独发射~

示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
Observable<String> observable1 = Observable.just("walid - 1", "walid - 2", "walid - 3");
Observable<String> observable2 = Observable.just("Jordan - 1", "Jordan - 2", "Jordan - 3", "Jordan - 4");
Observable.zip(observable1, observable2, new Func2<String, String, String>() {
    @Override
    public String call(String s1, String s2) {
        return s1 + " | " + s2;
    }
}).subscribe(new Subscriber<String>() {
    @Override
    public void onCompleted() {
        System.out.println("onCompleted");
    }
    @Override
    public void onError(Throwable e) {
        System.err.println("onError");
    }
    @Override
    public void onNext(String value) {
        System.out.println("onNext --> " + value);
    }
});

运行结果：

1
2
3
4
10-08 18:53:31.772 24793-24793/com.shenghan.haobaobei I/System.out: onNext --> walid - 1 | Jordan - 1
10-08 18:53:31.772 24793-24793/com.shenghan.haobaobei I/System.out: onNext --> walid - 2 | Jordan - 2
10-08 18:53:31.772 24793-24793/com.shenghan.haobaobei I/System.out: onNext --> walid - 3 | Jordan - 3
10-08 18:53:31.772 24793-24793/com.shenghan.haobaobei I/System.out: onCompleted

结论：

1.只有每个observable都依次产品一轮数据时，才会统一发射一次，当不会有完整一轮数据时，视为完成

从log日志来看，observable1产生的数据是与observable2一一对应的，也就是只有observable1与observable2同时产生数据时才会统一发射一次~

2.当不满足所有observable都有数据可产品时，视为完成状态

从log日志来看，observable2最后一项数据“Jordan - 4”并没有打印，原因是observable1并没有数据可以产生了，所以不满足发射条件，视为完成状态~

结语

组合类操作符，就简单介绍到这里，希望能够对同学有所帮助，谢谢~