RxJava使用详解--过滤操作符

时间:2022-06-12 17:49:09

RxJava使用详解系列文章

RxJava使用详解--创建操作符

RxJava使用详解--转换操作符


详细的例子可以查看文章末尾的源码

RxJava使用详解--过滤操作符

这篇文章主要讲RxJava中常见的过滤操作符


1.debounce操作符

源Observable每发射一个数据项,如果在debounce规定的间隔时间内Observable没有发射新的数据项,debounce就把这个结果提交给订阅者处理,如果在规定的间隔时间内产生了其他结果,就忽略掉发射的这个数据,通过制定的时间间隔来限流,可以过滤掉发射速率过快的数据项,默认在computatiion调度器上执行,可以指定执行线程。

注意:如果源Observable发射最后一个数据后,在debounce规定的时间间隔内调用了onCompleted,那么通过debounce操作符就把这个结果提交给订阅者
 throttleWithTimeOut使用也是调用了debounce操作符来实现

RxJava使用详解--过滤操作符

Observable.create(new Observable.OnSubscribe<Integer>() {
    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        if (subscriber.isUnsubscribed()) return;//如果没有订阅者就直接返回
        try {
            //发射数据的时间间隔:100~900毫秒
            for (int i = 0; i < 10; i++) {
                subscriber.onNext(i);
                Thread.sleep(i * 100);
            }
            subscriber.onCompleted();
        } catch (InterruptedException e) {
            subscriber.onError(e);
        }
    }
}).subscribeOn(Schedulers.newThread())
        .debounce(400, TimeUnit.MILLISECONDS)//超时时间为400毫秒,预期结果:时间间隔在400毫秒以上的数据都会提交给订阅者,其他的不会。
        .subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer integer) {
                System.out.println("debounce onNext:" + integer + " 所在线程:" + Thread.currentThread().getName());
            }
        });
输出结果:

12-19 15:22:26.910 16572-16622/com.dingmouren.rxjavademo I/System.out: debounce onNext:4 所在线程:RxComputationScheduler-1
12-19 15:22:27.310 16572-16622/com.dingmouren.rxjavademo I/System.out: debounce onNext:5 所在线程:RxComputationScheduler-1
12-19 15:22:27.811 16572-16622/com.dingmouren.rxjavademo I/System.out: debounce onNext:6 所在线程:RxComputationScheduler-1
12-19 15:22:28.411 16572-16622/com.dingmouren.rxjavademo I/System.out: debounce onNext:7 所在线程:RxComputationScheduler-1
12-19 15:22:29.111 16572-16622/com.dingmouren.rxjavademo I/System.out: debounce onNext:8 所在线程:RxComputationScheduler-1
12-19 15:22:29.911 16572-16622/com.dingmouren.rxjavademo I/System.out: debounce onNext:9 所在线程:RxComputationScheduler-1


2.distinct()操作符:

过滤掉重复的数据项。过滤规则是只允许没有发射过的数据项通过。

变体distinct(Func1)根据返回的key值去过滤,不用数据本身.
distinctUntilChanged()只判断这个数据项跟前一个数据项是否相同,distinctUnitilChanged(Func1)也是根据返回的key值去比较过滤。
默认不在任何特定的调度器上执行。

Observable.just(1,2,2,3,4,4,5).distinct().subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer integer) {
        System.out.println("distinct() onNext:" + integer + " 所在线程:" + Thread.currentThread().getName());
    }
});
输出结果:

distinct() onNext:1 所在线程:main
distinct() onNext:2 所在线程:main
distinct() onNext:3 所在线程:main
distinct() onNext:4 所在线程:main
distinct() onNext:5 所在线程:main

RxJava使用详解--过滤操作符

Observable.just(1,2,2,3,4,4,5).distinct(new Func1<Integer, String>() {
    @Override
    public String call(Integer integer) {
        return 3 < integer ? "第一组" :"第二组";//这里返回key值,小于3的key值是第一组,也就是说1和2的key值都是第一组,只会将1提交给订阅者,2的key值与1相同就直接被过滤掉了,这个变体是根据key值进行过滤的
    }
}).subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer integer) {
        System.out.println("distinct(Func1) onNext:" + integer + " 所在线程:" + Thread.currentThread().getName());
    }
});
输出结果:

distinct(Func1) onNext:1 所在线程:main
distinct(Func1) onNext:4 所在线程:main

RxJava使用详解--过滤操作符

Observable.just(1,2,3,2,4,5,4).distinctUntilChanged().subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer integer) {
        System.out.println("distinctUntilChanged() onNext:" + integer + " 所在线程:" + Thread.currentThread().getName());
    }
});
输出结果:

distinctUntilChanged() onNext:1 所在线程:main
distinctUntilChanged() onNext:2 所在线程:main
distinctUntilChanged() onNext:3 所在线程:main
distinctUntilChanged() onNext:2 所在线程:main
distinctUntilChanged() onNext:4 所在线程:main
distinctUntilChanged() onNext:5 所在线程:main
distinctUntilChanged() onNext:4 所在线程:main

3.elementAt(index)

