RxJava3.0 操作符之过滤操作符使用

过滤操作符可以有选择地从Observable中发射数据.

过滤操作符分类

Filter — 过滤,仅仅从源Observable中发射那些通过了谓词测试的项,过滤掉未通过测试的项
Skip/Skiplast — 跳过源Observable 中的前/后 n项
ElementAt — 仅发射源Observable 指定位置 n 的项
First/Last — 只发射源Observable中的第一/最后一项
Take/TakeLast — 只发射源Observable中的前/后 n项
Distinct — 去重,过滤掉重复的发射项,相同的项只能发射一次
IgnoreElements — 不发射任何项,只保留终止通知(onError/onCompleted)
Debounce — 仅当特定时间跨度已过而没有发出另一个项目时，才从可观察对象发出项目,中间时间的项目都将被丢弃
Sample— 定期发射最新的数据，等于是数据抽样

过滤操作符的使用 Filter

filter操作符

filter 操作符通过指定的一个谓词函数测试项测试源Observable的数据项,只筛选出通过测试的数据项进行发射

例: 输入一串整数,筛选出为偶数的数

    private void filterOperator() {
        Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).filter(new Predicate() {
            @Override
            public boolean test(Integer integer) throws Throwable {
                return integer % 2 == 0;
            }
        }).subscribe(new Consumer() {
            @Override
            public void accept(Integer integer) throws Throwable {
                Log.i("sky>>>", "filter : " + integer);
            }
        });
    }

输出:
	8298-8298/com.sky.rxjava I/sky>>>: filter : 2
	8298-8298/com.sky.rxjava I/sky>>>: filter : 4
	8298-8298/com.sky.rxjava I/sky>>>: filter : 6
	8298-8298/com.sky.rxjava I/sky>>>: filter : 8
	8298-8298/com.sky.rxjava I/sky>>>: filter : 10

ofType 操作符

ofType 是filter 操作符一种特殊形式的存在,过滤筛选出指定类型的数据项进行发射

例: 一个抽象类 Animal,两个实现类Dog,Cat 源数据发射类型的数据,我们需要只筛选出Dog类型的数据进行观察

    //抽象类
	public abstract class Animal {
        public String name;

        public abstract void speak();
    }

    public class Dog extends Animal {
        public Dog(String name) {
            this.name = name;
        }

        @Override
        public void speak() {
            Log.i("sky>>>", name + " speak : 汪!");
        }
    }

    public class Cat extends Animal {
        public Cat(String name) {
            this.name = name;
        }

        @Override
        public void speak() {
            Log.i("sky>>>", name + " speak : 瞄!");
        }
    }



	  //根据类型去转换
    private void ofTypeOperator() {
        //创建狗类
        Dog d1 = new Dog("Dog-Jack1");
        Dog d2 = new Dog("Dog-Jack2");
        Dog d3 = new Dog("Dog-Jack3");
        Dog d4 = new Dog("Dog-Jack4");
		
        //创建猫类
        Cat c1 = new Cat("Cat-Tom1");
        Cat c2 = new Cat("Cat-Tom2");
        Cat c3 = new Cat("Cat-Tom3");
        Cat c4 = new Cat("Cat-Tom4");
		
        Animal a1 = new Dog("AniDog - dd");
        Animal a2 = new Cat("AniCat - cc");


        Observable.just(a1, a2, d1, d2, d3, d4, c1, c2, c3, c4).ofType(Dog.class).subscribe(new Consumer() {
            @Override
            public void accept(Dog dog) throws Throwable {
                dog.speak();
            }
        });
    }

	输出 :
        8298-8298/com.sky.rxjava I/sky>>>: AniDog - dd speak : 汪!
        8298-8298/com.sky.rxjava I/sky>>>: Dog-Jack1 speak : 汪!
        8298-8298/com.sky.rxjava I/sky>>>: Dog-Jack2 speak : 汪!
        8298-8298/com.sky.rxjava I/sky>>>: Dog-Jack3 speak : 汪!
        8298-8298/com.sky.rxjava I/sky>>>: Dog-Jack4 speak : 汪!

