RxJava2最全面、最详细的讲解（二）

阳光穿透心脏的1/2处 2023-05-28 14:23 60阅读 0赞

# 一、概述 #

RxJava2在第一篇文章中基本用法作了详细的介绍，是一个基于事件流的异步操作库。相信大家对RxJava有了一定的理解，由于篇幅过长所以重新写了一篇，如果不了解Rxjava2可以参考下[RxJava2最全面、最详细的讲解(一)][RxJava2]。下面开始继续讲解RxJava2的其他用法。(源码和其他链接在文章最后给出)

在使用前记得在build.gradle文件中添加依赖：

implementation 'io.reactivex.rxjava2:rxjava:2.0.4'
        implementation 'io.reactivex.rxjava2:rxandroid:2.0.1'

注意：RxJava1与RxJava2依赖不能共存。不能同时添加RxJava1与RxJava2的依赖，只能添加其中一个。

RxJava的操作符就是java中的用法，详尽的操作符请看官网[ReactiveX的官方网站][ReactiveX]，这里列举了一部分：

> Creating Observables(Observable的创建操作符)，比如：Observable.create()、Observable.just()、Observable.from()等等；  
> Transforming Observables(Observable的转换操作符)，比如：observable.map()、observable.flatMap()、observable.buffer()等等；  
> Filtering Observables(Observable的过滤操作符)，比如：observable.filter()、observable.sample()、observable.take()等等；  
> Combining Observables(Observable的组合操作符)，比如：observable.join()、observable.merge()、observable.combineLatest()等等；  
> Error Handling Operators(Observable的错误处理操作符)，比如:observable.onErrorResumeNext()、observable.retry()等等；  
> Observable Utility Operators(Observable的功能性操作符)，比如：observable.subscribeOn()、observable.observeOn()、observable.delay()等等；  
> Conditional and Boolean Operators(Observable的条件操作符)，比如：observable.amb()、observable.contains()、observable.skipUntil()等等；  
> Mathematical and Aggregate Operators(Observable数学运算及聚合操作符)，比如：observable.count()、observable.reduce()、observable.concat()等等；  
> 其他如observable.toList()、observable.connect()、observable.publish()等等；

# 二、变换操作符 #

变换操作符的主要是对事件序列中的事件进行处理变换，使其转变成不同的事件，再加以处理。这里列举几种常用的操作符

### 1.Map() ###

map操作符把被观察者Observable产生的结果通过映射规则转换成另一种结果集，并交给订阅者处理。简单来说就是对被观察者发送的每个事件都通过指定函数的处理，从而转变成另一种事件。

//链式编程
    Observable.just(1, 2, 3, 4, 5)
               //使用Map操作符中的Function函数对被观察者发送的事件统一作出处理
               .map(new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer integer) throws Exception {
                //对被观察者just发送的结果，都全部乘以10处理
                return integer * 10;
            }
        }).subscribe(new Consumer<Integer>() {//订阅
            @Override
            public void accept(Integer integer) throws Exception {//接受事件结果，是处理后的结果
                Log.e(TAG, "map：accept == " + integer);
            }
        });

上面的例子中我们使用just()操作符连续发送1,2,3,4,5等5个事件，通过Map操作符中的Function函数对被观察者发送的事件统一作出乘以10处理，订阅后输出最终结果：

![2019101814113967.png][]

### 2.flatMap() ###

flatMap()操作符是将Observable(被观察者)产生的结果拆分和单独转换变成多个Observable，然后把多个Observable“扁平化”整合成新的一个Observable，并依次提交产生的结果给订阅者。

