Android之Rxjava2.X 5————Rxjava 过滤操作符

深碍√TFBOYSˉ_ 2022-05-16 02:58 203阅读 0赞

# Android之Rxjava2.X 5————Rxjava 过滤操作符 #

### 一. 目录 ###

### 文章目录 ###

*  Android之Rxjava2.X 5————Rxjava 过滤操作符
 *   *   *  一. 目录
         *  二. 概述
         *   *  1.作用
             *  2.类型
         *  三. 根据指定事件条件过滤事件
         *   *  1. filter()
             *  2.ofType()
             *  3.skip() & skipLast()
             *  4.distinct（） / distinctUntilChanged（）
         *  四. 根据指定事件数量过滤事件
         *   *  1.take（）
             *  2.takeLast（）
         *  五. 根据指定时间过滤事件
         *   *  1.throttleFirst（）/ throttleLast（）
             *  2.Sample（）
             *  3.throttleWithTimeout （） / debounce（）
         *  六. 根据指定事件位置过滤事件
         *   *  1.firstElement（） / lastElement（）
             *  2.elementAt（）
             *  3.elementAtOrError（）
         *  七.参考资料
         *  八.文章索引

### 二. 概述 ###

#### 1.作用 ####

过滤 / 筛选 被观察者（Observable）发送的事件 & 观察者 （Observer）接收的事件

#### 2.类型 ####

![这里写图片描述][700]

### 三. 根据指定事件条件过滤事件 ###

#### 1. filter() ####

*  作用：通过一定逻辑来过滤被观察者发送的事件，如果返回true则发送事件，否则不会发送
 *  应用场景：筛选符合要求的事件  
    原理图：  
    ![这里写图片描述][filter.png]

具体使用：

Observable.just(1,2,3,4,5).filter(new Predicate<Integer>() {
         
                @Override
                public boolean test(Integer integer) throws Exception {
         
                    return integer % 3 == 1;
                }
            }).subscribe(new Observer<Integer>() {
         
                @Override
                public void onSubscribe(Disposable d) {
         
                    Log.d(TAG, "onSubscribe: ");
                }
    
                @Override
                public void onNext(Integer integer) {
         
                    Log.d(TAG, "onNext: "+integer);
                }
    
                @Override
                public void onError(Throwable e) {
         
                    Log.d(TAG, "onError: ");
                }
    
                @Override
                public void onComplete() {
         
                    Log.d(TAG, "onComplete: ");
                }
            });

![这里写图片描述][20180814195720365]

#### 2.ofType() ####

*  作用：ofType是filter操作符的一个特殊形式。它过滤一个Observable只返回指定类型的数据。  
    原理图：  
    ![这里写图片描述][ofClass.png]

具体使用:

Observable.just(1, 2, "hello", "world")
                    .ofType(Integer.class)
                    .subscribe(new Observer<Integer>() {
         
                        @Override
                        public void onSubscribe(Disposable d) {
         
                            Log.d(TAG, "onSubscribe: ");
                        }
    
                        @Override
                        public void onNext(Integer integer) {
         
                            Log.d(TAG, "onNext: " + integer);
                        }
    
                        @Override
                        public void onError(Throwable e) {
         
                            Log.d(TAG, "onError: ");
                        }
    
                        @Override
                        public void onComplete() {
         
                            Log.d(TAG, "onComplete: ");
                        }
                    });

![这里写图片描述][20180814201906593]

#### 3.skip() & skipLast() ####

*  skip()作用：忽略Observable’发射的前N项数据，只保留之后的数据。
 *  skipLast()作用:从结尾往前数跳过制定数量的事件  
    原理图：  
    ![这里写图片描述][skip.c.png]  
    ![这里写图片描述][skipLast.c.png]

具体使用：

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

![这里写图片描述][20180814202606203]

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

![这里写图片描述][20180814202658100]

#### 4.distinct（） / distinctUntilChanged（） ####

*  作用：过滤事件序列中重复的事件 / 连续重复的事件

原理图：  
![这里写图片描述][distinct.png]  
![这里写图片描述][distinctUntilChanged.png]

具体使用;