Skip/SkipLast

跳过源Observable数据项的前/后 n项再开始发射数据,skip 与skipLast使用方式一样,这里就一起介绍了

例:

    private void skipOperator() {
        //设置跳过前3项,所以此处输出4,5
        Observable.just(1, 2, 3, 4, 5).skip(3).subscribe(new Consumer() {
            @Override
            public void accept(Integer integer) throws Throwable {
                Log.i("sky>>>", "skip: " + integer);
            }
        });
         //设置跳过后2项,所以此处输出1,2,3
        Observable.fromArray(1,2,3,4,5).skipLast(2).subscribe(new Consumer() {
            @Override
            public void accept(Integer integer) throws Throwable {
                Log.i("sky>>>", "skipLast: " + integer);
            }
        });
    }

输出 :
	8298-8298/com.sky.rxjava I/sky>>>: skip: 4
	8298-8298/com.sky.rxjava I/sky>>>: skip: 5

    8298-8298/com.sky.rxjava I/sky>>>: skip: 1
    8298-8298/com.sky.rxjava I/sky>>>: skip: 2
    8298-8298/com.sky.rxjava I/sky>>>: skip: 3

skip/skipLast 重载

skip(long time, @NonNull TimeUnit unit)

skipLast(long time, @NonNull TimeUnit unit)

过滤掉源Observable 前/后指定时间内发射的数据项. 因为这里需要一些时间间隔才能模拟,所以使用intervalRange操作符创建一个可延迟间隔时间发送的Observable.

    private void skipOperator() {
        
        Observable.intervalRange(0, 10, 1, 1, TimeUnit.SECONDS)
                .skip(3, TimeUnit.SECONDS)
                .subscribe(new Consumer() {
                    @Override
                    public void accept(Long l) throws Throwable {
                        Log.i("sky>>>", "skip2 : " + l);
                    }
                });  
    }

输出 :
    15:37:53.635 31123-31166/com.sky.rxjava I/sky>>>: skip Time : 3
    15:37:54.635 31123-31166/com.sky.rxjava I/sky>>>: skip Time : 4
    15:37:55.635 31123-31166/com.sky.rxjava I/sky>>>: skip Time : 5
    15:37:56.635 31123-31166/com.sky.rxjava I/sky>>>: skip Time : 6
    15:37:57.635 31123-31166/com.sky.rxjava I/sky>>>: skip Time : 7
    15:37:58.634 31123-31166/com.sky.rxjava I/sky>>>: skip Time : 8
    15:37:59.634 31123-31166/com.sky.rxjava I/sky>>>: skip Time : 9

由上面输出可见, 源数据发射确实是每隔一秒发射一项,由于跳过了前三秒的数据,所以输出是从3开始.下面skipLast同理

		
        Observable.intervalRange(0, 9, 1, 1, TimeUnit.SECONDS)
                .skipLast(5, TimeUnit.SECONDS)
                .subscribe(new Consumer() {
                    @Override
                    public void accept(Long aLong) throws Throwable {
                        Log.i("sky>>>", "skipLast2 : " + aLong);
                    }
                });

输出:
    15:40:24.640 31718-31764/com.sky.rxjava I/sky>>>: skipLast Time : 0
    15:40:25.640 31718-31764/com.sky.rxjava I/sky>>>: skipLast Time : 1
    15:40:26.640 31718-31764/com.sky.rxjava I/sky>>>: skipLast Time : 2
    15:40:27.639 31718-31764/com.sky.rxjava I/sky>>>: skipLast Time : 3

查看skip/skipLast源码得知

skip(long time,TimeUnit unit) 其实调用的是skipUntil,它默认在 Schedulers.computation()这个调度器上执行,所以skip 还有一个另一个重载函数,第三个参数传递我们指定的调度器,这里就不再多做介绍了.

skipLast(long time, @NonNull TimeUnit unit)默认在Schedulers.trampoline()这个调度器上执行.所以,它也有一个重载函数让用户指定执行的调度器.

