Rx系列<第十二篇>：RxJava之过滤操作符

过滤操作符分为四类
1.指定条件过滤事件
2.指定事件数量过滤事件
3.指定时间过滤事件
4.指定事件位置过滤事件

指定条件过滤事件

（1）filter

过滤指定条件。

    Observable.just("1", "2", "3", "4", "5").filter(new Predicate<String>() {
        @Override
        public boolean test(String s) throws Exception {
            // 根据test()的返回值 对被观察者发送的事件进行过滤 & 筛选
            // a. 返回true，则继续发送
            // b. 返回false，则不发送（即过滤）
            if(s.equals("1") || s.equals("2")){
                return true;
            }
            return false;
        }
    }).subscribeOn(Schedulers.newThread()).subscribe(new Consumer<String>() {
        @Override
        public void accept(String s) throws Exception {
            System.out.println(s);
        }
    });

结果：返回1和2。

（2）ofType

过滤指定类型。

    Observable.just("A", "B", 1, 2).ofType(Integer.class)
            .subscribeOn(Schedulers.newThread()).subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println(String.valueOf(integer));
        }
    });

结果返回：1和2

（3）skip和skipLast

skip：从前往后跳过
skipLast：从后往前跳过

    Observable.just("A", "B", "1", "2").skip(3)
            .subscribeOn(Schedulers.newThread()).subscribe(new Consumer<String>() {
        @Override
        public void accept(String s) throws Exception {
            System.out.println(String.valueOf(s));
        }
    });

返回结果：2

    Observable.just("A", "B", "1", "2").skipLast(3)
            .subscribeOn(Schedulers.newThread()).subscribe(new Consumer<String>() {
        @Override
        public void accept(String s) throws Exception {
            System.out.println(String.valueOf(s));
        }
    });

返回结果：A

（4）distinct和distinctUntilChanged

distinct：过滤相同的数据
distinctUntilChanged：过滤连续相同的数据

    Observable.just(1, 2, 1, 2, 3, 3, 5)
            .distinct()
            .subscribeOn(Schedulers.newThread())
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(String.valueOf(integer));
                }
            });

返回结果：1 2 3 5

    Observable.just(1, 2, 1, 2, 3, 3, 5)
            .distinctUntilChanged()
            .subscribeOn(Schedulers.newThread())
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(String.valueOf(integer));
                }
            });

返回结果：1 2 1 2 3 5

指定数量过滤事件

（1）take和takeLast

take：观察者只能接收到前n个数据
takeLast：观察者只能接收到后n个数据

    Observable.just(1, 2, 3, 4, 5)
            .take(2)
            .subscribeOn(Schedulers.newThread())
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(String.valueOf(integer));
                }
            });

返回结果：1 2

    Observable.just(1, 2, 3, 4, 5)
            .takeLast(2)
            .subscribeOn(Schedulers.newThread())
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(String.valueOf(integer));
                }
            });

返回结果：4 5

指定时间过滤事件

（1）throttleFirst和throttleLast

throttleFirst：接收者只会接收到在指定时间内发送的第一条数据
throttleLast：接收者只会接收到在指定时间内发送的最后一条数据

    Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> e) throws Exception {
            //每隔500毫秒发送一条数据
            e.onNext(1);
            Thread.sleep(500);
            e.onNext(2);
            Thread.sleep(500);
            e.onNext(3);
            Thread.sleep(500);
            e.onNext(4);
            Thread.sleep(500);
            e.onNext(5);
            Thread.sleep(500);
            e.onNext(6);
            Thread.sleep(500);
            e.onNext(7);
            Thread.sleep(500);
            e.onNext(8);
            Thread.sleep(500);
            e.onNext(9);
            Thread.sleep(500);
        }
    })
            .throttleFirst(1700, TimeUnit.MILLISECONDS)
            .subscribeOn(Schedulers.newThread())
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(String.valueOf(integer));
                }
            });

返回结果：1 5 9

    Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> e) throws Exception {
            //每隔500毫秒发送一条数据
            e.onNext(1);
            Thread.sleep(500);
            e.onNext(2);
            Thread.sleep(500);
            e.onNext(3);
            Thread.sleep(500);
            e.onNext(4);
            Thread.sleep(500);
            e.onNext(5);
            Thread.sleep(500);
            e.onNext(6);
            Thread.sleep(500);
            e.onNext(7);
            Thread.sleep(500);
            e.onNext(8);
            Thread.sleep(500);
            e.onNext(9);
            Thread.sleep(500);
        }
    })
            .throttleLast(1700, TimeUnit.MILLISECONDS)
            .subscribeOn(Schedulers.newThread())
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(String.valueOf(integer));
                }
            });