大意为：flatMap()通过传入一个函数作为参数转换源Observable(被观察者)，在这个函数中你可以自定义转换规则，最后在这个函数中返回一个新的Observable，然后flatMap函数回调方法通过合并这些Observable成一个新的Observable，发送给Observer(观察者)。（理论比较抽象，我们来看例子）

Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) {
                e.onNext("A");
                e.onNext("B");
                e.onComplete();
            }
        }).flatMap(new Function<String, ObservableSource<String>>() {
            @Override//通过flatMap将被观察者生产的事件进行拆分，再将新的事件转换成一个新的Observable发送
            public ObservableSource<String> apply(String s) {
                List<String> list = new ArrayList<>();
                Log.e(TAG, "flatMap：apply == 事件" + s);
                //将一个事件拆分成两个子事件，例如将A事件拆分成A0,A1两个事件，然后再整个合成一个Observable通过fromIterable发送给订阅者
                for (int j = 0; j < 2; j++) {
                    list.add("拆分后的子事件" + s + j);
                }
                return Observable.fromIterable(list);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) {
                Log.e(TAG, "flatMap：accept == " + s);
            }
        });

上面我们发送了A、B两个事件，在flatMap中接收到A事件后，将A事件拆分成A0，A1两个事件，再通过fromIterable(list)整合成一个Observable发送给订阅者。打印log如下：

![20191018153010688.png][]

**注意：flatMap在合并Observable结果时，可能存在交叉的情况出现，即新合并生成的序列可能是无序的。（上面的例子我演示多次并没有出现乱序）**

### 3.concatMap() ###

concatMap()和flatMap()方法类似，都是把Observable(被观察者)产生的事件转换成多个Observable(被观察者)，然后把Observable整合成一个Observable，并依次提交生产的结果给订阅者。

与flatMap()不同的是，concatMap()在处理产生的Observable时，采用的是“连接concat”的方式，而不是“marge合并”的方式，这样就保证了产生事件的顺序性，**也就是说提交给订阅者的结果是按照顺序提交的，不会存在交叉的可能性。**

Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) {
                e.onNext("A");
                e.onNext("B");
                e.onComplete();
            }
        }).concatMap(new Function<String, ObservableSource<String>>() {
            @Override//通过concatMap（有序）将被观察者生产的事件进行拆分，再将新的事件转换成一个新的Observable发送
            public ObservableSource<String> apply(String s) {
                List<String> strings = new ArrayList<>();
                Log.e(TAG, "concatMap：apply == 事件" + s);
                for (int j = 0; j < 2; j++) {
                    strings.add("拆分后的子事件" + s + j);
                }
                return Observable.fromIterable(strings);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) {
                Log.e(TAG, "concatMap：accept == " + s);
            }
        });

打印的log如下：

![20191018155508256.png][]

### 4.buffer() ###

buffer()操作符将Observable(被观察者)需要发送的事件周期性收集到列表中，并把这列表提交给Observer(观察者)，观察者处理后，清空buffer列表，同时接收下一次的结果交给订阅者，周而复始。

需要注意的是：一但初始的Observable在产生事件中出现异常，即使buffer()收集到已经存在的结果，订阅者也会马上收到这个异常，并结束整个过程。

Observable.just("A", "B", "C", "D", "E")//这里演示发送5个事件
             .buffer(3, 2)//缓存列表数量为3个，步长为2
                    .subscribe(new Observer<List<String>>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "buffer：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(List<String> strings) {
                Log.e(TAG, "buffer：onNext == 缓存事件数：" + strings.size());
                for (int j = 0; j < strings.size(); j++) {
                    Log.e(TAG, "buffer：子事件==" + strings.get(j));
                }
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "buffer：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "buffer：onComplete == ");
            }
        });

上述例子设置了"ABCDE"5个事件，缓存事件数为3个，步长为2。在被观察者的事件中，第一次获取3个事件数放到缓存区域，即"A"，"B"，"C"，发送给订阅者；第二次获取时，因为步长=2，所以事件往后移动2个，即指针往后移动2位，从"C"开始取三个事件，即"C"，"D"，"E"发送给订阅者；第三次获取时，事件再在第二次的基础上往后移动2个，即到了"E"，取3个事件，事件不足3个，只能够取"E"发送给订阅者。看图比较好理解，我们来看看buffer()的原理图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3Nzk2Njgz_size_16_color_FFFFFF_t_70][]

打印结果如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3Nzk2Njgz_size_16_color_FFFFFF_t_70 1][]