ElementAt

该运算符在Observable发射的项目序列中提取位于指定索引位置的项目，并将该项目作为其自身的唯一发射发射

    private void elementAt(){
        //根据索引取第3项 发射
        Observable.just(0,1,2,3,4,5).elementAt(3).subscribe(new Consumer() {
            @Override
            public void accept(Integer integer) throws Throwable {
                Log.i("sky>>>","elementAt : "+ integer);
            }
        });
    }

输出 :
	28023-28023/com.sky.rxjava I/sky>>>: elementAt : 3

First/Last

仅发送源Observable 的第一项/最后一项数据,由于操作类似,此处就一块介绍.

first

last

first/last 操作符接收一个参数defaultItem 作为默认item,如果发射器只发射了onComplete(),那么设置的这条默认数据句会作为源数据项发射出去,

first发射一次之后就会调用 canceled/disposed,所以发射onNext(),onComplete()之后在发送onError()就会报错.

而last却相反,无论发射器发射所少次onNext() 观察者都不会收到数据,只有当 onComplete()后,才会将最后一条数据发出,如果没有最后一条数据,就将默认的数据发出.而发出onError()后,无论之前有多少次onNext(),数据都会被丢弃,

    private void firstLastOperator() {
        Observable.create(new ObservableOnSubscribe() {
            @Override
            public void subscribe(@NonNull ObservableEmitter emitter) throws Throwable {
                //onNext() onComplete() 只要发送一条,就会调用 canceled/disposed,所以后面发的就无效了,继续发送onError,因为检测机制,会导致抛异常
//                emitter.onNext("a");
//                emitter.onNext("b");
//                emitter.onNext("c");
               	  emitter.onComplete();
//                emitter.onError(new Throwable("抛出异常"));
            }
        }).first("hello").subscribe(new Consumer() {
            @Override
            public void accept(String s) throws Throwable {
                Log.i("sky>>>", "first: " + s);
            }
        }, new Consumer() {
            @Override
            public void accept(Throwable throwable) throws Throwable {
                Log.i("sky>>>", "first error: " + throwable.getMessage());
            }
        });

        Observable.create(new ObservableOnSubscribe() {
            @Override
            public void subscribe(@NonNull ObservableEmitter emitter) throws Throwable {
                //再没有 onComplete()之前,无论调用多少次onNext(),数据都不会被发送出去
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                //只有调用onComplete(),才会将上面onNext最后一条发出,去过没有,就会使用 last(in defaultItem)里面传递的defaultItem作为数据发送.
                //emitter.onComplete();
                //此时,发送OnError,会丢弃之前所有数据
                emitter.onError(new Throwable("last 出错了"));
            }
        }).last(1).subscribe(new Consumer() {
            @Override
            public void accept(Integer integer) throws Throwable {
                Log.i("sky>>>", "last: " + integer);
            }
        }, new Consumer() {
            @Override
            public void accept(Throwable throwable) throws Throwable {
                Log.i("sky>>>", "last: " + throwable.getMessage());
            }
        });

    }

first,last 还有一些其他的操作符,他们都是可以取第一个或最后一个元素,但是又一一点不同:

first/last 操作符返回的是 Single类型可观察者.Single的行为与Observable类似，只是它只能发出一个成功的值或一个错误（与Observable不同，没有完成的通知).

firstOrError:取第一个元素. 这个操作符调用返回的也是一个Single类型的可观察者,与first区别是,再没有发射任何数据项时,直接调用onComplete(),此操作符会发出onError()事件,而带默认值的会将操作符作为第一项发出.

lastOrError:取最后一个元素. 同样是返回Single类型的可观察者,只能发出成功值或者错误的值.与last区别也是,直接调用onComplete()时,有默认值的会发出默认值,而此操作符会发出错误通知onError()