// 使用1：过滤事件序列中重复的事件
            Observable.just(1, 2, 3,4,1,2,1)
                    .distinct()
                    .subscribe(new Observer<Integer>() {
         
                        @Override
                        public void onSubscribe(Disposable d) {
         
                            Log.d(TAG, "onSubscribe: ");
                        }
    
                        @Override
                        public void onNext(Integer integer) {
         
                            Log.d(TAG, "onNext: " + integer);
                        }
    
                        @Override
                        public void onError(Throwable e) {
         
                            Log.d(TAG, "onError: ");
                        }
    
                        @Override
                        public void onComplete() {
         
                            Log.d(TAG, "onComplete: ");
                        }
                    });
    
            //使用2：过滤事件序列中 连续重复的事件
            // 下面序列中，连续重复的事件 = 3、4
            Observable.just(1,2,3,1,2,3,3,4,4 )
                    .distinctUntilChanged()
                    .subscribe(new Observer<Integer>() {
         
                        @Override
                        public void onSubscribe(Disposable d) {
         
                            Log.d(TAG, "onSubscribe: ");
                        }
    
                        @Override
                        public void onNext(Integer integer) {
         
                            Log.d(TAG, "onNext: " + integer);
                        }
    
                        @Override
                        public void onError(Throwable e) {
         
                            Log.d(TAG, "onError: ");
                        }
    
                        @Override
                        public void onComplete() {
         
                            Log.d(TAG, "onComplete: ");
                        }
                    });

![这里写图片描述][20180814205651557]

### 四. 根据指定事件数量过滤事件 ###

需求场景：通过设置指定的事件数量，仅发送特定数量的事件

#### 1.take（） ####

*  作用： 只发射前面的N项数据、  
    原理图：  
    ![这里写图片描述][take.png]

具体使用：

// 使用1：过滤事件序列中重复的事件
            Observable.just(1, 2, 3,4,1,2,1)
                    .take(4)
                    .subscribe(new Observer<Integer>() {
         
                        @Override
                        public void onSubscribe(Disposable d) {
         
                            Log.d(TAG, "onSubscribe: ");
                        }
    
                        @Override
                        public void onNext(Integer integer) {
         
                            Log.d(TAG, "onNext: " + integer);
                        }
    
                        @Override
                        public void onError(Throwable e) {
         
                            Log.d(TAG, "onError: ");
                        }
    
                        @Override
                        public void onComplete() {
         
                            Log.d(TAG, "onComplete: ");
                        }
                    });

![这里写图片描述][70]

#### 2.takeLast（） ####

*  作用:发射Observable发射的最后N项数据

原理图：  
![这里写图片描述][takeLast.n.png]

具体使用：

// 使用1：过滤事件序列中重复的事件
            Observable.just(1, 2, 3,4,1,2,1)
                    .takeLast(2)
                    .subscribe(new Observer<Integer>() {
         
                        @Override
                        public void onSubscribe(Disposable d) {
         
                            Log.d(TAG, "onSubscribe: ");
                        }
    
                        @Override
                        public void onNext(Integer integer) {
         
                            Log.d(TAG, "onNext: " + integer);
                        }
    
                        @Override
                        public void onError(Throwable e) {
         
                            Log.d(TAG, "onError: ");
                        }
    
                        @Override
                        public void onComplete() {
         
                            Log.d(TAG, "onComplete: ");
                        }
                    });

![这里写图片描述][2018081421082066]

### 五. 根据指定时间过滤事件 ###

#### 1.throttleFirst（）/ throttleLast（） ####

*  作用：在某段时间内，只发送该段时间内第1次事件 / 最后1次事件（1段时间内连续点击按钮，但只执行第1次的点击操作）

原理图：  
![这里写图片描述][throttleFirst.png]

具体使用：

