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

迷南。 2023-07-17 12:57 18阅读 0赞

# 一、概述 #

RxJava2在第一篇文章作了详细的介绍，第二篇文章中讲解了RxJava的变换操作符，组合操作符，合并操作符；第三篇讲解了延迟、do相关、错误处理等操作符的使用，如果英文比较好而且有兴趣的可以到官网学习：[ReactiveX的官方网站][ReactiveX]。下面开始继续讲解RxJava2的其他用法。（链接和源码在文章最后给出）

# 二、过滤操作符 #

### 1.filter() ###

filter()对源Observable(被观察者)产生的结果按照执行条件过虑，只有满足条件才会提交给观察者。

Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9)
                .filter(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                //条件：能否被2整除
                return integer % 2 == 0;
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "filter：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "filter：apply == 事件" + integer);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "filter：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "filter：onComplete == ");
            }
        });

例子中放1~9共9个数据，在filter回调方法中添加条件判断，满足则将结果发送给观察者，满足能被2整除的数是：2,4,6,8；打印数据如下：

![20191024155733131.png][]

### 2.ofType() ###

ofType()根据类型对源Observable(被观察者)产生的结果进行过滤，只有满足条件才会提交给观察者Observer。ofType()与filter()类似，但是ofType()是根据类型对结果进行过滤的。

Observable.just(1, "大闸蟹", true, 0.23f, 5L, new Goods("商品名"))
                .ofType(Goods.class).subscribe(new Observer<Goods>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "ofType：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Goods goods) {
                Log.e(TAG, "ofType：onNext == " + goods.getName());
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "ofType：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "ofType：onComplete == ");
            }
        });

这里事件队列有多种类型，int、String、Boolean、float、Long还有自定义的实体类，过滤实体类的数据Goods.calss，打印数据如下：

![20191101212548359.png][]

### 3.elementAt() ###

elementAt()把源Observable（被观察者）生产的结果中，只提交索引指示的结果给Observer（观察者）。注意：索引从0开始。

Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9)
                 .elementAt(7)
                 .subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "elementAt：accept == " + integer);
            }
        });

这里打印索引（索引从0开始计算）为7的数据：8，打印数据如下：

![20191024184827877.png][]

### 4.distinct() ###

distinct()：对源Observable的结果进行过滤，去掉重复出现的结果，只输出不重复的结果给Observer观察者，简单来说就是去重。

Observable.just(1, 2, 2, 3, 4, 4, 4)
                 .distinct()
                    .subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "distinct：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "distinct：onNext == " + integer);
     
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "distinct：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "distinct：onComplete == ");
            }
        });

这里的事件队列中，2和4皆有多个重复的数据，如果过滤重复的那么只会出现：1,2,3,4，打印数据如下：

![20191024185904266.png][]

### 5. debounce() ###

debounce()：只接收到倒计时时间外的源Observable(被观察者)发出的事件，每次发送的事件都会重置倒计时时间。

Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) {
                try {//产生结果的时间间隔为0,100,200……900毫秒，每次发送的数据为0,1,2……9
                    for (int i = 0; i < 10; i++) {
                        e.onNext(i);
                        Thread.sleep(i * 100);
                    }
                } catch (Exception exception) {
                    e.onError(exception);
                }
            }
        }).debounce(300, TimeUnit.MILLISECONDS)
                    .subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "debounce：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "debounce：onNext ==" + integer);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "debounce：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "debounce：onComplete == ");
            }
        });

这里每隔0,100,200……900毫秒发送的数据分别为0,1,2……9，debounce()中指定倒计时的时间为300毫秒，只接受到300毫秒后的结果，打印数据如下：

![20191031175613436.gif][]

### 6.first() ###

first()：将源Observable(被观察者)产生的第一次结果提交给Observer(被观察者)，first()可以用elementAt(0)和take(1)替代。

*  first(T defaultItem)：defaultItem表示默认值。

Observable.just(1,2,3)
                .first(0)//defaultItem：默认值
                    .subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "first：accept ==" + integer);
            }
        });

这里发送1,2,3，接受到第一个结果为：1，如果接受不到数据则直接输出默认值，这里设置的默认值为0，打印数据如下：

![20191031182509933.png][]

### 7.last() ###

last()：表示将源Observable(被观察者)产生的最后一次结果提交给Observer(被观察者)，与first()相对应，可以用takelast()替代。

Observable.just(1, 2, 3, 4)
            .last(-1)
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    Log.e(TAG, "last：accept ==" + integer);
                  }
            });

发出4个事件，返回最后一个事件的结果：4，默认值为-1（在获取不到结果的时候返回），打印结果如下：

![20191031185305636.png][]

### 8.skip() ###