firstElement: 取第一个元素(内部其实调用elementAt(0)).返回值是 Maybe 类型的可观察者.Maybe 类型观察者可发出没有数据,一个数据,或者错误,与firstOrError/first带默认值又不同的是,他可以返回onComplete(),所以不存在另外两个在发onComplete()时,判断默认值去最终发默认数据还是onError()通知的操作.

lastElemet: 取最后一个元素.返回也是Maybe,他发射的数据项需要在发出onComplete()后才能做判断,有数据项就取最后一项发出,没有就发出onComplete()通知.

Take/TakeLast

take操作符

发射源Observable 的前n项数据

    private void takeOperator(){
        //输出前两项
        Observable.just(1,2,3,4,5).take(2).subscribe(new Consumer() {
            @Override
            public void accept(Integer integer) throws Throwable {
                Log.i("sky>>>", "take: " + integer);
            }
        });
    }

takeLast操作符

发射源Observable 的后n项数据

    private void takeLastOperator(){
        //输出后两项
        Observable.just(1,2,3,4,5).takeLast(2).subscribe(new Consumer() {
            @Override
            public void accept(Integer integer) throws Throwable {
                Log.i("sky>>>", "take: " + integer);
            }
        });
    }

take/与takeLast 调用方式看起来差不多,其实内部实现是不一样的

	//take 操作符
	public final Observable take(long count) {
        if (count < 0) {
            throw new IllegalArgumentException("count >= 0 required but it was " + count);
        }
        return RxJavaPlugins.onAssembly(new ObservableTake<>(this, count));
    }
	//takeLast操作符
	public final Observable takeLast(int count) {
        if (count < 0) {
            throw new IllegalArgumentException("count >= 0 required but it was " + count);
        }
        if (count == 0) {
            return RxJavaPlugins.onAssembly(new ObservableIgnoreElements<>(this));
        }
        if (count == 1) {
            return RxJavaPlugins.onAssembly(new ObservableTakeLastOne<>(this));
        }
        return RxJavaPlugins.onAssembly(new ObservableTakeLast<>(this, count));
    }

take 最终返回了ObservableTake ,而ObservableTake再订阅时,实际是订阅了他的内部类TakeObserver,TakeObserver内部有一个计数器 long remaining,就是我们传进来的取值个数,每次再Observable onNext()时候,都会减一,知道等于0时,发送onComplete()通知,后面的就不在发送,最终实现了取前几项的操作.

而takeLast相对就比较复杂,首先会判断要取的个数

个数为0 时,最终订阅的是ObservableIgnoreElements的内部类IgnoreObservable,而这个IgnoreObservable的onNext()方法是空实现,最终取个数为0项时,他不会发出任何数据项,只会发出onComPlete()和onError()通知.

个体数为1时,最终订阅ObservableTakeLastOne的内部类TakeLastOneObserver,它在收到源Observable发射数据项时,不会去发出onNext(),只是将数据项存在一个 value变量里,等到最终onComplete时,采取取value判断,有value 就发射value,并最终发出onComplete()通知.

个数大于1时,最终定于TakeLastObserver,内部有count变量存储我们传递进来的个数,有一个队列用来存储源Observable发射的数据项.

当收到新的数据时,先判断队列的大小,如果和count相等,就需要poll()将队列第一个数据出队,然后offer()将新数据入队,最终再源Observable发出onComplete时,循环取出所有数据,依次发出,最终再通知onComplete(),由此可以看出,他的数据一定是在源Observable所有数据发射完成之后,才会取出再往下游发射.

take/takelast重载

//take重载 只发射再指定时间内的数据
public final Observable take(long time, @NonNull TimeUnit unit) {
    return takeUntil(timer(time, unit));
}
//takeLast重载  只发射最后指定时间内的数据
public final Observable takeLast(long count, long time, @NonNull TimeUnit unit) {
        return takeLast(count, time, unit, Schedulers.trampoline(), false, bufferSize());
}

Distinct

