前言

Rxjava，由于其基于事件流的链式调用、逻辑简洁 & 使用简单的特点，深受各大 Android开发者的欢迎。

Github截图

如果还不了解 RxJava，请看文章：Android：这是一篇清晰 & 易懂的Rxjava 入门教程

RxJava如此受欢迎的原因，在于其提供了丰富 & 功能强大的操作符，几乎能完成所有的功能需求
今天，我将为大家详细介绍RxJava操作符中最常用的 条件 / 布尔操作符，希望你们会喜欢。

本系列文章主要基于 Rxjava 2.0

接下来的时间，我将持续推出 Android中 Rxjava 2.0 的一系列文章，包括原理、操作符、应用场景、背压等等 ，有兴趣可以继续关注Carson_Ho的安卓开发笔记！！

1. 作用

通过设置函数，判断被观察者（Observable）发送的事件是否符合条件

2. 类型

RxJava2中，条件 / 布尔操作符的类型包括：

示意图

下面，我将对每个操作符进行详细讲解

3. 具体操作符详解

注：在使用RxJava 2操作符前，记得在项目的Gradle中添加依赖：

dependencies {
      compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
      compile 'io.reactivex.rxjava2:rxjava:2.0.7'
      // 注：RxJava2 与 RxJava1 不能共存，即依赖不能同时存在
}

3.1 all（）

作用
判断发送的每项数据是否都满足设置的函数条件

若满足，返回 true；否则，返回 false

具体代码


  Observable.just(1,2,3,4,5,6)
                .all(new Predicate<Integer>(){
                    @Override
                    public boolean test( Integer integer) throws Exception {
                        return (integer<=10);
                        // 该函数用于判断Observable发送的10个数据是否都满足integer<=10
                    }
                }).subscribe(new Consumer<Boolean>() {
            @Override
            public void accept(Boolean aBoolean) throws Exception {
                Log.d(TAG,"result is "+ aBoolean);
                // 输出返回结果
            }

        });

测试结果

因为所有数据都满足函数内条件（每项数据<=10）

示意图

3.2 exists（）

作用
判断发送的数据是否存在1项数据满足设置的函数条件

若满足，返回 true；否则，返回 false

具体代码

Observable.just(1,2,3,4,5,6)
                .all(new Predicate<Integer>(){
                    @Override
                    public boolean test( Integer integer) throws Exception {
                        return (integer<=3);
                        // 该函数用于判断Observable发送的6个数据是否存在1项数据满足函数条件：integer<=3
                    }
                }).subscribe(new Consumer<Boolean>() {
            @Override
            public void accept(Boolean aBoolean) throws Exception {
                Log.d(TAG,"result is "+ aBoolean);
                // 输出返回结果
            }

        });

测试结果

因为存在数据（1 & 2）满足函数内条件（每项数据<=3）

示意图

3.3 takeWhile（）

作用
判断发送的每项数据是否满足设置函数条件

若发送的数据满足该条件，则发送该项数据；否则不发送

具体代码

// 1. 每1s发送1个数据 = 从0开始，递增1，即0、1、2、3
        Observable.interval(1, TimeUnit.SECONDS)
                // 2. 通过takeWhile传入一个判断条件
                .takeWhile(new Predicate<Long>(){
                    @Override
                    public boolean test( Long integer) throws Exception {
                        return (integer<3);
                        // 当发送的数据满足<3时，才发送Observable的数据
                    }
                }).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
            }

            @Override
            public void onNext(Long value) {
                Log.d(TAG,"发送了事件 "+ value);
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {
            }
        });

测试结果

示意图

3.4 skipWhile（）

作用
判断发送的每项数据是否满足设置函数条件

直到该判断条件 = false时，才开始发送Observable的数据

具体使用

