RxJava基础三-过滤操作符

作者: 清水杨杨 | 来源:发表于2019-05-02 16:41 被阅读0次

此篇内容均是来自书籍《RxJava响应式编程》李衍顺著

3.3 过滤操作符

过滤操作符的作用类似于SQL里的where语句，能够让Observable只发送出满足条件的数据。

3.3.1 debounce

debounce操作符是用来限流的，debounce过滤掉的数据会被丢弃掉。这个操作符又被分为throttleWithTimeout和debounce两个操作符。

throttleWithTimeout这个操作符通过时间来限流，每次接收到源Observable发送出来的数据后会进行计时，如果在设定好的时间结束前又接收到源Observable发送的新数据，那么上一个数据就会被丢弃，同时重新开始计时。如果每次都是在计时结束前接收到新数据，那么这个限流就会走向极端：只发送最后一个数据

下面我们来创建一个Observable，会每隔100毫秒发送一个数据，当要发送的数据是3的倍数时，下一个数据就延迟到300毫秒后发送。我们使用throttleWithTimeout来过滤一下这个Observable，设定过滤时间为200毫秒，也就是说发送间隔小于200毫秒的数据会被过滤掉。

private void debounce(){
        Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                for(int i=0; i<10; i++){
                    if(!subscriber.isUnsubscribed()){
                        subscriber.onNext(i);
                    }
                    int sleep = 100;
                    if(i%3==0){
                        sleep = 300;
                    }
                    try {
                        Thread.sleep(sleep);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.computation())
                .throttleWithTimeout(200, TimeUnit.MILLISECONDS)
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        System.out.println("throttleWithTimeout: " + integer);
                    }
                });

        Observable.interval(1000, TimeUnit.MILLISECONDS)
                .debounce(new Func1<Long, Observable<Long>>() {
                    @Override
                    public Observable<Long> call(Long aLong) {
                        return Observable.timer(aLong%2*1500, TimeUnit.MILLISECONDS);
                    }
                }).subscribe(new Action1<Long>() {
            @Override
            public void call(Long aLong) {
                System.out.println("debounce: " + aLong);
            }
        });
    }

但是发现结果什么都没有输出，不知道是哪里写错啦，需要进一步研究。

3.3.2 distinct

distinct操作符的用处就是去重，非常好理解。所有重复数据都会被过滤掉，其作用范围是全局的，也就是说所有发送出去的数据肯定是没有重复的。还要一个特殊的实现操作符：distinctUntilChanged，这个操作符被用来过滤掉连续的重复数据，作用范围是局部的，也就是说发送出来的数据可能会有重复，但是这些重复的数据肯定不是连续的。

private void distinct() {
        Observable.just(1,2,3,4,5,4,3,2,1)
                .distinct()
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        System.out.println("distinct: "+ integer);
                    }
                });
        Observable.just(1,2,3,3,3,1,2)
                .distinctUntilChanged()
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        System.out.println("untilchanged: "+ integer);
                    }
                });
    }

distinct: 1
distinct: 2
distinct: 3
distinct: 4
distinct: 5
untilchanged: 1
untilchanged: 2
untilchanged: 3
untilchanged: 1
untilchanged: 2

3.3.3 elementAt

elementAt只会过滤出源Observable发送出来的顺序为N的数据。同数组的下标一样，这里的N也是从0开始的，也就是说如果N为0，将只会过滤出源Observable发送的第一个数据。

private void elementAt(){
        Observable.just(2,4,4,2,5)
                .elementAt(2)
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        System.out.println("elementAt: "+ integer);
                    }
                });
    }
结果：
elementAt: 4

3.3.4 filter

filter操作符根据一个函数来进行过滤操作。源Observable发送的数据作为参数传递到函数里，如果函数的返回值为true，则这个数据将继续被发送出去，否则数据将会被过滤掉。

private void filter(){
        Observable.just(0,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: "+ integer);
            }
        });
    }
结果为：
Filter: 0
Filter: 1
Filter: 2

3.3.5 first 和 last

first操作符只会返回第一条或者满足条件的第一条数据。last操作符只会返回最后一条或者满足条件的最后一条数据。

另外还有一个特殊的Observable：BlockingObservable，可以使用Observable.toBlocking或者BlockingObservable.from方法来将一个Observable对象转化为BlockingObservable对象。BlockingObservable不会对Observable做任何处理，只会阻塞住。只有当满足条件的数据发送出来的时候才会发送第一个BlockingObservable对象。BlockingObservable也有可以配合使用的first操作符，不过其first操作符返回的不是Observable，而是对应的数据。

private void first_last(){
        Observable.just(1,2,3,4,5)
                .first(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("First: " + integer);
            }
        });
        int result = Observable.just(3,4,5,6,7)
                .toBlocking()
                .first(new Func1<Integer, Boolean>() {
                    @Override
                    public Boolean call(Integer integer) {
                        return integer > 4;
                    }
                });
        System.out.println("blocking: "+ result);
    }
结果
First: 4
blocking: 5

3.3.6 skip和take， skipLast 和takeLast

skip操作符能够将源Observable发送的数据过滤掉前n项，与之相反，take操作符则只取前n项。另外还有skipLast和takeLast，分别是从后面进行相应的过滤操作。

private void skip_take(){
        Observable.just(1,2,3,4,5)
                .skip(2)
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        System.out.println("skip: "+ integer);
                    }
                });
        Observable.just(2,3,4,5,6)
                .take(3)
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        System.out.println("take: "+ integer);
                    }
                });
    }
结果：
skip: 3
skip: 4
skip: 5
take: 2
take: 3
take: 4

3.3.7 sample 和 throttleFirst

sample操作符会指定一段时长，在每段时长结束的时候发送源Observable发送的最新数据，其它的都会被过滤掉。也就是说在sample指定的这段时间内无论源Observable发送了多少数据，只有最后一个数据才能通过sample的过滤。

sample是取规定时间段内的最后一个数据，而throttleFirst操作符取的则是规定时间段内的第一个数据，其它的会被过滤掉。在Android开发中，经常会有过滤掉过多的点击事件的需求，使用throttleFirst就可以很容易地将过多的点击事件过滤掉。

private void sample_throttleFirst(){
        Observable.interval(0,200, TimeUnit.MILLISECONDS)
                .sample(1000, TimeUnit.MILLISECONDS)
                .subscribe(new Action1<Long>() {
                    @Override
                    public void call(Long aLong) {
                        System.out.println("sample: "+ aLong);
                    }
                });

        Observable.interval(0,200, TimeUnit.MILLISECONDS)
                .throttleFirst(1000, TimeUnit.MILLISECONDS)
                .subscribe(new Action1<Long>() {
                    @Override
                    public void call(Long aLong) {
                        System.out.println("throttleFirst: "+ aLong);
                    }
                });

    }
结果：
throttleFirst: 0
是的， 第一个程序的结果没有输出，interval究竟是怎么用的呢，好像这本书的用法不对，也有可能是因为更新了库，这个后面再研究吧。

RxJava基础三-过滤操作符

3.3 过滤操作符

3.3.1 debounce

3.3.2 distinct

3.3.3 elementAt

3.3.4 filter

3.3.5 first 和 last

3.3.6 skip和take， skipLast 和takeLast

3.3.7 sample 和 throttleFirst

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读