3.6 条件 / 布尔操作符

3.6.1. 作用

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

3.6.2. 类型

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

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

3.6.3. 具体操作符详解

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

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

3.6.3.1 all（）

作用

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

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

具体代码

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

测试结果

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

3.6.3.2 takeWhile（）

作用

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

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

具体代码

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

测试结果

3.6.3.3 skipWhile（）

作用

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

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

具体使用

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

测试结果

3.6.3.4 takeUntil（）

作用

执行到某个条件时，停止发送事件具体使用

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

测试结果

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

// （原始）第1个Observable：每隔1s发送1个数据 = 从0开始，每次递增1Observable.interval(1, TimeUnit.SECONDS)// 第2个Observable：延迟5s后开始发送1个Long型数据.takeUntil(Observable.timer(5, TimeUnit.SECONDS)).subscribe(new Observer<Long>() {@Overridepublic void onSubscribe(Disposable d) {Log.d(TAG, "开始采用subscribe连接");}@Overridepublic void onNext(Long value) {Log.d(TAG, "接收到了事件"+ value );}@Overridepublic void onError(Throwable e) {Log.d(TAG, "对Error事件作出响应");}@Overridepublic void onComplete() {Log.d(TAG, "对Complete事件作出响应");}});

测试结果

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

3.6.3.5 skipUntil（）

作用

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

具体使用

// （原始）第1个Observable：每隔1s发送1个数据 = 从0开始，每次递增1Observable.interval(1, TimeUnit.SECONDS)// 第2个Observable：延迟5s后开始发送1个Long型数据.skipUntil(Observable.timer(5, TimeUnit.SECONDS)).subscribe(new Observer<Long>() {@Overridepublic void onSubscribe(Disposable d) {Log.d(TAG, "开始采用subscribe连接");}@Overridepublic void onNext(Long value) {Log.d(TAG, "接收到了事件"+ value );}@Overridepublic void onError(Throwable e) {Log.d(TAG, "对Error事件作出响应");}@Overridepublic void onComplete() {Log.d(TAG, "对Complete事件作出响应");}});

测试结果

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

3.6.3.6 SequenceEqual（）

作用

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

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

具体使用

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

测试结果

3.6.3.7 contains（）

作用

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

若包含，返回true；否则，返回false内部实现 =exists（）

具体代码

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

测试结果

因为发送的数据中包含4

3.6.3.8 isEmpty（）

作用

判断发送的数据是否为空

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

具体代码

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

测试结果

因为发送的数据不为空

3.6.3.9 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>() {@Overridepublic void accept(Integer integer) throws Exception {Log.e(TAG, "接收到了事件 "+integer);}});

测试结果

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

3.6.3.10 defaultIfEmpty（）

作用

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

具体使用

Observable.create(new ObservableOnSubscribe<Integer>() {@Overridepublic void subscribe(ObservableEmitter<Integer> e) throws Exception {// 不发送任何有效事件// e.onNext(1);// e.onNext(2);// 仅发送Complete事件e.onComplete();}}).defaultIfEmpty(10) // 若仅发送了Complete事件，默认发送 值 = 10.subscribe(new Observer<Integer>() {@Overridepublic void onSubscribe(Disposable d) {Log.d(TAG, "开始采用subscribe连接");}@Overridepublic void onNext(Integer value) {Log.d(TAG, "接收到了事件"+ value );}@Overridepublic void onError(Throwable e) {Log.d(TAG, "对Error事件作出响应");}@Overridepublic void onComplete() {Log.d(TAG, "对Complete事件作出响应");}});

测试结果

3.6.4. 总结

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