Observable.create(new ObservableOnSubscribe<Integer>() {
         
                @Override
                public void subscribe(ObservableEmitter<Integer> e) throws Exception {
         
                    e.onNext(1);
                    Thread.sleep(500);
                    e.onNext(2);
                    Thread.sleep(400);
                    e.onNext(3);
                    Thread.sleep(300);
                    e.onNext(4);
                    Thread.sleep(300);
                    e.onNext(5);
                    Thread.sleep(300);
                    e.onNext(6);
                    Thread.sleep(400);
                    e.onNext(7);
                    Thread.sleep(300);
                    e.onNext(8);
                    Thread.sleep(300);
                    e.onNext(9);
                    Thread.sleep(300);
                    e.onComplete();
    
                }
            }).throttleFirst(1, TimeUnit.SECONDS)//每1秒中采用数据
                    .subscribe(new Observer<Integer>() {
         
                        @Override
                        public void onSubscribe(Disposable d) {
         
                            Log.d(TAG, "onSubscribe: ");
                        }
    
                        @Override
                        public void onNext(Integer integer) {
         
                            Log.d(TAG, "onNext: " + integer);
                        }
    
                        @Override
                        public void onError(Throwable e) {
         
                            Log.d(TAG, "onError: ");
                        }
    
                        @Override
                        public void onComplete() {
         
                            Log.d(TAG, "onComplete: ");
                        }
                    });

![这里写图片描述][20180814212618457]

Observable.create(new ObservableOnSubscribe<Integer>() {
         
                @Override
                public void subscribe(ObservableEmitter<Integer> e) throws Exception {
         
                    e.onNext(1);
                    Thread.sleep(500);
                    e.onNext(2);
                    Thread.sleep(400);
                    e.onNext(3);
                    Thread.sleep(300);
                    e.onNext(4);
                    Thread.sleep(300);
                    e.onNext(5);
                    Thread.sleep(300);
                    e.onNext(6);
                    Thread.sleep(400);
                    e.onNext(7);
                    Thread.sleep(300);
                    e.onNext(8);
                    Thread.sleep(300);
                    e.onNext(9);
                    Thread.sleep(300);
                    e.onComplete();
    
                }
            }).throttleLast(1, TimeUnit.SECONDS)//每1秒中采用数据
                    .subscribe(new Observer<Integer>() {
         
                        @Override
                        public void onSubscribe(Disposable d) {
         
                            Log.d(TAG, "onSubscribe: ");
                        }
    
                        @Override
                        public void onNext(Integer integer) {
         
                            Log.d(TAG, "onNext: " + integer);
                        }
    
                        @Override
                        public void onError(Throwable e) {
         
                            Log.d(TAG, "onError: ");
                        }
    
                        @Override
                        public void onComplete() {
         
                            Log.d(TAG, "onComplete: ");
                        }
                    });

![这里写图片描述][2018081421273630]

#### 2.Sample（） ####

*  作用:在某段时间内，只发送该段时间内最新（最后）1次事件
 *  和throttleLast（） 操作符类似，仅需要把上文的 throttleLast（） 改成Sample（）操作符即可

#### 3.throttleWithTimeout （） / debounce（） ####

*  作用：发送数据事件时，若2次发送事件的间隔＜指定时间，就会丢弃前一次的数据，直到指定时间内都没有新数据发射时才会发送后一次的数据

原理图  
![这里写图片描述][debounce.png]

具体使用：

Observable.create(new ObservableOnSubscribe<Integer>() {
         
                @Override
                public void subscribe(ObservableEmitter<Integer> e) throws Exception {
         
                    e.onNext(1);
                    Thread.sleep(500);
                    e.onNext(2);
                    Thread.sleep(1011);
                    e.onNext(3);
                    Thread.sleep(300);
                    e.onNext(4);
                    Thread.sleep(600);
                    e.onNext(5);
                    Thread.sleep(1200);
                    e.onNext(6);
                    Thread.sleep(400);
                    e.onNext(7);
                    Thread.sleep(300);
                    e.onNext(8);
                    Thread.sleep(300);
                    e.onNext(9);
                    Thread.sleep(300);
                    e.onComplete();
    
                }
            }).throttleWithTimeout(1, TimeUnit.SECONDS)//每1秒中采用数据
                    .subscribe(new Observer<Integer>() {
         
                        @Override
                        public void onSubscribe(Disposable d) {
         
                            Log.d(TAG, "onSubscribe: ");
                        }
    
                        @Override
                        public void onNext(Integer integer) {
         
                            Log.d(TAG, "onNext: " + integer);
                        }
    
                        @Override
                        public void onError(Throwable e) {
         
                            Log.d(TAG, "onError: ");
                        }
    
                        @Override
                        public void onComplete() {
         
                            Log.d(TAG, "onComplete: ");
                        }
                    });

