RxJava系列之常用过滤操作符_rxjava常用的过滤函数-CSDN博客

本文链接：https://blog.csdn.net/CSDN_Mew/article/details/103857990

本文深入解析RxJava中的过滤操作符，如filter、take、distinct和elementAt的使用场景与实现原理，通过实例演示如何高效筛选和处理数据流。

摘要生成于 C知道，由 DeepSeek-R1 满血版支持，前往体验 >

系列文章

过滤操作符

上篇文章中，我们学习了变换操作符。过滤操作符顾名思义就是对数据的筛选与过滤，与变换操作符的位置相同，作用不同。

filter

如果过滤操作符的方法返回false，则过滤掉该条数据，返回true则不过滤该条数据。

  /**
     * filter 过滤
     * 需求：过滤掉 哪些不合格的奶粉，输出哪些合格的奶粉
     * @param view
     */
    public void r01(View view) {

        // 上游
        Observable.just("三鹿", "合生元", "飞鹤")

        .filter(new Predicate<String>() {
            @Override
            public boolean test(String s) throws Exception {
                // return true; // 不去过滤，默认全部都会打印
                // return false; // 如果是false 就全部都不会打印

                if ("三鹿".equals(s)) {
                    return false; // 不合格
                }

                return true;
            }
        })

        // 订阅
        .subscribe(new Consumer<String>() { // 下游
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, "accept: " + s);
            }
        });
    }

take

take过滤操作符需要跟定时器配合使用，是定时器的停止条件。

/**
     * take过滤操作符
     * @param view
     */
    public void r02(View view) {
        // 定时器 运行   只有再定时器运行基础上 加入take过滤操作符，才有take过滤操作符的价值

        // 上游
        Observable.interval(2, TimeUnit.SECONDS)

                // 增加过滤操作符，停止定时器
                .take(8) // 执行次数达到8 停止下来

                .subscribe(new Consumer<Long>() { // 下游
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.d(TAG, "accept: " + aLong);
                    }
                });

    }

distinct

distinct过滤操作符可以过滤掉相同数据的，底层应该调用了equals方法判断两个对象是否相同。

 /**
     * distinct过滤重复事件
     * @param view
     */
    public void r03(View view) {
        // 上游
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onNext(4);
                e.onNext(4);
                e.onComplete();
            }
        })

        .distinct() // 过滤重复 发射的事件

        .subscribe(new Consumer<Integer>() { // 下游 观察者
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "accept: " + integer); // 事件不重复
            }
        });
    }

elementAt

elementAt过滤操作符只要指定索引置为的数据,其它数据都不要了

 /**
     * elementAt 指定过滤的内容
     * @param view
     */
    public void r04(View view) {
        // 上游
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("九阴真经");
                e.onNext("九阳真经");
                e.onNext("易筋经");
                e.onNext("神照经");
                e.onComplete();
            }
        })

        // 过滤操作符
        .elementAt(100, "默认经") // 指定下标输出 事件

        // 订阅
        .subscribe(new Consumer<String>() { // 下游
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, "accept: " + s);
            }
        });


    }