skip()：针对源Observable(被观察者)产生的结果跳过N个不处理，而把后面的结果提交给Observer(观察者)处理。与takeLast()类似。

Observable.just(1, 2, 3, 4, 5, 6, 7)
             .skip(3)
             .subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "skip：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "skip：onNext ==" + integer);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "skip：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "skip：onComplete == ");
            }
        });

这里发送7个事件，跳过前三个，打印数据如下：

![2019103119161114.png][]

### 9.take() ###

take()：源Observable(被观察者)产生的结果，把前面的n个提交给Observer(观察者)处理，后面的不做处理。与skip()相对。

Observable.just(1, 2, 3, 4, 5, 6, 7)
             .take(3)
             .subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "take：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "take：onNext ==" + integer);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "take：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "take：onComplete == ");
            }
        });

这里发送了7个事件，只提交前面三个数据给Observer，打印数据如下：

![20191031193436830.png][]

takeLast()表示把最后的n项结果提交给Observer，与skip()类似，这里就不举例了。

这里对过滤操作符做一个简单的总结：

<table> 
 <tbody> 
  <tr> 
   <th colspan="2">创建类型</th> 
   <th>作用</th> 
   <th>备注</th> 
   <th>使用场景</th> 
  </tr> 
  <tr> 
   <td>过滤操作符</td> 
   <td>filter()</td> 
   <td>按照指定的条件(自定义)对被观察者发出的结果进行过滤，再提交给被观察者</td> 
   <td>在回调方法中实现规则过滤</td> 
   <td colspan="1"> <p>需要过滤或者选择某一步操作</p> </td> 
  </tr> 
  <tr> 
   <td>ofType()</td> 
   <td>根据指定的数据类型对被观察者发出的结果进行过滤，再提交给被观察者</td> 
   <td>参数传入对结果过滤的类型</td> 
  </tr> 
  <tr> 
   <td>elementAt()</td> 
   <td>将指定索引被观察者发出的事件提交给观察者，索引为0</td> 
   <td>参数传入索引</td> 
  </tr> 
  <tr> 
   <td>distinct()</td> 
   <td>去重，去掉被观察者产生的重复的结果，将不重复的提交给观察者</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>debounce()</td> 
   <td>只接收到倒计时时间外的被观察者发出的事件，每个事件发出时会重置倒计时时间。</td> 
   <td>参数传入每个事件的倒计时时间</td> 
  </tr> 
  <tr> 
   <td>first()</td> 
   <td>将被观察者产生的第一个结果提交给被观察者处理</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>last()</td> 
   <td>将被观察者产生的最后一个结果提交给观察者处理</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>skip()</td> 
   <td>跳过前n个被观察者产生的结果，将后面的提交给观察者处理</td> 
   <td>参数传入指定跳过的数量</td> 
  </tr> 
  <tr> 
   <td>take()</td> 
   <td>将前n个被观察者产生的结果提交给观察者处理，其他的不做处理</td> 
   <td>参数传入指定执行的数量</td> 
  </tr> 
 </tbody> 
</table>

# 三、其他操作符 #

### 1.groupBy() ###

groupBy()：对于源Observable(被观察者)产生的结果进行分组，形成一个类型为GroupedObservable的结果集，GroupedObservable存在一个方法为getKey()，通过该方法获取结果集的key值。

注意：由于结果集GroupedObservable把分组结果缓存起来，如果对每个GroupedObservable都不进行处理（既不订阅也没有别的操作）就有可能出现内存泄漏，如果某个GroupedObservable不处理那可以使用take(0)处理掉。

//每隔一秒发送一次事件，一共发送五次
    Observable.interval(1, TimeUnit.SECONDS)
                .take(5)
                .groupBy(new Function<Long, Object>() {
            @Override
            public Object apply(Long aLong) throws Exception {
                return aLong * 10;
            }
        }).subscribe(new Observer<GroupedObservable<Object, Long>>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "take：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(GroupedObservable<Object, Long> objectLongGroupedObservable) {
                Log.e(TAG, "take：onNext == key:" + objectLongGroupedObservable.getKey());
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "take：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "take：onComplete == ");
            }
        });

这里每隔1秒发送一次事件，一共发送五次，key值\*10处理，打印结果如下：

![20191101114203135.gif][]

### 2.cast() ###

cast()：主要是用于类型转换的，传入的参数类型为.class,如果源Observable(被观察者)的源类型不能转为指定的.class,  则会抛出ClassCastException类型转换异常。

Observable.just(1, 2, 3, 4, 5, 6)
             .cast(Integer.class)
             .subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "cast：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "cast：onNext == " + integer);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "cast：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "cast：onComplete == ");
            }
        });