这里先对变换操作符做出表格总结：

<table> 
 <tbody> 
  <tr> 
   <th colspan="2">创建类型</th> 
   <th>作用</th> 
   <th>使用场景</th> 
  </tr> 
  <tr> 
   <td>变换操作符</td> 
   <td>map()</td> 
   <td>对Observable发送的每一个事件进行转换</td> 
   <td colspan="1"> <p>数据类型、事件等需要转换</p> </td> 
  </tr> 
  <tr> 
   <td>flatMap()</td> 
   <td>对Observable发送的整个个事件进行转换（无序）</td> 
  </tr> 
  <tr> 
   <td>concatMap()</td> 
   <td>对Observable发送的整个个事件进行转换（有序）</td> 
  </tr> 
  <tr> 
   <td>buffer()</td> 
   <td>从Observable的事件中获取事件放到缓存区再发送事件</td> 
  </tr> 
 </tbody> 
</table>

# 三、组合操作符 #

### 1.concat()与concatArray() ###

concat()与concatArray()都是组合多个被观察者的一起发送数据，合并后安先后顺序执行。

二者的区别是：concat()的观察者数量最多是4个，而concatArray()的个数没有限制

> //concat的构造方法  
> concat(source1)；  
> concat(source1, source2)；  
> concat(source1, source2, source3)；  
> concat(source1, source2, source3, source4)；

//concatArray的构造方法  
concatArray(ObservableSource<? extends T>... sources)；

我们来看看简单的运用：

Observable.concat(Observable.just(1, 2), Observable.just(3, 4), Observable.just(5, 6), Observable.just(7, 8))
                .subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "concat：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "concat：onNext ==" + integer);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "concat：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "concat：onComplete == ");
            }
        });
     
      Observable.concatArray(Observable.just("一", "二"), Observable.just("三", "四"),
                    Observable.just("五", "六"), Observable.just("七", "八"), Observable.just("九", "十"))
                    .subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "concatArray：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(String s) {
                Log.e(TAG, "concatArray：onNext ==" + s);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "concatArray：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "concatArray：onComplete == ");
            }
        });

上面concat()创建了四个被观察者，发送1,2,3,4,5,6,7,8个事件，concatArray()创建了五个被观察者，发送一,二,三,四,五,六,七,八,九,十，打印log如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3Nzk2Njgz_size_16_color_FFFFFF_t_70 2][]

### 2.Merge()与MergeArray() ###

Merge()与MergeArray()组合多个被观察者发送数据，按照时间先后顺序来执行。与concat()与concatArray()组合类似

Merge()与MergeArray()的区别：Merge()只能发送最多4个Observable(被观察者)，MergeArray()的数量没有限制。MergeArray()这里就不演示例子了。

> Megre()的构造方法：

merge(source1)；  
merge(source1, source2)；  
merge(source1, source2,  source3)；  
merge(source1, source2,  source3,  source4)；

MergeArray()的构造方法：

mergeArray(ObservableSource<? extends T>... sources)；

//起始值为1，发送3个事件，第一个事件延迟1秒发送，事件间隔为1秒
        Observable<Long> observable1 = Observable.intervalRange(1, 3, 1, 1, TimeUnit.SECONDS);
        //起始值为10，发送3个事件，第一个事件延迟1秒发送，事件间隔为1秒
        Observable<Long> observable2 = Observable.intervalRange(10, 3, 1, 1, TimeUnit.SECONDS);
     
            Observable.merge(observable1, observable2)
                .subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "concatArray：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Long aLong) {
                Log.e(TAG, "concatArray：onNext ==" + aLong);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "concatArray：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "concatArray：onComplete == ");
            }
        });

这里我们创建了两个被观察者，一个起始值为1，延迟1秒发送3个事件，另一个为起始值为10，延迟1秒发送3个事件，按照Observable提交结果的时间顺序来执行，对Observable进行合并，Observable1每隔1秒产生数据为：1,2,3，Observable2每隔1秒产生的数据为：10,11,12，都延迟1秒产生，最后合并的结果为：1,10,2,11,3,12，MergeArray()同理，打印log如下：