去重,只发射未发射过得数据项

    //直接根据原数据去重
	private void distinctOperator(){
        Observable.just(1,2,3,3,4,5,5,6).distinct().subscribe(new Consumer() {
            @Override
            public void accept(Integer integer) throws Throwable {
                Log.i("sky>>>", "distinct1: " + integer);
            }
        });
        
 输出: 
        10273-10273/com.sky.rxjava I/sky>>>: distinct1: 1
        10273-10273/com.sky.rxjava I/sky>>>: distinct1: 2
        10273-10273/com.sky.rxjava I/sky>>>: distinct1: 3
        10273-10273/com.sky.rxjava I/sky>>>: distinct1: 4
        10273-10273/com.sky.rxjava I/sky>>>: distinct1: 5
        10273-10273/com.sky.rxjava I/sky>>>: distinct1: 6
		
        //通过Function 函数接口给每项指定 一个key,通过key来判断是否相同,
        //这里我们指定偶数的key是"a".奇数key为自身转为字符串,输出是,奇数相同的被过滤,偶数全部被过滤,只留了第一个2    
        Observable.just(1,2,3,3,4,5,5,6,7).distinct(new Function() {
            @Override
            public String apply(Integer integer) throws Throwable {
                if(integer%2 ==0){
                    return "a";
                }
                return integer+"";
            }
        }).subscribe(new Consumer() {
            @Override
            public void accept(Integer integer) throws Throwable {
                Log.i("sky>>>", "distinct2: " + integer);
            }
        });
    }

输出:
        11093-11093/com.sky.rxjava I/sky>>>: distinct2: 1
        11093-11093/com.sky.rxjava I/sky>>>: distinct2: 2
        11093-11093/com.sky.rxjava I/sky>>>: distinct2: 3
        11093-11093/com.sky.rxjava I/sky>>>: distinct2: 5
        11093-11093/com.sky.rxjava I/sky>>>: distinct2: 7

IgnoreElements

不发出源Observable任何数据项,但可接收停止通知onComplete,onError

   public void ignoreElementsOperator(){
        Observable.create(new ObservableOnSubscribe() {
            @Override
            public void subscribe(@NonNull ObservableEmitter emitter) throws Throwable {
                //发射不作用,因为下面观察者根本不关心发射的数据,只关注onComplete,和onError
                emitter.onNext("aaaaa");
                emitter.onNext("bbbbbbb");
				//这两个只能发出一个
                //emitter.onComplete();
                //emitter.onError(new Throwable("出错了"));
            }
        }).ignoreElements().subscribe(new CompletableObserver() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onComplete() {

            }

            @Override
            public void onError(@NonNull Throwable e) {

            }
        });
    }

Debounce

过滤掉设定间隔时间内发送的数据项

例: 创建一个Observable,发送10个事件,每个事件发射前,先随机暂停一段时间,造成每次发射都有一个间隔时间,使用debounce设置源Observable2秒没有发送事件,就去发送事件.

由输出我们得知,首次发射数据,观察者肯定能收到数据,后面再次发射数据,如果数据和前次发射间隔小于我们设定的2秒,数据就会被过滤,不会发射,而当大于我们设定的2秒,此数据就会被发出,由观察者观察到.

    private void debounce(){
        Observable.create(new ObservableOnSubscribe() {
                    @Override
                    public void subscribe(@NonNull ObservableEmitter emitter) throws Throwable {
                        for (int i = 0; i < 10; i++) {
                            Thread.sleep(new Random().nextInt(5)*1000);
                            Log.e("sky>>>","emitter send "+ i);
                            emitter.onNext(i);
                        }
                        emitter.onComplete();
                    }
                })
                .debounce(2,TimeUnit.SECONDS)
                .subscribeOn(Schedulers.computation())
                .subscribe(new Consumer() {
                    @Override
                    public void accept(Integer i) throws Throwable {
                        Log.i("sky>>>", "debounce: " + i);
                    }
                });
    }