返回结果：4 7 9

（2）sample

与throttleLast类似，sample接收者只会接收到在指定时间内发送的最后一条数据

    Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> e) throws Exception {
            //每隔500毫秒发送一条数据
            e.onNext(1);
            Thread.sleep(500);
            e.onNext(2);
            Thread.sleep(500);
            e.onNext(3);
            Thread.sleep(500);
            e.onNext(4);
            Thread.sleep(500);
            e.onNext(5);
            Thread.sleep(500);
            e.onNext(6);
            Thread.sleep(500);
            e.onNext(7);
            Thread.sleep(500);
            e.onNext(8);
            Thread.sleep(500);
            e.onNext(9);
            Thread.sleep(500);
        }
    })
            .sample(1700, TimeUnit.MILLISECONDS)
            .subscribeOn(Schedulers.newThread())
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(String.valueOf(integer));
                }
            });

返回结果：4 7 9

（3）throttleWithTimeout和debounce

throttleWithTimeout和debounce是一样的

假设指定时间为1秒，那么被观察者发送第一条数据时，等待1秒之后仍然没有发送第二条数据，那么观察者将收到该数据，如果还没到1秒钟就重新发送了数据，那么第一条数据就被取消发送。

    Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> e) throws Exception {
            //每隔500毫秒发送一条数据
            e.onNext(1);
            Thread.sleep(500);
            e.onNext(2);
            Thread.sleep(500);
            e.onNext(3);
            Thread.sleep(500);
            e.onNext(4);
            Thread.sleep(500);
            e.onNext(5);
            Thread.sleep(500);
            e.onNext(6);
            Thread.sleep(500);
            e.onNext(7);
            Thread.sleep(500);
            e.onNext(8);
            Thread.sleep(500);
            e.onNext(9);
            Thread.sleep(500);
        }
    })
            .throttleWithTimeout(600, TimeUnit.MILLISECONDS)
            .subscribeOn(Schedulers.newThread())
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(String.valueOf(integer));
                }
            });

返回结果：9

指定事件位置过滤事件

（1）firstElement和lastElement

firstElement：只取第一个元素
lastElement：只取最后一个元素

    Observable.just(1, 2, 3, 4, 5)
            .firstElement()
            .subscribeOn(Schedulers.newThread())
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(String.valueOf(integer));
                }
            });

返回结果：1

    Observable.just(1, 2, 3, 4, 5)
            .lastElement()
            .subscribeOn(Schedulers.newThread())
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(String.valueOf(integer));
                }
            });

返回结果：5

（2）elementAt

elementAt：只接收指定位置的元素（角标从0开始）

    Observable.just(1, 2, 3, 4, 5)
            .elementAt(2)
            .subscribeOn(Schedulers.newThread())
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(String.valueOf(integer));
                }
            });

返回结果：3

（3）elementAtOrError

elementAtOrError：elementAt即使角标越界也不会报错，但是elementAtOrError如果角标越界的话会报错。

    Observable.just(1, 2, 3, 4, 5)
            .elementAtOrError(5)
            .subscribeOn(Schedulers.newThread())
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(String.valueOf(integer));
                }
            });

以上代码报错如下：

图片.png

为了防止崩溃，还是把代码补全的好，添加接收异常的回调

    Observable.just(1, 2, 3, 4, 5)
            .elementAtOrError(5)
            .subscribeOn(Schedulers.newThread())
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(String.valueOf(integer));
                }
            }, new Consumer<Throwable>() {
                @Override
                public void accept(Throwable throwable) throws Exception {
                    System.out.println("角标越界了");
                }
            });

返回结果：角标越界了

最后编辑于：2019.06.17 10:00:23

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 219,753评论 6赞 508
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 93,668评论 3赞 396
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 166,090评论 0赞 356
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 59,010评论 1赞 295
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 68,054评论 6赞 395
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 51,806评论 1赞 308
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 40,484评论 3赞 420
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 39,380评论 0赞 276
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 45,873评论 1赞 319
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 38,021评论 3赞 338
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 40,158评论 1赞 352
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 35,838评论 5赞 346
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 41,499评论 3赞 331
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 32,044评论 0赞 22
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 33,159评论 1赞 272
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 48,449评论 3赞 374
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 45,136评论 2赞 356