![20191022164841236.gif][]

可以看出都是按照时间的先后顺序来执行被观察者发送事件。

### 3.concatDelayError() ###

concat()与concatArray()组合中，如果某一个Observable(被观察者)发出onError()事件，则会马上停止其他事件的发送。如果需要onError()事件推迟到其他事件发送完成才出发的话则需要用到concatDelayError()方法；

我们来看看concat()方法中途抛出异常的例子：

Observable<String> observable1 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("第一个事件");
                e.onNext("第二个事件");
                e.onError(new Exception("中途抛出异常"));
                e.onComplete();
            }
        });
        Observable<String> observable2 = Observable.just("第三个事件");
     
        //中途抛出异常列子：
        Observable.concat(observable1, observable2).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "concatDelayError：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(String s) {
                Log.e(TAG, "concatDelayError：onNext ==" + s);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "concatDelayError：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "concatDelayError：onComplete == ");
            }
        });

这里我们依次发送第一个事件，第二个事件，中途抛出异常，调用onComplete()，第三个事件，打印log如下：

# ![20191021160639196.png][] #

可以看到，调用onError()方法以后，其他事件概不执行了，那么我们来看看解决办法，改用concatDelayError()：

Observable<String> observable1 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("第一个事件");
                e.onNext("第二个事件");
                e.onError(new Exception("中途抛出异常"));
                e.onComplete();
            }
        });
        Observable<String> observable2 = Observable.just("第三个事件");
     
        //中途抛出异常列子：
        Observable.concatArrayDelayError(observable1, observable2).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "concatDelayError：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(String s) {
                Log.e(TAG, "concatDelayError：onNext ==" + s);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "concatDelayError：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "concatDelayError：onComplete == ");
            }
        });

会把错误在所有结果合并完后才执行，打印的数据如下：

![20191021162231576.png][]

### 4.megreDelayError() ###

megreDelayError()方法同上，meger()合并的某个Observable(被观察者)出现异常，会马上停止合并，并对订阅者回到onError()方法，而megreDelayError()会把错误在所有结果合并完后才执行：

//创建被观察者-异常
        Observable<Object> errorObservable = Observable.error(new Exception("抛出异常"));
     
        //产生0,2,4的事件序列，每隔1秒发送事件，一共发送3次
        Observable<Object> observable1 = Observable.interval(0, 1, TimeUnit.SECONDS)
                .map(new Function<Long, Object>() {
                    @Override
                    public Object apply(Long aLong) throws Exception {
                        return aLong * 2;
                    }
                }).take(3)/*.mergeWith(errorObservable.delay(4, TimeUnit.SECONDS))*/;
     
        //产生0,10,20的事件序列，每隔1秒发送事件，一共发送3次，
        Observable<Long> observable2 = Observable.interval(1, 1, TimeUnit.SECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        return aLong * 10;
                    }
                }).take(3);
     
        Observable.mergeDelayError(observable1, errorObservable, observable2)
                .subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "mergeDelayError：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Object o) {
                Log.e(TAG, "mergeDelayError：onNext ==" + o);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "mergeDelayError：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "mergeDelayError：onComplete == ");
            }
        });

打印效果如下：

![20191021173555376.gif][]

### 5.startWith()与startWithArray() ###

startWith()与startWithArray()操作符都是在源Observable(被观察者)提交结果之前，插入指定的某些数据，注意调用顺序：先调用最后加入的数据。

> startWith()的构造方法：

//传入单个数据  
startWith(T item)  
//传入集合，可将数据加入一个集合中  
startWith(Iterable<? extends T> items)

startWithArray()的构造方法：

//可传入多个参数，可变长参数  
startWithArray(T... items)

ArrayList<String> strings = new ArrayList<>();
            strings.add("Array:1");
            Observable.just("一", "二", "三", "四")
                    .startWith(strings)//插入单个集合
                    .startWith("startWith:2")//插入单个数据
                    .startWithArray("startWithArray:3", "startWithArray:4")//插入多个数据
                    .subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "startWith：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(String s) {
                Log.e(TAG, "startWith：onNext 结果== " + s);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "startWith：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "startWith：onComplete == ");
            }
        });