这里发出1~6个事件，打印数据如下：

![20191101115600899.png][]

### 3.scan() ###

scan()：通过遍历源Observable(被观察者)产生的结果，依次每个结果按照指定的规则进行计算，计算后的结果作为下一项迭代的参数，每一次迭代项都会把计算结果输出给订阅者。简单来说就是每次计算后的结果都会用于下一次计算。

Observable.just(1, 2, 3)
             .scan(new BiFunction<Integer, Integer, Integer>() {
            @Override
            public Integer apply(Integer sum, Integer integer) throws Exception {
                //sum是上次计算的记过，integer是本次计算的参数
                Log.e(TAG, "scan：apply == sum + integer = " + sum + " + " + integer);
                return sum + integer;
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "scan：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "scan：onNext == " + integer);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "scan：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "scan：onComplete == ");
            }
        });

这里只有1,2,3共三个事件，第一次计算：0+1；第二次计算：第一次计算的结果+第二个参数，即：1+2=3；第三次计算：上一次计算的结果+第三个参数，即：3+3=6，打印数据如下：

![20191101122016742.png][]

### 4.join() ###

join()：把两个Observable(被观察者)产生的结果进行合并，合并成的结果组成一个新的Observable，可以控制每个Observable产生结果的生命周期，在每个结果的生命周期内，可以与另一个Observable产生的结果按照一定的规则进行合并。类似combineLatest()。

join()的用法如下：

> ObservableA.join(ObservableB, ObservableA产生结果的生命周期控制函数，ObservableB产生结果的生命周期控制函数, ObservableA与ObservableB结果合并规则函数)

//每隔一秒，产生0,5,10,15,20事件队列
        Observable<Long> observable1 = Observable.interval(0,1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        return aLong * 5;
                    }
                }).take(5);
     
        //延时500毫秒，每隔一秒，产生0,10,20,30,40 事件队列
        Observable<Long> observable2 = Observable.interval(500, 1000,TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) {
                        return aLong * 10;
                    }
                }).take(5);
     
     
        observable1.join(observable2, new Function<Long, ObservableSource<Long>>() {
            @Override
            public ObservableSource<Long> apply(Long aLong) {
                //使Observable延时600毫秒执行(控制observable1的生命周期)
                return Observable.just(aLong).delay(600,TimeUnit.MILLISECONDS);
            }
        }, new Function<Long, ObservableSource<Long>>() {
            @Override
            public ObservableSource<Long> apply(Long aLong) {
                //使Observable延时600毫秒执行(控制observable2的生命周期)
                return Observable.just(aLong).delay(600,TimeUnit.MILLISECONDS);
            }
        }, new BiFunction<Long, Long, Long>() {
            @Override
            public Long apply(Long aLong, Long aLong2) {
                //合并逻辑
                Log.e(TAG, "BiFunction：apply == aLong1 + aLong2：" + aLong + " + " + aLong2);
                return aLong + aLong2;
            }
        }).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "join：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Long aLong) {
                Log.e(TAG, "join：onNext == " + aLong);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "join：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "join：onComplete == ");
            }
        });

Observable1每隔1秒产生0,5,10,15,20 事件队列，Observable2延时0.5秒，每隔1秒产生0,10,20,30,40 事件队列，第一个Function函数控制Observable1结果的生命周期，第二个Function函数控制Observable2结果的生命周期，都做了延时600毫秒发送，第三个BiFunction函数整合Observable1，Observable2产生的结果，打印数据如下：

![20191101194645352.gif][]

### 5.groupJoin() ###

groupJoin()：类似于join()，也是把两个Observable产生的结果进行合并，组成一个新的Observable，可以控制每个Observable产生结果的生命周期，在每个结果的生命周期内，与另一个Observable的结果进行合并按照一定的规则进行合并。与join()不同的是第四个参数传入的函数不一致。