将指定索引的数据项提交给订阅者,索引是从0开始,当没有这个索引或者索引为负数会抛异常。

elementAtOrDefault(index,default):这个会设置一个默认值,当没有指定的索引就提交默认值给订阅者,为负数就抛异常。
Observable.just(1,2,3,4).elementAt(3).subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer integer) {
        System.out.println("elementAt onNext:" + integer + " 所在线程:" + Thread.currentThread().getName());
    }
});
RxJava使用详解--过滤操作符

Observable.just(1,2,3,4).elementAt(3).subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer integer) {
        System.out.println("elementAt onNext:" + integer + " 所在线程:" + Thread.currentThread().getName());
    }
});
输出结果:elementAt onNext:4 所在线程:main


RxJava使用详解--过滤操作符

Observable.just(1,2,3,4).elementAtOrDefault(6,6).subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer integer) {
        System.out.println("elementAtOrDefault onNext:" + integer + " 所在线程:" + Thread.currentThread().getName());
    }
});
输出结果:elementAtOrDefault onNext:6 所在线程:main

4. filter操作符对源Observable发射的数据项按照指定的条件进行过滤,满足的条件的才会调给订阅者。默认不在任何特定的调度器上执行

RxJava使用详解--过滤操作符

Observable.just(1,2,3,4,5).filter(new Func1<Integer, Boolean>() {
    @Override
    public Boolean call(Integer integer) {
        return integer > 3;
    }
}).subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer integer) {
        System.out.println("filter(Func1) onNext:" + integer + " 所在线程:" + Thread.currentThread().getName());
    }
});
输出结果:

filter(Func1) onNext:4 所在线程:main
filter(Func1) onNext:5 所在线程:main

5.first()操作符提交源Observable发射的第一项数据,如果只是想要一个过滤符,最好使用take(2)或者elementAt(0)

first(Func1)操作符是提交第一项符合自定义条件的数据
firstOrDefault(T)操作符是在Observable没有发射任何数据时提交一个指定的默认值
takeFirst(Func1)操作符提交符合自定义条件的的第一项数据,
与first(Func1)不同的是,takeFirst(Func1)在没有符合条件的时候,会调用onCompleted,而first(Func1)会抛一个NoSuchElementException的异常
RxJava使用详解--过滤操作符

Observable.just(1,2,3).first().subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer integer) {
        System.out.println("first() onNext:" + integer + " 所在线程:" + Thread.currentThread().getName());
    }
});
输出结果:first() onNext:1 所在线程:main

RxJava使用详解--过滤操作符

Observable.just(1,2,3).first(new Func1<Integer, Boolean>() {
    @Override
    public Boolean call(Integer integer) {
        return 1 < integer;
    }
}).subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer integer) {
        System.out.println("first(Func1) onNext:" + integer + " 所在线程:" + Thread.currentThread().getName());
    }
});
输出结果:first(Func1) onNext:2 所在线程:main

6.ignoreElements()操作符不提交任何数据给订阅者,只提交终止通知(onError或者onCompeleted)给订阅者,默认不在任何特定的调度器上执行

RxJava使用详解--过滤操作符

 

7.last()操作符与first()操作符相反,只提交最后一个数据项给订阅者,如果只是作为过滤操作符,最好使用takeLast(1),

官方文档解释说:first()操作符和last()操作符在某些实现中会返回一个阻塞函数。
与first()操作符系列对应,也有last(Func1)、lastOrDefault(T)、lastOrDefault(T,Func1)
RxJava使用详解--过滤操作符

8.ofType操作符类似于filter操作符,区别在于ofType按照数据项的类型进行过滤,默认不在任何特定的调度器上执行

RxJava使用详解--过滤操作符

Observable.just(1,"String类型",true).ofType(String.class).subscribe(new Action1<String>() {
    @Override
    public void call(String s) {
        System.out.println("ofType(class) onNext:" + s + " 所在线程:" + Thread.currentThread().getName());
    }
});
输出结果:ofType(class) onNext:String类型 所在线程:main

9.sample操作符对Observable发射的数据定时进行采样,采的是自从上一次采样以来,Observable最近发射的一项数据,也就是这段时间间隔中最后一个数据项。如果自上一次采样以来,源Observable没有发射任何数据,sample操作符返回的Observable在此段时间也不会发射任何数据

默认在computation调度器上执行,但是可以指定它执行的线程 sample(long,TimeUnit,Scheduler)
与之对应的操作符是throttleFirst,它采样的是采样时间间隔中第一项数据,在最后一个时间段会发射最后一个数据项,看下面例子
* △time = 2.2s
* Data :    1------2------3------4------5------6------7
* Time(s) : 0------1------2------3------4------5------6------7................
* △time点: 0------1------2-。----3------4-。---5------6----。-7................
* SampleResults:  3,5,7
* ThrottleFirst:  1,4,7
RxJava使用详解--过滤操作符

