Android之Rxjava2.X 7————Rxjava 条件操作符

一. 目录

二. 概述

1.作用

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

2.类型

三. 具体操作符详解

1.all（）

作用：判定是否Observable发射的所有数据都满足某个条件

具体使用：

 private void rxJavaDemo10() {
        Observable.just(1, 2, 3, 4, 5).
                all(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                return integer <= 10;
            }
        }).subscribe(new Consumer<Boolean>() {
            @Override
            public void accept(Boolean aBoolean) throws Exception {
                Log.d(TAG, "accept: "+aBoolean);
            }
        });


    }

2.takeWhile（）

• 作用：发射Observable发射的数据，直到一个指定的条件不成立

原理图：

具体使用

  Observable.just(1, 2, 3, 4, 5)
                .takeWhile(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                         return (integer!=3);
                    }
                }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "accept: "+integer);

            }
        });

3.skipWhile（）

• 作用:丢弃Observable发射的数据，直到一个指定的条件不成立

原理图:

具体使用

Observable.just(1, 2, 3, 4, 5)
                .skipWhile(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                         return (integer!=3);
                    }
                }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "accept: "+integer);

            }
        });

4.takeUntil（）

• 作用1：takeUntil（new Predicate） 执行到某个条件时，停止发送事件
• 作用2：takeUntil（new Observer） takeUntil（） 传入的Observable开始发送数据，（原始）第1个Observable的数据停止发送数据

原理图


具体使用1:

 Observable.just(1, 2, 3, 4, 5)
                .takeUntil(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                         return (integer==3);
                    }
                }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "accept: "+integer);

            }
        });

具体使用2

Observable.interval(1, TimeUnit.SECONDS)
                // 第2个Observable：延迟5s后开始发送1个Long型数据
                .takeUntil(Observable.timer(5, TimeUnit.SECONDS))
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long integer) throws Exception {
                        Log.d(TAG, "accept: " + integer);

                    }
                });

5.skipUntil（）

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

原理图:

具体使用

Observable.interval(1, TimeUnit.SECONDS)
                // 第2个Observable：延迟5s后开始发送1个Long型数据
                .skipUntil(Observable.timer(5, TimeUnit.SECONDS))
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long integer) throws Exception {
                        Log.d(TAG, "accept: " + integer);

                    }
                });

6.SequenceEqual（）

• 作用：判定两个Observables是否发射相同的数据序列。若相同，返回 true；否则，返回 false

原理图：

具体使用：

 Observable
                .sequenceEqual(Observable.just(1,2,3,4),Observable.just(1,2,3,4))
                .subscribe(new Consumer<Boolean>() {
                    @Override
                    public void accept(Boolean integer) throws Exception {
                        Log.d(TAG, "accept: " + integer);

                    }
                });

7.contains（）

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

原理图：

具体使用

Observable.just(1, 2, 3, 4, 5, 6)
                .contains(4)
                .subscribe(new Consumer<Boolean>() {
                    @Override
                    public void accept(Boolean integer) throws Exception {
                        Log.d(TAG, "accept: " + integer);

                    }
                });

8.isEmpty（）

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

原理图：

具体使用

Observable.just(1, 2, 3, 4, 5, 6)
                .isEmpty()
                .subscribe(new Consumer<Boolean>() {
                    @Override
                    public void accept(Boolean integer) throws Exception {
                        Log.d(TAG, "accept: " + integer);

                    }
                });

9.amb（）

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

原理图：

具体使用

 List<ObservableSource<Integer>> list= new ArrayList <>();
        list.add( Observable.just(1,2,3).delay(1,TimeUnit.SECONDS));
        list.add( Observable.just(4,5,6));

        Observable.amb(list)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept: " + integer);

                    }
                });

10.defaultIfEmpty（）

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

原理图：

具体使用

  Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onComplete();
            }
        }).defaultIfEmpty(10)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept: " + integer);

                    }
                });

四. 参考资料

Rxjava中文文档
Android RxJava2(六)条件操作符
Android RxJava：详解 条件 / 布尔操作符

五.文章索引

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