RxJava学习笔记操作符学习（6）

作者: 西红柿的眼泪 | 来源:发表于2018-08-27 09:32 被阅读0次

过滤操作符

作用

过滤 / 筛选被观察者（Observable）发送的事件 & 观察者（Observer）接收的事件

类型

类型图.png

应用场景 & 对应操作符详解

1.Filter（）
作用
过滤特定条件的事件。
实例

public void FilterOperators(View view){
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onNext(4);
                emitter.onNext(5);
                
            }
        }).filter(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                return integer>3;
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e("sss", "开始采用subscribe连接");
            }

            @Override
            public void onNext(Integer integer) {
                Log.e("sss", "过滤后得到的事件是："+ integer );
            }

            @Override
            public void onError(Throwable e) {
                Log.e("sss", "对Error事件作出响应");
            }

            @Override
            public void onComplete() {
                Log.e("sss", "对Complete事件作出响应");
            }
        });
    }

2.ofType（）
作用
过滤特定数据类型的数据
事例

public void ofTypeOperators(View view){
      Observable.just(1,"hh",6,"lll",8)
                .ofType(Integer.class)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e("sss","获取到的整型事件元素是： "+ integer);
                    }
                });
    }

3.skip（） / skipLast（）
作用
跳过某个事件
事例

public void skipOperators(View view){
        Observable.just(1,2,3,4,5)
                .skip(1)
                .skipLast(2)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e("sss","获取到的整型事件元素是： "+ integer);
                    }
                });
    }

4.distinct（） / distinctUntilChanged（）
作用
过滤事件序列中重复的事件 / 连续重复的事件
事例

 public void distinctOperators(View view){
        Observable.just(1,2,5,1,5,6,3)
                .distinct()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e("sss","获取到的不重复事件元素是： "+ integer);
                    }
                });
    }

5.take（）
作用
指定观察者最多能接收到的事件数量
事例

 public void takeOperators(View view){
        Observable.just(1,2,3,4,6)
                .take(2)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e("sss","获取到的事件元素是： "+ integer);
                    }
                });
    }

6.takeLast（）
作用
指定观察者只能接收到被观察者发送的最后几个事件
事例

public void takeLastOperators(View view){
        Observable.just(1,2,3,4,5)
                .takeLast(3)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.e("sss", "开始采用subscribe连接");
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e("sss", "接受后得到的事件是："+ integer );
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e("sss", "对Error事件作出响应");
                    }

                    @Override
                    public void onComplete() {
                        Log.e("sss", "对Complete事件作出响应");
                    }
                });
    }

7.throttleFirst（）/ throttleLast（）
作用
在某段时间内，只发送该段时间内第1次事件 / 最后1次事件
事例

public void throttleFirstOperators(View view){
        Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                Thread.sleep(500);
                emitter.onNext(2);
                Thread.sleep(400);
                emitter.onNext(3);
                Thread.sleep(300);

                emitter.onNext(4);
                Thread.sleep(300);

                emitter.onNext(5);
                Thread.sleep(300);

                emitter.onNext(6);
                Thread.sleep(400);

                emitter.onNext(7);
                Thread.sleep(300);
                emitter.onNext(8);

                Thread.sleep(300);
                emitter.onNext(9);

                Thread.sleep(300);
                emitter.onComplete();

            }
        }).throttleFirst(1, TimeUnit.SECONDS)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.e("sss", "开始采用subscribe连接");
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e("sss", "接收到了事件"+ integer  );
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e("sss", "对Error事件作出响应" );
                    }

                    @Override
                    public void onComplete() {
                        Log.e("sss","对Complete事件作出响应");
                    }
                });
    }

8.Sample（）
作用
在某段时间内，只发送该段时间内最新（最后）1次事件与 throttleLast（）操作符类似。
9.throttleWithTimeout （） / debounce（）
作用
发送数据事件时，若2次发送事件的间隔＜指定时间，就会丢弃前一次的数据，直到指定时间内都没有新数据发射时才会发送后一次的数据。
事例

public void throttleWithTimeoutOperators(View view){
        Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                Thread.sleep(500);
                emitter.onNext(2);
                Thread.sleep(1500);
                emitter.onNext(3);
                Thread.sleep(1500);
                emitter.onNext(4);
                Thread.sleep(500);
                emitter.onNext(5);
                Thread.sleep(1500);
                emitter.onNext(6);
                Thread.sleep(500);
                emitter.onComplete();
            }
        }).throttleWithTimeout(1,TimeUnit.MICROSECONDS)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e("sss", "接收到了事件"+ integer  );
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e("sss", "对Error事件作出响应");
                    }

                    @Override
                    public void onComplete() {
                        Log.e("sss", "对Complete事件作出响应");
                    }
                });
    }

firstElement（） / lastElement（）
作用
仅选取第1个元素 / 最后一个元素
事例

public void firstElementOperators(View view){
        Observable.just(1,2,3,4,5)
                .firstElement()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e("sss","获取到的第一个事件是： "+ integer);
                    }
                });
    }

11.elementAt（）
作用
指定接收某个元素（通过索引值确定）。
事例

public void elementAtOperators(View view){
        Observable.just(1,2,3,4,5)
                .elementAt(2)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e("sss","获取到的事件元素是： "+ integer);
                    }
                });
    }

12.elementAtOrError（）
作用
在elementAt（）的基础上，当出现越界情况（即获取的位置索引＞发送事件序列长度）时，即抛出异常。
事例

  public void elementAtOrErrorOperators(View view){
        Observable.just(1,2,3,4,5)
                .elementAtOrError(6)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e("sss","获取到的事件元素是： "+ integer);
                    }
                });

    }

实际开发中的应用

1.功能防抖
throttleFirst的使用

Button bt=findViewById(R.id.bt);

        RxView.clicks(bt)
                .throttleFirst(4, TimeUnit.SECONDS)
                .subscribe(new Observer<Object>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Object o) {
                        Log.e("sss", "发送了网络请求" );
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e("sss", "对Error事件作出响应" + e.toString());
                    }

                    @Override
                    public void onComplete() {
                        Log.e("sss", "对Complete事件作出响应");
                    }
                });
    }

RxJava学习笔记操作符学习（6）

过滤操作符

作用

类型

应用场景 & 对应操作符详解

实际开发中的应用

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读