// 1. 每隔1s发送1个数据 = 从0开始，每次递增1
        Observable.interval(1, TimeUnit.SECONDS)
                // 2. 通过skipWhile（）设置判断条件
                .skipWhile(new Predicate<Long>(){
                    @Override
                    public boolean test( Long aLong) throws Exception {
                        return (aLong<5);
                        // 直到判断条件不成立 = false = 发射的数据≥5，才开始发送数据
                    }
                }).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
            }

            @Override
            public void onNext(Long value) {
                Log.d(TAG,"发送了事件 "+ value);
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {
            }
        });

测试结果

示意图

3.5 takeUntil（）

作用
执行到某个条件时，停止发送事件

具体使用

// 1. 每1s发送1个数据 = 从0开始，递增1，即0、1、2、3
        Observable.interval(1, TimeUnit.SECONDS)
                // 2. 通过takeUntil的Predicate传入判断条件
                .takeUntil(new Predicate<Long>(){
                    @Override
                    public boolean test( Long integer) throws Exception {
                        return (integer>3);
                        // 返回true时，就停止发送事件
                        // 当发送的数据满足>3时，就停止发送Observable的数据
                    }
                }).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
            }

            @Override
            public void onNext(Long value) {
                Log.d(TAG,"发送了事件 "+ value);
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {
            }
        });

测试结果

示意图

该判断条件也可以是Observable，即等到 takeUntil（） 传入的Observable开始发送数据，（原始）第1个Observable的数据停止发送数据

// （原始）第1个Observable：每隔1s发送1个数据 = 从0开始，每次递增1
        Observable.interval(1, TimeUnit.SECONDS)
                // 第2个Observable：延迟5s后开始发送1个Long型数据
                .takeUntil(Observable.timer(5, TimeUnit.SECONDS))
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(TAG, "开始采用subscribe连接");
                    }

                    @Override
                    public void onNext(Long value) {
                        Log.d(TAG, "接收到了事件"+ value  );
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, "对Error事件作出响应");
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "对Complete事件作出响应");
                    }

                });

测试结果

当第 5s 时，第2个 Observable 开始发送数据，于是（原始）第1个 Observable 停止发送数据

示意图

3.6 skipUntil（）

作用
等到 skipUntil（） 传入的Observable开始发送数据，（原始）第1个Observable的数据才开始发送数据
具体使用

                // （原始）第1个Observable：每隔1s发送1个数据 = 从0开始，每次递增1
        Observable.interval(1, TimeUnit.SECONDS)
                // 第2个Observable：延迟5s后开始发送1个Long型数据
                .skipUntil(Observable.timer(5, TimeUnit.SECONDS))
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(TAG, "开始采用subscribe连接");
                    }

                    @Override
                    public void onNext(Long value) {
                        Log.d(TAG, "接收到了事件"+ value  );
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, "对Error事件作出响应");
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "对Complete事件作出响应");
                    }

                });

测试结果

5s后（ skipUntil（） 传入的Observable开始发送数据），（原始）第1个Observable的数据才开始发送

示意图

3.7 SequenceEqual（）

作用
判定两个Observables需要发送的数据是否相同

若相同，返回 true；否则，返回 false

具体使用

Observable.sequenceEqual(
                Observable.just(4,5,6),
                Observable.just(4,5,6)
        )
                .subscribe(new Consumer<Boolean>() {
                    @Override
                    public void accept( Boolean aBoolean) throws Exception {
                        Log.d(TAG,"2个Observable是否相同："+ aBoolean);
                        // 输出返回结果
                    }
                });

测试结果

示意图

3.8 contains（）

作用
判断发送的数据中是否包含指定数据

若包含，返回 true；否则，返回 false

内部实现 = exists（）

具体代码

Observable.just(1,2,3,4,5,6)
                .contains(4)
                .subscribe(new Consumer<Boolean>() {
            @Override
            public void accept(Boolean aBoolean) throws Exception {
                Log.d(TAG,"result is "+ aBoolean);
                // 输出返回结果
            }

        });

测试结果

因为发送的数据中包含4

示意图

3.9 isEmpty（）

作用
判断发送的数据是否为空

若为空，返回 true；否则，返回 false

具体代码

