RxJava2过滤操作符篇

上次我们讲了RxJava2中的变换操作符,这次我们来看一下过滤操作符

<br />

Filter

Filter,顾名思义,过滤器,可以过滤掉一部分不符合要求的事件,当上游给我们发送的数据超多,而下游需要的只是一些特定的数据,如果全部接收上游发送的数据,很容易造成OOM,为了避免OOM的出现,我们则需要对上游数据进行过滤,具体操作如下:

    Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                for (int i = 0; i < 10000; i++) {
                    e.onNext(i);
                }
            }
        }).observeOn(Schedulers.io())
                .subscribeOn(AndroidSchedulers.mainThread())
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer % 7 == 0;
                    }
                }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e("XYK",integer + "");
            }
        });
    }

在上面的代码中,我们朝下游发送了10000个数据,而我只需要其中可以被7整除的数据,利用filter,将其他的数据过滤出去,留下需要的数据.
<br />

图解

(发现一个神奇的网站,可以生成这种图,就不用自己画了,我才不会承认是我懒! http://rxmarbles.com/)

Filter方法使我们经常用到的一个过滤方法,基本已经可以满足大部分应用场所了,最常见的是过滤一些null对象,但是除此之外,还有一些其他的过滤方法,我们也来看下.

Sample

Sample,样品,其功能也是,sample会每隔一段时间对上游数据进行取样,发送到下游,但是这样会导致丢失了大量事件,比较适合特定场合,如对一组数中进行抽样,代码如下:

   Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                for (int i = 0; ; i++) {
                    e.onNext(i);
                }
            }
        }).sample(1,TimeUnit.SECONDS)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e("XYK",integer + "");
                    }
                });
    }

在上边的代码中,使用sample之后,每隔1秒对上游数据采样一次,发送到下游,其他事件则被过滤.

<br />

take/takeList

take和takeLast方法可以将上游事件中的前N项或者最后N项发送到下游,其他事件则进行过滤,代码如下:

 Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                for (int i = 0; ; i++) {
                    e.onNext(i);
                }
            }
        }).take(3)
        //.takeLast(3)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e("XYK",integer + "");
                    }
                });
    }

<br />

distinct

distinct方法,可以将重复对象去除重复对象,这里我们要用到一个方法,repeat(),产生重复事件,这里重复事件,再去除有些多余,只作为一个例子来展示.

Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                for (int i = 0;i < 50; i++) {
                    e.onNext(i);
                }
            }
        }).take(3)
                //生成重复事件
                .repeat(3)
                .distinct()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e("XYK",integer + "");
                    }
                });

<br />

小结

过滤操作符就介绍到这里,除此之外还有很多过略操作符,如first,last,都比较简单,大家自行尝试即可,下期预告,RxJava2操作符之组合操作符.