//每隔1秒，产生0,5,10,15,20事件队列
        Observable<Long> observable1 = Observable.interval(0, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) {
                        return aLong * 5;
                    }
                }).take(5);
     
        //延迟0.5秒，每秒产生0,10,20,30,40事件队列
        Observable<Long> observable2 = Observable.interval(500, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) {
                        return aLong * 10;
                    }
                }).take(5);
     
            observable1.groupJoin(observable2, new Function<Long, ObservableSource<Long>>() {
            @Override
            public ObservableSource<Long> apply(Long aLong) {
                //使Observable延时1600毫秒执行(控制observable1的生命周期)
                return Observable.just(aLong).delay(1600, TimeUnit.MILLISECONDS);
            }
        }, new Function<Long, ObservableSource<Long>>() {
            @Override
            public ObservableSource<Long> apply(Long aLong) {
                //使Observable延时600毫秒执行(控制observable2的生命周期)
                return Observable.just(aLong).delay(600, TimeUnit.MILLISECONDS);
            }
        }, new BiFunction<Long, Observable<Long>, Observable<Long>>() {
            @Override
            public Observable<Long> apply(final Long aLong, Observable<Long> longObservable) {
                return longObservable.map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong2) {
                        //合并逻辑
                        Log.e(TAG, "BiFunction：apply == aLong1 + aLong2：" + aLong + " + " + aLong2);
                        return aLong + aLong2;
                    }
                });
            }
        }).subscribe(new Observer<Observable<Long>>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "groupJoin：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Observable<Long> longObservable) {
                longObservable.subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) {
                        Log.e(TAG, "groupJoin：onNext == " + aLong);
                    }
                });
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "groupJoin：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "groupJoin：onComplete == ");
            }
        });

Observable1每隔1秒产生0,5,10,15,20事件队列，Observable2延时0.5秒，每隔一秒产生0,10,20,30,40事件队列，第一个Function函数控制Observable1结果的生命周期，第二个Function函数控制Observable2结果的生命周期，第三个函数BiFunction整合Observable1和Observable2的结果，返回的结果是新的Observable。打印结果如下：

![20191101203014365.gif][]

我们来对这部分的操作符做个小结：

<table> 
 <tbody> 
  <tr> 
   <th colspan="2">创建类型</th> 
   <th>作用</th> 
   <th>备注</th> 
  </tr> 
  <tr> 
   <td>过滤操作符</td> 
   <td>groupBy()</td> 
   <td>对于源Observable(被观察者)产生的结果进行分组，形成一个类型为GroupedObservable的结果集，存在一个方法为getKey()，通过该方法获取结果集的key值</td> 
   <td>如果事件结果没有处理，那么take()处理一下，否则会内存泄漏</td> 
  </tr> 
  <tr> 
   <td>cast()</td> 
   <td>用于类型转换，将源Observable的源类型转为指定的.class</td> 
   <td>如果类型不正确则会抛出转换异常。</td> 
  </tr> 
  <tr> 
   <td>scan()</td> 
   <td>遍历源Observable产生的结果，依次每个结果按照指定的规则进行计算，计算后的结果作为下一项迭代的参数，每一次迭代项都会把计算结果输出给订阅者。</td> 
   <td>每次计算的结果用于下次计算</td> 
  </tr> 
  <tr> 
   <td>join()</td> 
   <td>把两个Observable产生的结果进行合并，合并结果组成一个新的Observable，可以控制每个Observable产生结果的生命周期，在每个结果的生命周期内，可以与另一个Observable产生的结果按照一定的规则进行合并</td> 
   <td>合并结果产生新的Observable</td> 
  </tr> 
  <tr> 
   <td>groupJoin()</td> 
   <td>类似上述，第三个参数不一致，返回新的Observable处理</td> 
   <td>&nbsp;</td> 
  </tr> 
 </tbody> 
</table>

至此，本文结束！

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

[ReactiveX]: http://reactivex.io/documentation/operators.html
[20191024155733131.png]: /images/20230528/9328fa2f53104201bac3786b33bc7c93.png
[20191101212548359.png]: /images/20230528/a0551899022843228f103264666853c1.png
[20191024184827877.png]: /images/20230528/44bd4b535d7d4be9ab5af047375f323e.png
[20191024185904266.png]: /images/20230528/16d811412bd24fcd9877ef81366c2ad0.png
[20191031175613436.gif]: /images/20230528/67b5dbb604fc4459b038bb15e1f099b3.png
[20191031182509933.png]: /images/20230528/7bfe6e703d62498f8597c525859dae64.png
[20191031185305636.png]: /images/20230528/4ad9337485864d98ab40b56980795f0c.png
[2019103119161114.png]: /images/20230528/3496893d1313414a86166dcacfc29aee.png
[20191031193436830.png]: /images/20230528/6fb7e433b4b342f8ab5cbc8f7f090b64.png
[20191101114203135.gif]: /images/20230528/cbf0df13c5d045299c58a36c18052035.png
[20191101115600899.png]: /images/20230528/5bc1265b6fca45c89746a5a42fe1f965.png
[20191101122016742.png]: /images/20230528/8d4c9541ac9e4002a3afa92d8a685e95.png
[20191101194645352.gif]: /images/20230528/e3f5786dc0534353b289454924b2accd.png
[20191101203014365.gif]: /images/20230528/2366f3fdfd004a1fb1cdcc6f8c866c9e.png
[https_github.com_FollowExcellence_Rxjava_Retrofit]: https://github.com/FollowExcellence/Rxjava_Retrofit