Android函数响应式编程最新RxJava-操作符入门

Posted 2021-04-30 秦子帅

今天介绍一下过滤操作符和组合操作符附带说一下辅助操作符，错误处理操作符，条件操作符，转换操作符

1.过滤操作符

过滤操作符用于过滤和选择Observable发射的数据序列，让Observable只返回满足我们条件的数据。过滤操作符有filter、elementAt、distinct、skip、take、ignoreElements、throttleFirst、throttlewithtimeout等等。

(1)filter

filter操作符是对Observable产生的结果自定义规则进行过滤,只有满足条件的才能提交给订阅者。

    Observable.just(1,2,3,4).filter(new AppendOnlyLinkedArrayList.NonThrowingPredicate<Integer>() {        @Override        public boolean test(Integer integer) {            return integer>2;        }    }).subscribe(new Observer<Integer>() {        @Override        public void onSubscribe(@NonNull Disposable d) {        }        @Override        public void onNext(@NonNull Integer integer) {            Log.e(TAG,integer+"");        }        @Override        public void onError(@NonNull Throwable e) {        }        @Override        public void onComplete() {        }    });

上面代码是把just里的值大于2的提交给订阅者,打印的LOG如下：

filter.png

(2)distinct

distinct操作符用来去重，就是去掉重复的数据，也就是每次发射的数据不一样，而distinctUntilChanged是去掉连续重复的数据.

         Observable.just(1,2,2,2,3,5,5,7).distinct().subscribe(new Observer<Integer>() {        @Override        public void onSubscribe(@NonNull Disposable d) {        }        @Override        public void onNext(@NonNull Integer integer) {            Log.e(TAG,integer+"");        }        @Override        public void onError(@NonNull Throwable e) {        }        @Override        public void onComplete() {        }    });

distinct.png

(3)skip 与 take

skip操作符是把Observable发射的数据过滤掉前N项，而take操作符只取前N项。
下面是skip的代码：

 Observable.just(1,2,3,4,5).skip(2).subscribe(new Observer<Integer>() {        @Override        public void onSubscribe(@NonNull Disposable d) {        }        @Override        public void onNext(@NonNull Integer integer) {            Log.e(TAG,integer+"");        }        @Override        public void onError(@NonNull Throwable e) {        }        @Override        public void onComplete() {        }    });

运行结果如下：

skip.png

take操作符的代码我就不贴出来了 ，意思是相同的。

(4)ignoreElements

ignoreElements操作符是忽略Observable产生的结果，只走onComplete()和onError()方法。

   Observable.just(1,2,3,4).ignoreElements().subscribe(new CompletableObserver() {       @Override       public void onSubscribe(@NonNull Disposable d) {           Log.e(TAG,"onSubscribe");       }       @Override       public void onComplete() {           Log.e(TAG,"onComplete");       }       @Override       public void onError(@NonNull Throwable e) {           Log.e(TAG,"onError");       }   });

ignoreElements.png

(5)throttleFirst

throttleFirst 操作符则会定期发这个时间里Observable发射的第一个。

  Observable.create(new ObservableOnSubscribe<Integer>() {        @Override        public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {            for (int i=0;i<9;i++){                e.onNext(i);                try {                    Thread.sleep(100);                } catch (InterruptedException e1) {                    e1.printStackTrace();                }            }            e.onComplete();        }    }).throttleFirst(300, TimeUnit.MILLISECONDS).subscribe(new Observer<Integer>() {        @Override        public void onSubscribe(@NonNull Disposable d) {        }        @Override        public void onNext(@NonNull Integer integer) {            Log.e(TAG,""+integer);        }        @Override        public void onError(@NonNull Throwable e) {        }        @Override        public void onComplete() {        }    });

每隔300ms,就取发射源里的第一个数据，打印如下:

throttleFirst.jpg

2.组合操作符

组合操作符可以同时处理多个Observable，组合操作符有 mergestartwith,join,switch,concat,cmbineLastest等等

(1)startWith

startWith操作符是在发射的数据前插入一条数据。

Observable.just(4,5).startWith(1).subscribe(new Observer<Integer>() {        @Override        public void onSubscribe(@NonNull Disposable d) {        }        @Override        public void onNext(@NonNull Integer integer) {            Log.e(TAG,""+integer);        }        @Override        public void onError(@NonNull Throwable e) {        }        @Override        public void onComplete() {        }    });

startwith.png

(2)merge

merge操作符是将多个Observable合并到一个中并发射，但是发射的数据是交错的，concat操作符是严格按照顺序合并发射的。

   Observable<Integer> ob1=Observable.just(1,2).subscribeOn(Schedulers.computation());    Observable<Integer> ob2=Observable.just(3,4).subscribeOn(Schedulers.computation());    Observable.merge(ob1,ob2).subscribe(new Observer<Integer>() {        @Override        public void onSubscribe(@NonNull Disposable d) {        }        @Override        public void onNext(@NonNull Integer integer) {            Log.e(TAG,""+integer);        }        @Override        public void onError(@NonNull Throwable e) {        }        @Override        public void onComplete() {        }    });

merge.png

3.其余四个操作符

(1)辅助操作符：辅助操作符可以更方便的处理Observable，包括delay,DO,subscribeOn，timeout，materialize，timeInterval，timestamp。

(2)错误处理操作符:错误处理操作符有catch和retry.

(3)条件操作符: 条件操作符可用于根据条件发射或叫唤Observable.条件操作符包括amb，defaultIfEmpty，skipUntil，skipwhile等等

(4)转换操作符：转换操作符用来将Observable转换为另一个对象或数据结构，包括toList，tosortdList，toMultiMap，getIterator等等