![这里写图片描述][20180814213741635]

### 六. 根据指定事件位置过滤事件 ###

#### 1.firstElement（） / lastElement（） ####

*  作用：仅选取第1个元素 / 最后一个元素  
    具体使用：

Observable.just(1, 2, 3, 4, 5)
                    .firstElement()
                    .subscribe(new Consumer<Integer>() {
         
                        @Override
                        public void accept(Integer integer) throws Exception {
         
                            Log.d(TAG, "获取到的第一个事件是： " + integer);
                        }
                    });
    
            Observable.just(1, 2, 3, 4, 5)
                    .lastElement()
                    .subscribe(new Consumer<Integer>() {
         
                        @Override
                        public void accept(Integer integer) throws Exception {
         
                            Log.d(TAG, "获取到的第一个事件是： " + integer);
                        }
                    });

![这里写图片描述][20180814214302945]

#### 2.elementAt（） ####

*  作用：指定接收某个元素（通过 索引值 确定）
 *  允许越界，即获取的位置索引 ＞ 发送事件序列长度  
    具体使用：

Observable.just(1, 2, 3, 4, 5)
                    .elementAt(2)
                    .subscribe(new Consumer<Integer>() {
         
                        @Override
                        public void accept(Integer integer) throws Exception {
         
                            Log.d(TAG, "获取到的第一个事件是： " + integer);
                        }
                    });

![这里写图片描述][20180814214701354]

#### 3.elementAtOrError（） ####

*  作用:在elementAt（）的基础上，当出现越界情况（即获取的位置索引 ＞ 发送事件序列长度）时，即抛出异常

具体使用:

Observable.just(1, 2, 3, 4, 5)
                   .elementAtOrError(7)
                   .subscribe(new Consumer<Integer>() {
         
                       @Override
                       public void accept(Integer integer) throws Exception {
         
                           Log.d(TAG, "获取到的第一个事件是： " + integer);
                       }
                   });

![这里写图片描述][70 1]

### 七.参考资料 ###

[Android RxJava：过滤操作符 全面讲解][Android RxJava_]  
[RxJava文档中文版][RxJava]  
[Android RxJava2(四)过滤操作符][Android RxJava2]

### 八.文章索引 ###

*  [Android之Rxjava2.X 1————Rxjava概述][Android_Rxjava2.X 1_Rxjava]
 *  [Android之Rxjava2.X 2————Rxjava 创建操作符][Android_Rxjava2.X 2_Rxjava]
 *  [Android之Rxjava2.X 3————Rxjava 变换操作符][Android_Rxjava2.X 3_Rxjava]
 *  [Android之Rxjava2.X 4————Rxjava 组合操作符][Android_Rxjava2.X 4_Rxjava]
 *  [Android之Rxjava2.X 5————Rxjava 过滤操作符][Android_Rxjava2.X 5_Rxjava]
 *  [Android之Rxjava2.X 6————Rxjava 功能操作符][Android_Rxjava2.X 6_Rxjava]
 *  [Android之Rxjava2.X 7————Rxjava 条件操作符][Android_Rxjava2.X 7_Rxjava]
 *  [Android之Rxjava2.X 8————Rxjava 背压策略][Android_Rxjava2.X 8_Rxjava]
 *  [Android之Rxjava2.X 9————Rxjava源码阅读1][Android_Rxjava2.X 9_Rxjava_1]
 *  [Android之Rxjava2.X 10————Rxjava源码阅读2][Android_Rxjava2.X 10_Rxjava_2]
 *  [Android之Rxjava2.X 11————Rxjava源码阅读3][Android_Rxjava2.X 11_Rxjava_3]

