RxJava学习（七）过滤操作符

1.filter
过滤特定条件的事件

 /**
     * 过滤操作符filter
     * 过滤特定条件的事件
     */
    private void filter() {
        Observable.just(1, 2, 3, 4, 5).filter(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                //自定义过滤条件
                //过滤掉大于3的事件
                return integer > 3;
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "accept: " + integer);
            }
        });

    }

输出结果：

04-09 09:47:27.507 5838-5838/com.mujin.keji.rxjava_01 E/MainActivity: accept: 4
04-09 09:47:27.507 5838-5838/com.mujin.keji.rxjava_01 E/MainActivity: accept: 5

2.ofType
过滤出指定的数据类型

   /**ofType操作符
     * 过滤出指定的数据类型
     */
    private void ofType(){
        Observable.just(1,"你好",1.01,true).ofType(String.class).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
              //筛选出的字符串类型数据
                Log.e(TAG, "accept: "+s );
            }
        });

    }

输出结果：

04-09 09:52:02.850 6299-6299/com.mujin.keji.rxjava_01 E/MainActivity: accept: 你好

3.skip,skipLast
skip：跳过事件正序的前几项
skipLast:跳过事件正序的后几项

/**
     * skip：跳过事件正序的前几项
     * skipLast:跳过事件正序的后几项
     */
    private void skipAndSkipLast(){
        Observable.just(1,2,3,4,5,6,7,8).skip(2).skipLast(3).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "accept: "+integer );
            }
        });

    }

输出结果：

04-09 09:57:00.280 6713-6713/com.mujin.keji.rxjava_01 E/MainActivity: accept: 3
04-09 09:57:00.280 6713-6713/com.mujin.keji.rxjava_01 E/MainActivity: accept: 4
04-09 09:57:00.280 6713-6713/com.mujin.keji.rxjava_01 E/MainActivity: accept: 5

4.distinc,distincUntilChanged
distinc:过滤事件中重复的事件
distincUntilChanged:过滤事件中连续重复的事件

  /**
     * distinc:过滤事件中重复的事件
     */
    private void distincAndDistincUntilChanged(){
        Observable.just(1,2,3,4,5,1,2).distinct().subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "accept: "+integer );
            }
        });

    }

输出结果：

04-09 10:01:56.287 7306-7306/com.mujin.keji.rxjava_01 E/MainActivity: accept: 1
04-09 10:01:56.287 7306-7306/com.mujin.keji.rxjava_01 E/MainActivity: accept: 2
04-09 10:01:56.287 7306-7306/com.mujin.keji.rxjava_01 E/MainActivity: accept: 3
04-09 10:01:56.287 7306-7306/com.mujin.keji.rxjava_01 E/MainActivity: accept: 4
04-09 10:01:56.287 7306-7306/com.mujin.keji.rxjava_01 E/MainActivity: accept: 5

    /**
     * distincUntilChanged:过滤事件中连续重复的事件
     */
    private void distincAndDistincUntilChanged(){
        Observable.just(1,1,1,4,5,1,2).distinct().subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "accept: "+integer );
            }
        });

    }

输出结果：

04-09 10:03:45.652 7581-7581/com.mujin.keji.rxjava_01 E/MainActivity: accept: 1
04-09 10:03:45.652 7581-7581/com.mujin.keji.rxjava_01 E/MainActivity: accept: 4
04-09 10:03:45.652 7581-7581/com.mujin.keji.rxjava_01 E/MainActivity: accept: 5
04-09 10:03:45.652 7581-7581/com.mujin.keji.rxjava_01 E/MainActivity: accept: 2

5.take,takeLast
take:指定观察者接收前几个事件
takeLast:指定观察者接受后几个事件

 /**
     * take:指定观察者接受前几个事件
     * 
     */
    private void  takeAndTakeLast(){
        Observable.just(1,2,3,4,5).take(2).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "accept: "+integer );
            }
        });

    }

输出结果：