上面例子中，在just()的前面依次加入的是：集合\[Array:1\]，“startWith:2”，"startWithArray:3", "startWithArray:4"，但是打印出来的结果顺序却刚好相反，说明Observable优先发送最后插入的数据，log如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3Nzk2Njgz_size_16_color_FFFFFF_t_70 3][]

组合操作符简单总结一下：

<table> 
 <tbody> 
  <tr> 
   <th colspan="2">创建类型</th> 
   <th>作用</th> 
   <th>备注</th> 
   <th>使用场景</th> 
  </tr> 
  <tr> 
   <td>组合操作符</td> 
   <td>concat()与concatArray()</td> 
   <td>组合多个被观察者发送数据，合并后按照顺序执行</td> 
   <td> <p>区别：组合被观察者数量：</p> <p>concat()&lt;=4,concatArray()无限制</p> </td> 
   <td colspan="1"> <p>合并数据源</p> <p>联合处理</p> </td> 
  </tr> 
  <tr> 
   <td>Merge()与MergeArray()</td> 
   <td>组合多个被观察者发送数据，合并后按照时间顺序执行</td> 
   <td> <p>区别：组合被观察者数量：</p> <p>Merge()&lt;=4,MergeArray()无限制</p> </td> 
  </tr> 
  <tr> 
   <td>concatDelayError()</td> 
   <td colspan="1">将错误事件延迟到所以事件执行完后才执行错误事件</td> 
   <td colspan="1">&nbsp;</td> 
  </tr> 
  <tr> 
   <td>megreDelayError()</td> 
  </tr> 
  <tr> 
   <td>startWith()与startWithArray()</td> 
   <td>在一个被观察者发送数据前，追加发送一些数据</td> 
   <td>调用顺序：先追加的后调用，后追加的先调用</td> 
  </tr> 
 </tbody> 
</table>

# 四、合并操作符 #

### 1.zip() ###

zip()是把两个Observable(被观察者)提交的结果，严格按照顺序对位进行合并，最后发送给订阅者，最终合并的数量等于多个Observable数量最少的数量。

注意：下面的Observable1中的4并没有事件与其合并打印出来，但是事件还是会发送的。

//设置需要传入的被观察者数据
        Observable<Integer> observable1 = Observable.just(1, 2, 3, 4);
        Observable<String> observable2 = Observable.just("A", "B", "C");
     
        //回调apply()方法，并在里面自定义合并结果的逻辑
        // BiFunction<Integer, String, String>，第一个类型为observable1的参数类型，第二个类型为observable2的参数类型，第三个为合并后的参数类型
        Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
            @Override
            public String apply(Integer integer, String str) throws Exception {
                return integer + str;
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "zip：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(String s) {
                Log.e(TAG, "zip：onNext == " + s);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "zip：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "zip：onComplete == ");
            }
        });

上面根据顺序的对位合并数据为：1A,2B,3C，4没有与其合并的事件，打印log如下：

![20191021181306111.png][]

### 2.combineLatest() ###

combineLatest()是将两个Observable(被观察者)产生的结果进行合并，合并的结果将组成一个新的Observable发送给订阅者。这两个Observable中任意的一个Observable产生的结果，都与另一个Observable最后的结果，按照一定的规则进行合并。

与zip()类似，区别在于zip()根据个数1对1合并，combineLatest()根据时间点一对一合并。

