Android小白_努力中Android开发经验谈Android技术知识

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

2017-08-08  本文已影响73人  秦子帅

Android函数响应式编程最新RxJava-基本用法
Android函数响应式编程最新RxJava-操作符入门(1)
Github地址
今天介绍一下过滤操作符组合操作符附带说一下辅助操作符错误处理操作符条件操作符转换操作符

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等等

上一篇下一篇

猜你喜欢

热点阅读