04-09 10:07:29.233 7946-7946/com.mujin.keji.rxjava_01 E/MainActivity: accept: 1
04-09 10:07:29.233 7946-7946/com.mujin.keji.rxjava_01 E/MainActivity: accept: 2


    /**
     *
     * takeLast:指定观察者接受后几个事件
     */
    private void  takeAndTakeLast(){
        Observable.just(1,2,3,4,5).takeLast(2).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "accept: "+integer );
            }
        });

    }

输出结果：

04-09 10:10:26.300 8364-8364/com.mujin.keji.rxjava_01 E/MainActivity: accept: 4
04-09 10:10:26.300 8364-8364/com.mujin.keji.rxjava_01 E/MainActivity: accept: 5
```

6.根据指定事件时间过滤事件：thottleFirst,thottleLast,sample
thottleFirst:每隔某段时间内只发送第一次事件
throttleLast:每隔某段时间内只发送最后一次事件
sample:和throttleLast原理一样

 /**
     * throttleFirst：每隔某段时间内只发送第一次事件
     * throttleLast:每隔某段时间内只发送最后一次事件
     */
    private void throttleFirstAndThrottleLast()
    {
        Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                Thread.sleep(300);
                e.onNext(2);
                Thread.sleep(300);
                e.onNext(3);
                Thread.sleep(300);
                e.onNext(4);
                Thread.sleep(300);
                e.onNext(5);
                Thread.sleep(300);
                e.onNext(6);

            }
        }).throttleFirst(1,TimeUnit.SECONDS).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "accept: "+integer );
            }
        });
        Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                Thread.sleep(300);
                e.onNext(2);
                Thread.sleep(300);
                e.onNext(3);
                Thread.sleep(300);
                e.onNext(4);
                Thread.sleep(300);
                e.onNext(5);
                Thread.sleep(300);
                e.onNext(6);

            }
        }).throttleLast(1,TimeUnit.SECONDS).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "accept: "+integer );
            }
        });
    }

输出结果：

04-09 10:40:08.541 18221-18221/com.mujin.keji.rxjava_01 E/MainActivity: throttleFirst: 1
04-09 10:40:09.747 18221-18221/com.mujin.keji.rxjava_01 E/MainActivity: throttleFirst: 5
04-09 10:40:11.052 18221-18335/com.mujin.keji.rxjava_01 E/MainActivity: throttleLast: 10
04-09 10:40:12.052 18221-18335/com.mujin.keji.rxjava_01 E/MainActivity: throttleLast: 12

7.根据指定事件位置过滤事件
firstElement:选取第一个事件
lastElement:选取最后一个事件
elementAt:选择特定索引的事件


    /**
     * firstElement：选取第一个事件
     * lastElement:选取最后一个事件
    *  elementAt:选择特定索引的事件(从0开始)
     *
     */
    private void firstElementAndLastElement(){
        Observable.just(1,2,3,4).firstElement().subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "firstElement: "+ integer);
            }
        });
        Observable.just(1,2,3,4).lastElement().subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "lastElement: "+ integer);
            }
        });
    }

输出结果：

04-09 10:47:04.592 19358-19358/com.mujin.keji.rxjava_01 E/MainActivity: firstElement: 1
04-09 10:47:04.593 19358-19358/com.mujin.keji.rxjava_01 E/MainActivity: lastElement: 4
04-09 10:47:04.593 19358-19358/com.mujin.keji.rxjava_01 E/MainActivity: elementAt: 3

8.elementAtOrError
elementAtOrError和elementAt一样
选择指定索引的事件
区别在于，如果索引越界，会抛出异常

    /**
     * elementAtOrError和elementAt一样
     * 选择指定索引的事件
     * 区别在于，如果索引越界，会抛出异常
     */
    private void  elementAtOrError(){
        Observable.just(1,2,3,4).elementAtOrError(4).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "elementAtOrError和elementAt一样: "+ integer);
            }
        });
    }