[700]: /images/20220516/17c464cff25e493292cf27b602207de7.png
[filter.png]: https://mcxiaoke.gitbooks.io/rxdocs/content/images/operators/filter.png
[20180814195720365]: /images/20220516/4407350d740c4c5eb4b744bc3b3b5919.png
[ofClass.png]: https://mcxiaoke.gitbooks.io/rxdocs/content/images/operators/ofClass.png
[20180814201906593]: /images/20220516/ce598343f9ba4996bddb6c9444460ca8.png
[skip.c.png]: https://mcxiaoke.gitbooks.io/rxdocs/content/images/operators/skip.c.png
[skipLast.c.png]: https://mcxiaoke.gitbooks.io/rxdocs/content/images/operators/skipLast.c.png
[20180814202606203]: /images/20220516/59da5f3711f94bcda44b03dbe9e98ced.png
[20180814202658100]: /images/20220516/004dddb931a74f4abca02bd97c90d1a9.png
[distinct.png]: https://mcxiaoke.gitbooks.io/rxdocs/content/images/operators/distinct.png
[distinctUntilChanged.png]: https://mcxiaoke.gitbooks.io/rxdocs/content/images/operators/distinctUntilChanged.png
[20180814205651557]: /images/20220516/2d70e17fdf06493784c4a066bb3a2ca5.png
[take.png]: https://mcxiaoke.gitbooks.io/rxdocs/content/images/operators/take.png
[70]: /images/20220516/c1965e9b1cc94ce59259b67271c9d4e0.png
[takeLast.n.png]: https://mcxiaoke.gitbooks.io/rxdocs/content/images/operators/takeLast.n.png
[2018081421082066]: /images/20220516/cc04d8edd51845bfb9e09320b663dd4b.png
[throttleFirst.png]: https://mcxiaoke.gitbooks.io/rxdocs/content/images/operators/throttleFirst.png
[20180814212618457]: /images/20220516/cd416294a60e4e7cb6317b5d9380003f.png
[2018081421273630]: /images/20220516/2dbc82a9732b475a83614f2c60cb1d64.png
[debounce.png]: https://mcxiaoke.gitbooks.io/rxdocs/content/images/operators/debounce.png
[20180814213741635]: /images/20220516/fc15d68670de4ca1ada796e95eb1e5c5.png
[20180814214302945]: /images/20220516/fbd3b61c2c9a46bbbe3838ec3a81ec05.png
[20180814214701354]: /images/20220516/5b25f2bb9a6a4c8bac9ccc420de556c7.png
[70 1]: /images/20220516/02a7f667a4834972b36997374f1a5f13.png
[Android RxJava_]: https://www.jianshu.com/p/c3a930a03855
[RxJava]: https://mcxiaoke.gitbooks.io/rxdocs/content/operators/First.html
[Android RxJava2]: https://blog.csdn.net/mixin716/article/details/80607295#firstelement-lastelement
[Android_Rxjava2.X 1_Rxjava]: https://blog.csdn.net/qq_38499859/article/details/81611870
[Android_Rxjava2.X 2_Rxjava]: https://blog.csdn.net/qq_38499859/article/details/81637932
[Android_Rxjava2.X 3_Rxjava]: https://blog.csdn.net/qq_38499859/article/details/81668545
[Android_Rxjava2.X 4_Rxjava]: https://blog.csdn.net/qq_38499859/article/details/81670980
[Android_Rxjava2.X 5_Rxjava]: https://blog.csdn.net/qq_38499859/article/details/81675322
[Android_Rxjava2.X 6_Rxjava]: https://blog.csdn.net/qq_38499859/article/details/81700187
[Android_Rxjava2.X 7_Rxjava]: https://blog.csdn.net/qq_38499859/article/details/81705745
[Android_Rxjava2.X 8_Rxjava]: https://blog.csdn.net/qq_38499859/article/details/81747334
[Android_Rxjava2.X 9_Rxjava_1]: https://blog.csdn.net/qq_38499859/article/details/81775520
[Android_Rxjava2.X 10_Rxjava_2]: https://blog.csdn.net/qq_38499859/article/details/81839955
[Android_Rxjava2.X 11_Rxjava_3]: https://blog.csdn.net/qq_38499859/article/details/82119900