输出:
        19:13:45.221 19221-19263/com.sky.rxjava E/sky>>>: ###发射器发射 0
        19:13:47.267 19221-19272/com.sky.rxjava I/sky>>>: ---观察者接收: 0
        19:13:47.267 19221-19263/com.sky.rxjava E/sky>>>: ###发射器发射 1
        19:13:49.300 19221-19263/com.sky.rxjava E/sky>>>: ###发射器发射 2
        19:13:49.302 19221-19263/com.sky.rxjava E/sky>>>: ###发射器发射 3
        19:13:51.346 19221-19263/com.sky.rxjava E/sky>>>: ###发射器发射 4
        19:13:51.347 19221-19272/com.sky.rxjava I/sky>>>: ---观察者接收: 3
        19:13:52.389 19221-19263/com.sky.rxjava E/sky>>>: ###发射器发射 5
        19:13:53.399 19221-19263/com.sky.rxjava E/sky>>>: ###发射器发射 6
        19:13:55.405 19221-19263/com.sky.rxjava E/sky>>>: ###发射器发射 7
        19:13:57.410 19221-19272/com.sky.rxjava I/sky>>>: ---观察者接收: 7
        19:13:59.450 19221-19263/com.sky.rxjava E/sky>>>: ###发射器发射 8
        19:14:01.493 19221-19272/com.sky.rxjava I/sky>>>: ---观察者接收: 8
        19:14:03.494 19221-19263/com.sky.rxjava E/sky>>>: ###发射器发射 9
        19:14:03.497 19221-19263/com.sky.rxjava I/sky>>>: ---观察者接收: 9

Sample

取样,定期n秒发射源Observable最近一次发射的数据

例:随机时间发射20条数据 ,由输出结果得知.在源数据发射每3秒时,才会接收最近一次发射的数据

    private void sampleOperator() {
        Observable.create(new ObservableOnSubscribe() {
                    @Override
                    public void subscribe(@NonNull ObservableEmitter emitter) throws Throwable {
                        for (int i = 0; i < 20; i++) {
                           Thread.sleep(new Random().nextInt(4)*1000);
                            Log.w("sky>>>", "sample 发射: "+ i);
                           emitter.onNext(i);
                        }
                        emitter.onComplete();
                    }
                }).subscribeOn(Schedulers.newThread())
                .sample(3, TimeUnit.SECONDS)

                .subscribe(new Consumer() {
                    @Override
                    public void accept(Integer i) throws Throwable {
                        Log.i("sky>>>", "sample: "  + i);
                    }
                });
    }

输出 :
	10:01:16.138 4633-4661/com.sky.rxjava W/sky>>>: sample 发射: 0
    10:01:17.142 4633-4661/com.sky.rxjava W/sky>>>: sample 发射: 1
    10:01:17.143 4633-4661/com.sky.rxjava W/sky>>>: sample 发射: 2
    10:01:18.144 4633-4661/com.sky.rxjava W/sky>>>: sample 发射: 3
    10:01:19.107 4633-4660/com.sky.rxjava I/sky>>>: sample: 3
    10:01:21.146 4633-4661/com.sky.rxjava W/sky>>>: sample 发射: 4
    10:01:22.103 4633-4660/com.sky.rxjava I/sky>>>: sample: 4
    10:01:22.156 4633-4661/com.sky.rxjava W/sky>>>: sample 发射: 5
    10:01:23.156 4633-4661/com.sky.rxjava W/sky>>>: sample 发射: 6
    10:01:25.102 4633-4660/com.sky.rxjava I/sky>>>: sample: 6
    10:01:25.157 4633-4661/com.sky.rxjava W/sky>>>: sample 发射: 7
    10:01:28.101 4633-4660/com.sky.rxjava I/sky>>>: sample: 7
    10:01:28.160 4633-4661/com.sky.rxjava W/sky>>>: sample 发射: 8
    10:01:29.164 4633-4661/com.sky.rxjava W/sky>>>: sample 发射: 9

sample还有两个变体:

throttleFirst: 与smaple 区别,定期取间隔时间内,源Observable 发射的第一条数据.

throttleLast:与sample 结果一致

end

RxJava 3.0 过滤操作符总结完毕.

RxJava3.0 操作符之过滤操作符使用

Java相关栏目本月热门文章