Observable.create(new Observable.OnSubscribe<Integer>() {
    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        if (subscriber.isUnsubscribed()) return;//如果没有订阅者直接返回
        try {
            //前3个数字的时间间隔设置1秒,最后一个设置2秒
            for (int i = 1; i <8 ; i++) {
                subscriber.onNext(i);
                Thread.sleep(1000);
            }
            subscriber.onCompleted();
        } catch (InterruptedException e) {
           subscriber.onError(e);
        }
    }
}).sample(2200, TimeUnit.MILLISECONDS)//采样间隔时间为2200毫秒
        .subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer integer) {
        System.out.println("sample  onNext:" + integer + " 所在线程:" + Thread.currentThread().getName());
    }
});
输出结果:

sample  onNext:3 所在线程:RxComputationScheduler-1
sample  onNext:5 所在线程:RxComputationScheduler-1
sample  onNext:7 所在线程:RxComputationScheduler-1


RxJava使用详解--过滤操作符


Observable.create(new Observable.OnSubscribe<Integer>() {

    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        if (subscriber.isUnsubscribed()) return;
        try {
            for (int i = 1; i < 9 ; i++) {
                subscriber.onNext(i);
                Thread.sleep(1000);
            }
            subscriber.onCompleted();
        } catch (InterruptedException e) {
            subscriber.onError(e);
        }
    }
}).throttleFirst(2200,TimeUnit.MILLISECONDS)
        .subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer integer) {
                System.out.println("throttleFirst  onNext:" + integer + " 所在线程:" + Thread.currentThread().getName());
            }
        });
输出结果:

throttleFirst  onNext:1 所在线程:main
throttleFirst  onNext:4 所在线程:main
throttleFirst  onNext:7 所在线程:main

 

10.single()操作符在源Observable只发射一个数据项的时候,single()操作符会将这个数据提交给订阅者,大于1个就抛Sequence contains too many elements的异常,不是正好是一个数据项就会抛异常

single(Func1)操作符是对源Observable发射的数据项进行判断,如果符合条件的数据数量大于1就会抛异常。不是正好是一个数据项就会抛异常
也有设置默认值得api,默认不在任何特定的调度器上执行
11.skip操作符
skip(count) 对于源Observable发射的数据项,跳过前count项,将后面的数据项提交给订阅者
skip(long,TimeUnit)对于原Observalbe发射的数据项,跳过long前的数据项,将之后的数据提交给订阅者,可以指定执行线程

 

Observable.just(1,2,3,4,5,6).skip(3).subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer integer) {
        System.out.println("skip(count) onNext:" + integer + " 所在线程:" + Thread.currentThread().getName());
    }
});
输出结果:

skip(count) onNext:4 所在线程:main
skip(count) onNext:5 所在线程:main
skip(count) onNext:6 所在线程:main

RxJava使用详解--过滤操作符

Observable.create(new Observable.OnSubscribe<Integer>() {
    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        if (subscriber.isUnsubscribed()) return;
        try {
            //每隔一秒发射一项数据
            for (int i = 1; i < 5 ; i++) {
                subscriber.onNext(i);
                Thread.sleep(1000);
            }
            subscriber.onCompleted();
        } catch (InterruptedException e) {
            subscriber.onError(e);
        }
    }
}).skip(2, TimeUnit.SECONDS, Schedulers.newThread()).subscribe(new Action1<Integer>() {//发射2秒之后的数据项
    @Override
    public void call(Integer integer) {
        System.out.println("skip(long,TimeUnit) onNext:" + integer + " 所在线程:" + Thread.currentThread().getName());
    }
});
输出结果:

skip(long,TimeUnit) onNext:3 所在线程:main
skip(long,TimeUnit) onNext:4 所在线程:main

12.skipLast操作符

skipLast(count) 对于源Observable发射的数据项,省略最后count项,将前面的数据项提交给订阅者
skipLast(long,TimeUnit)对于原Observalbe发射的数据项,省略最后long时间段的数据项,将之前的数据提交给订阅者,可以指定执行线程
RxJava使用详解--过滤操作符

RxJava使用详解--过滤操作符


13.take操作符

take(count)操作符对于源Observable发射的数据项,提取前面的count项数据提交给订阅者,忽略后面的
take(long,TimeUnit)操作符对于源Obsrvable发射的数据项,提取前面long时间段里的数据项提交给订阅者,忽略后面的,可以指定线程
RxJava使用详解--过滤操作符

RxJava使用详解--过滤操作符

14.takeLast操作符

takeLast(count)操作符对于源Observable发射的数据项,提取前面的count项数据提交给订阅者,忽略后面的
takeLast(long,TimeUnit)操作符对于源Obsrvable发射的数据项,提取前面long时间段里的数据项提交给订阅者,忽略后面的,可以指定线程
RxJava使用详解--过滤操作符

RxJava使用详解--过滤操作符


15.takeLastBuffer操作符

takeLastBuffer(count)操作符与takeLast(count)操作符类似,唯一不同就是takeLastBuffer(count)将最后的那些数据项收集到一个list集合中,提交这个集合给订阅者
takeLastBuffer(long,TimeUnit)操作符与takeLast(long,TimeUnit)操作符类似,唯一不同就是将在最后时间段Long中数据项收集到一个list集合中,
将这个集合提交给了订阅者,可以指定运行的线程。


更多详细内容和例子,可以查看源码