Observable.just(1,2,3,4,5,6)
          .isEmpty() // 判断发送的数据中是否为空
        }).subscribe(new Action1<Boolean>() {
            @Override
            public void call(Boolean aBoolean) {
                Log.d(TAG,"result is "+ aBoolean); 
                // 输出返回结果
            }
        });

测试结果

因为发送的数据不为空

示意图

3.10 amb（）

作用
当需要发送多个 Observable时，只发送 先发送数据的Observable的数据，而其余 Observable则被丢弃。
具体代码

       // 设置2个需要发送的Observable & 放入到集合中
        List<ObservableSource<Integer>> list= new ArrayList <>();
        // 第1个Observable延迟1秒发射数据
        list.add( Observable.just(1,2,3).delay(1,TimeUnit.SECONDS));
        // 第2个Observable正常发送数据
        list.add( Observable.just(4,5,6));

        // 一共需要发送2个Observable的数据
        // 但由于使用了amba（）,所以仅发送先发送数据的Observable
        // 即第二个（因为第1个延时了）
        Observable.amb(list).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "接收到了事件 "+integer);
            }
        });

测试结果

即只发送了先发送数据的Observable的数据 = 4，5，6

示意图

3.11 defaultIfEmpty（）

作用
在不发送任何有效事件（ Next事件）、仅发送了 Complete 事件的前提下，发送一个默认值
具体使用

Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                // 不发送任何有效事件
                //  e.onNext(1);
                //  e.onNext(2);

                // 仅发送Complete事件
                e.onComplete();
            }
        }).defaultIfEmpty(10) // 若仅发送了Complete事件，默认发送 值 = 10
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(TAG, "开始采用subscribe连接");
                    }

                    @Override
                    public void onNext(Integer value) {
                        Log.d(TAG, "接收到了事件"+ value  );
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, "对Error事件作出响应");
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "对Complete事件作出响应");
                    }
                });

测试结果

示意图

至此，RxJava2中常用的条件 / 布尔操作符讲解完毕

4. Demo地址

上述所有的Demo源代码都存放在：Carson_Ho的Github地址：RxJava2_功能性操作符

5. 总结

下面，我将用一张图总结 RxJava2 中常用的条件 / 布尔操作符

示意图

下面我将继续对RxJava2的其他操作符进行深入讲解，有兴趣可以继续关注Carson_Ho的安卓开发笔记

请点赞！因为你的鼓励是我写作的最大动力！

相关文章阅读

操作符使用
Android：这是一篇清晰 & 易懂的Rxjava 入门教程
 Android RxJava：最基础的操作符详解 - 创建操作符
 Android RxJava：图文详解变换操作符
 Android RxJava：组合 / 合并操作符详细教程
 Android RxJava：功能性操作符全面讲解

实际应用讲解
Android RxJava 实际应用讲解：（无条件）网络请求轮询
 Android RxJava 实际应用讲解：（有条件）网络请求轮询
 Android RxJava 实际应用讲解：网络请求嵌套回调
 Android RxJava 实际应用讲解：合并数据源
 Android RxJava 实际应用讲解：从磁盘 / 内存缓存中获取缓存数据
 Android RxJava 实际应用讲解：联合判断
 Android RxJava：细说线程控制（切换 / 调度）（含Retrofit实例讲解）
Android RxJava 实际应用讲解：网络请求出错重连（结合Retrofit）

欢迎关注Carson_Ho的简书！

不定期分享关于安卓开发的干货，追求短、平、快，但却不缺深度。

Android ：详解 RxJava 条件 / 布尔操作符

前言

目录

1. 作用

2. 类型

3. 具体操作符详解

3.1 all（）

3.2 exists（）

3.3 takeWhile（）

3.4 skipWhile（）

3.5 takeUntil（）

3.6 skipUntil（）

3.7 SequenceEqual（）

3.8 contains（）

3.9 isEmpty（）

3.10 amb（）

3.11 defaultIfEmpty（）

4. Demo地址

5. 总结

请点赞！因为你的鼓励是我写作的最大动力！

欢迎关注Carson_Ho的简书！