//产生0,10,20的事件序列，每隔1秒发送事件，一共发送3次
        Observable<Long> observable1 = Observable.interval(0, 1, TimeUnit.SECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        return aLong * 10;
                    }
                }).take(3);
     
        //产生0,1,2,3,4的事件序列，起始值为0，一共发送4次，延迟1秒后开始发送，每隔1秒发送事件
        Observable<Long> observable2 = Observable.intervalRange(0, 4, 1, 1, TimeUnit.SECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        return aLong * 1;
                    }
                });
     
        Observable.combineLatest(observable1, observable2, new BiFunction<Long, Long, Long>() {
            @Override
            public Long apply(Long o1, Long o2) throws Exception {
                Log.e(TAG, "combineLatest：apply: o1+o2：" + o1 + "+" + o2);
                //observable1的最后的一个数据都与observable2的每一个数据相加
                return o1 + o2;
            }
        }).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "combineLatest：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Long aLong) {
                Log.e(TAG, "combineLatest：onNext 合并的结果== " + aLong);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "combineLatest：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "combineLatest：onComplete == ");
            }
        });

Observable1的事件序列为0,10,20；Observable2的事件序列为0,1,2,3, Observable2因为延迟了1秒发送，所以在Observable1执行1秒后即Observable1=10时，Observable2才开始执行事件，否则在Observable2未开始执行事件前，Observable2的默认值与Observable1合并事件，如果Observable1执行完事件后，以Observable1最后的值为准Observable1=20，那么合并出来的数据为：10+0,20+0,20+1,20+2,20+3，打印log如下：

![20191021190904972.gif][]

### 3.combineLatestDelayError() ###

combineLatest()合并过程中一但出现异常就会立即停止后续合并事件，并且回调onError()方法，combineLatestDelayError()是把错误放到所有结果都合并完成之后才执行。

combineLatestDelayError()类似concatDelayError()、mergeDelayError()，这里就不在举例子了。

### 4.reduce() ###

reduce()把观察者需要发送的事件聚合成一个事件并且发送。本质上都是前两个数据聚合，再与后一个数据聚合，依次类推。

Observable.just(1, 2, 3, 4, 5).
        reduce(new BiFunction<Integer, Integer, Integer>() {
            @Override
            public Integer apply(Integer integer, Integer integer2) throws Exception {
                Log.e(TAG, "reduce：accept 计算结果== " + integer + "*" + integer2);
                //按先后顺序，两个事件聚合处理后 ，将结果再与下一事件聚合处理，依次类推
                return integer * integer2;
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "reduce：accept 合并的结果== " + integer);
            }
        });

比如上面的1, 2, 3, 4, 5共5个事件，1与2聚合后生成新的Observable(被观察者)，新的Observable与下一个数3再聚合，依次类推，所以数据聚合完成，打印数据如下：

![20191022104858288.png][]

### 5.collect() ###

将Observable(被观察者)发送的数据收集到一个容器中（数据结构）。Observable将事件队列分别装入一个准备好的容器中，然后在新的Observable将该容器发送给订阅者。

//第一个参数：声明容器的类型，第二个参数：处理数据的逻辑，加入容器中
       Observable.just(1, 2, 3, 4, 5).collect(new Callable<ArrayList<Integer>>() {
            @Override
            public ArrayList<Integer> call() throws Exception {
                return new ArrayList<>();
            }
        }, new BiConsumer<ArrayList<Integer>, Integer>() {
            @Override
            public void accept(ArrayList<Integer> list, Integer integer) throws Exception {
                Log.e(TAG, "reduce：collect 加入容器的数据== " + integer);
                list.add(integer);
            }
        }).subscribe(new Consumer<ArrayList<Integer>>() {
            @Override
            public void accept(ArrayList<Integer> integers) throws Exception {
                Log.e(TAG, "reduce：collect 最后结果== " + integers);
            }
        });

Callable方法声明一个容器，Callable实现的方法将数据加入容器中，打印数据如下：

![20191022122139759.png][]

合并操作符的简单总结：

<table> 
 <tbody> 
  <tr> 
   <th colspan="2">创建类型</th> 
   <th>作用</th> 
   <th>备注</th> 
   <th>使用场景</th> 
  </tr> 
  <tr> 
   <td>合并操作符</td> 
   <td>zip()</td> 
   <td>合并多个被观察者发送的事件，生成一个新的事件序列。</td> 
   <td> <p>严格按照事件序列进行合并</p> <p>最后Observable的数据=多个Observable中数量最少的</p> </td> 
   <td colspan="1"> <p>合并数据源</p> <p>联合处理</p> </td> 
  </tr> 
  <tr> 
   <td>combineLatest()</td> 
   <td>将两个Observable产生的结果进行合并，合并新的Observable发送给订阅者</td> 
   <td>当其中一个Observable发送数据，都与另一个Observable最后发送的数据结合，类似zip()，不同的是按照时间点合并</td> 
  </tr> 
  <tr> 
   <td>combineLatestDelayError()</td> 
   <td colspan="1">将错误事件延迟到所以事件执行完后才执行错误事件</td> 
   <td colspan="1">&nbsp;</td> 
  </tr> 
  <tr> 
   <td>reduce()</td> 
   <td colspan="1">把观察者需要发送的事件聚合成一个事件并且发送。</td> 
   <td colspan="1">前两个被观察者聚合成新的被观察者，再与下一数据聚合，一次类推</td> 
  </tr> 
  <tr> 
   <td>collect()</td> 
   <td>将被观察者发送的数据放到一个容器中处理</td> 
   <td>&nbsp;</td> 
  </tr> 
 </tbody> 
</table>

# 五、统计发送事件的数量 #

### 1.count() ###

count()方法是统计Observable(被观察者)发送的事件的数量。

Observable.just(1, 2, 3, 4, 5)
                 .count()
                 .subscribe(new Consumer<Long>() {
            @Override
            public void accept(Long l) throws Exception {
                Log.e(TAG, "reduce：accept 事件数== " + l);
            }
        });

这个方法比较简单，just()发送了1, 2, 3, 4, 5共5个事件，打印log如下：

![20191022122200831.png][]

> **源码地址：**[https://github.com/FollowExcellence/Rxjava\_Retrofit][https_github.com_FollowExcellence_Rxjava_Retrofit]

[RxJava2]: https://blog.csdn.net/m0_37796683/article/details/102525484
[ReactiveX]: http://reactivex.io/documentation/operators.html
[2019101814113967.png]: /images/20230528/391a154f4d0544f38fa4d3f76b368a52.png
[20191018153010688.png]: /images/20230528/4d000b728e464911af232040482e4de6.png
[20191018155508256.png]: /images/20230528/78e33aba592544989dea3e7174e6baee.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3Nzk2Njgz_size_16_color_FFFFFF_t_70]: /images/20230528/37609c867ef14657988e5ab6fb59f2bd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3Nzk2Njgz_size_16_color_FFFFFF_t_70 1]: /images/20230528/fddc9fb542ec42fdb5143b1276bdbd12.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3Nzk2Njgz_size_16_color_FFFFFF_t_70 2]: /images/20230528/d6ce2251f0ab4bf19df33a309bccd5ff.png
[20191022164841236.gif]: /images/20230528/a3a8ad6d10ad4b749e000490b58821d3.png
[20191021160639196.png]: /images/20230528/9403a02a8cad4932b0abe80f00137538.png
[20191021162231576.png]: /images/20230528/9bd3a2f6abb748b09a83c28b74c3a28a.png
[20191021173555376.gif]: /images/20230528/3622310acc364ed0925267162fd81b1e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3Nzk2Njgz_size_16_color_FFFFFF_t_70 3]: /images/20230528/9507910c39d943e19c8c8120c0bee614.png
[20191021181306111.png]: /images/20230528/98aae029cb2e40d8ac95b916976ef674.png
[20191021190904972.gif]: /images/20230528/576acf08b6f442b9a24ad47a4073c67f.png
[20191022104858288.png]: /images/20230528/0ca6bfc471d84dcbac9aab142fb4012a.png
[20191022122139759.png]: /images/20230528/60ad02dc71c549c79db20669af758d02.png
[20191022122200831.png]: /images/20230528/ba2421c7b36e4a5ea6f2e421d5dbb11a.png
[https_github.com_FollowExcellence_Rxjava_Retrofit]: https://github.com/FollowExcellence/Rxjava_Retrofit