Chinaunix首页 | 论坛 | 博客
  • 博客访问: 1226731
  • 博文数量: 114
  • 博客积分: 3526
  • 博客等级: 中校
  • 技术积分: 1825
  • 用 户 组: 普通用户
  • 注册时间: 2009-09-08 09:46
  • 认证徽章:
个人简介

记录总结自己的工作

文章分类

全部博文(114)

文章存档

2015年(19)

2014年(10)

2013年(6)

2012年(17)

2011年(24)

2010年(21)

2009年(17)

分类: Android平台

2015-09-25 16:33:21

    在这篇文章中我们来了解一些辅助操作符,可以让我们更加地方便地处理Observable。

    一、Delay
    顾名思义,Delay操作符就是让发射数据的时机延后一段时间,这样所有的数据都会依次延后一段时间发射。在Rxjava中将其实现为Delay和DelaySubscription。不同之处在于Delay是延时数据的发射,而DelaySubscription是延时注册Subscriber。


    下面我们使用Delay和DelaySubscribtion操作符来延迟两个Observable数据的发射
点击(此处)折叠或打开
  1. private Observable<Long> delayObserver() {
  2.         return createObserver(2).delay(2000, TimeUnit.MILLISECONDS);
  3.     }

  4.     private Observable<Long> delaySubscriptionObserver() {
  5.         return createObserver(2).delaySubscription(2000, TimeUnit.MILLISECONDS);
  6.     }

  7.     private Observable<Long> createObserver(int index) {
  8.         return Observable.create(new Observable.OnSubscribe<Long>() {
  9.             @Override
  10.             public void call(Subscriber<? super Long> subscriber) {
  11.                 log("subscrib:" + getCurrentTime());
  12.                 for (int i = 1; i <= index; i++) {
  13.                     subscriber.onNext(getCurrentTime());
  14.                     try {
  15.                         Thread.sleep(1000);
  16.                     } catch (InterruptedException e) {
  17.                         e.printStackTrace();
  18.                     }
  19.                 }
  20.             }
  21.         }).subscribeOn(Schedulers.newThread());
  22.     }

  23.     private long getCurrentTime() {
  24.         return System.currentTimeMillis()/1000;
  25.     }
   
     分别对其进行注册
点击(此处)折叠或打开
  1. mLButton.setText("delay");
  2.         mLButton.setOnClickListener(e -> {
  3.             log("start subscrib:" + getCurrentTime());
  4.             delayObserver().subscribe(i -> log("delay:" + (getCurrentTime() - i)));
  5.         });
  6.         mRButton.setText("delaySubscription");
  7.         mRButton.setOnClickListener(e -> {
  8.             log("start subscrib:" + getCurrentTime());
  9.             delaySubscriptionObserver().subscribe(i -> log("delaySubscription:" + i));
  10.         });
    运行结果如下。可以看到两个操作符都让我们达到了延迟2秒后再发射数据的目的


    二、Do
    Do操作符就是给Observable的生命周期的各个阶段加上一系列的回调监听,当Observable执行到这个阶段的时候,这些回调就会被触发。在Rxjava实现了很多的doxxx操作符。
    DoOnEach可以给Observable加上这样的样一个回调:Observable每发射一个数据的时候就会触发这个回调,不仅包括onNext还包括onError和onCompleted。
    DoOnNext则只有onNext的时候才会被触发。

    
    doOnSubscribe和doOnUnSubscribe则会在Subscriber进行订阅和反订阅的时候触发回调。当一个Observable通过OnError或者OnCompleted结束的时候,会反订阅所有的Subscriber。

    DoOnError会在OnError发生的时候触发回调,并将Throwable对象作为参数传进回调函数里;DoOnComplete会在OnCompleted发生的时候触发回调

    DoOnTerminate会在Observable结束前触发回调,无论是正常还是异常终止;finallyDo会在Observable结束后触发回调,无论是正常还是异常终止。

    好了,介绍了这么多do的操作符,我们接下来创建两个Observable对象,并分别用上面的一系列do操作符进行注册回调

点击(此处)折叠或打开

  1. private Observable<Integer> doOnEachObserver() {
  2.         return Observable.just(1, 2, 3)
  3.                 .doOnEach(notification -> log("doOnEach send " + notification.getValue() + " type:" + notification.getKind()))
  4.                 .doOnNext(aInteger -> log("doOnNext send " + aInteger))
  5.                 .doOnSubscribe(() -> log("on subscribe"))
  6.                 .doOnUnsubscribe(() -> log("on unsubscribe\n"))
  7.                 .doOnCompleted(() -> log("onCompleted"));

  8.     }

  9.     private Observable<Integer> doOnErrorObserver() {
  10.         return createObserver()
  11.                 .doOnEach(notification -> log("doOnEach send " + notification.getValue() + " type:" + notification.getKind()))
  12.                 .doOnError(throwable -> log("OnError:" + throwable.getMessage()))
  13.                 .doOnTerminate(() -> log("OnTerminate"))
  14.                 .finallyDo(() -> log("finallyDo"));
  15.     }

  16.     private Observable<Integer> createObserver() {
  17.         return Observable.create(new Observable.OnSubscribe<Integer>() {
  18.             @Override
  19.             public void call(Subscriber<? super Integer> subscriber) {
  20.                 for (int i = 1; i <= 5; i++) {
  21.                     if (i <= 3) {
  22.                         subscriber.onNext(i);
  23.                     } else {
  24.                         subscriber.onError(new Throwable("num>3"));
  25.                     }
  26.                 }
  27.             }
  28.         });
  29.     }
    分别进行订阅

点击(此处)折叠或打开

  1. mLButton.setText("do");
  2.         mLButton.setOnClickListener(e -> doOnEachObserver().subscribe(i -> log("do:" + i)));
  3.         mRButton.setText("doOnError");
  4.         mRButton.setOnClickListener(e -> doOnErrorObserver().subscribe(new Subscriber<Integer>() {
  5.             @Override
  6.             public void onCompleted() {
  7.                 log("onCompleted");
  8.             }

  9.             @Override
  10.             public void onError(Throwable e) {
  11.                 log("subscriber onError:" + e.getMessage());
  12.             }

  13.             @Override
  14.             public void onNext(Integer integer) {
  15.                 log("subscriber onNext:" + integer);
  16.             }
  17.         }));
    运行结果如下所示。可以看到各个回调的监听都被依次触发。


    三、Meterialize
    Meterialize操作符将OnNext/OnError/OnComplete都转化为一个Notification对象并按照原来的顺序发射出来,而DeMeterialize则是执行相反的过程。


    下面我们使用这两个操作符来处理两个Observable对象

点击(此处)折叠或打开

  1. private Observable<Notification<Integer>> meterializeObserver() {
  2.         return Observable.just(1, 2, 3).materialize();
  3.     }

  4.     private Observable<Integer> deMeterializeObserver() {
  5.         return eterializeObserver().dematerialize();
  6.     }
    分别进行订阅

点击(此处)折叠或打开

  1. mLButton.setText("meterialize");
  2. mLButton.setOnClickListener(e -> meterializeObserver().subscribe(i -> log("meterialize:" + i.getValue() + " type" + i.getKind())));
  3. mRButton.setText("deMeterialize");
  4. mRButton.setOnClickListener(e -> deMeterializeObserver().subscribe(i->log("deMeterialize:"+i)));
    运行结果如下所示,可以看到onComplete也被meterialize包装后发射了出来,onError也同样。


    四、SubscribOn/ObserverOn
    这两个操作符在前面的例子中我们已经使用过多次了,使用起来十分方便。在android开发中,相信大家一定都遇到过不能在主线程修改UI的问题,所以不得不使用Handler、AsyncTask等来更新UI界面。使用SubscribOn和ObserverOn操作符,各种线程的问题都将变得十分地简单。
    SubscribOn用来指定Observable在哪个线程上运行,我们可以指定在IO线程上运行也可以让其新开一个线程运行,当然也可以在当前线程上运行。一般来讲会指定在各种后台线程而不是主线程上运行,就如同AsyncTask的doInBackground一样。

    ObserverOn用来指定观察者所运行的线程,也就是发射出的数据在那个线程上使用。在android中,如果我们要修改UI界面,观察者就必须在主线程上运行,就如同AsyncTask的
onPostExecute。


     下面创建两个Observable并使用ObserverOn和SubscribOn使Observable和观察者运行在不同的线程上。

点击(此处)折叠或打开

  1. private Observable<Integer> observerOnObserver() {
  2.         return createObserver()
  3.                 .observeOn(AndroidSchedulers.mainThread())
  4.                 .subscribeOn(Schedulers.newThread());
  5.     }

  6.     private Observable<Integer> subscribeOnObserver() {
  7.         return createObserver()
  8.                 .subscribeOn(Schedulers.computation())
  9.                 .observeOn(Schedulers.immediate());
  10.     }

  11.     private Observable<Integer> createObserver() {
  12.         return Observable.create(new Observable.OnSubscribe<Integer>() {
  13.             @Override
  14.             public void call(Subscriber<? super Integer> subscriber) {
  15.                 log("on subscrib:" + Thread.currentThread().getName());
  16.                 subscriber.onNext(1);
  17.                 subscriber.onCompleted();
  18.             }
  19.         });
  20.     }
     分别进行订阅

点击(此处)折叠或打开

  1. mLButton.setText("observerOn");
  2. mLButton.setOnClickListener(e -> observerOnObserver().subscribe(i -> log("observerOn:" + Thread.currentThread().getName())));
  3. mRButton.setText("subscribeOn");
  4. mRButton.setOnClickListener(e -> subscribeOnObserver().subscribe(i -> log("subscribeOn:" + Thread.currentThread().getName())));
     运行结果如下

     
     五、TimeInterval\TimeStamp
     TimeInterval会拦截发射出来的数据,取代为前后两个发射两个数据的间隔时间。对于第一个发射的数据,其时间间隔为订阅后到首次发射的间隔。

     TimeStamp会将每个数据项给重新包装一下,加上了一个时间戳来标明每次发射的时间

     
     下面使用这两个操作符来处理两个Observable对象

点击(此处)折叠或打开

  1. private Observable<TimeInterval<Integer>> timeIntervalObserver() {
  2.         return createObserver().timeInterval();
  3.     }

  4.     private Observable<Timestamped<Integer>> timeStampObserver() {
  5.         return createObserver().timestamp();
  6.     }

  7.     private Observable<Integer> createObserver() {
  8.         return Observable.create(new Observable.OnSubscribe<Integer>() {
  9.             @Override
  10.             public void call(Subscriber<? super Integer> subscriber) {
  11.                 for (int i = 0; i <= 3; i++) {
  12.                     try {
  13.                         Thread.sleep(1000);
  14.                     } catch (InterruptedException e) {
  15.                         e.printStackTrace();
  16.                     }
  17.                     subscriber.onNext(i);
  18.                 }
  19.                 subscriber.onCompleted();
  20.             }
  21.         }).subscribeOn(Schedulers.newThread());
  22.     }
     分别进行订阅

点击(此处)折叠或打开

  1. mLButton.setText("timeInterval");
  2. mLButton.setOnClickListener(e -> timeIntervalObserver().subscribe(i -> log("timeInterval:" + i.getValue()+"-"+i.getIntervalInMilliseconds())));
  3. mRButton.setText("timeStamp");
  4. mRButton.setOnClickListener(e -> timeStampObserver().subscribe(i -> log("timeStamp:" + i.getValue()+"-"+i.getTimestampMillis())));
     运行结果如下所示。


     六、Timeout
     Timeout操作符给Observable加上超时时间,每发射一个数据后就重置计时器,当超过预定的时间还没有发射下一个数据,就抛出一个超时的异常。
     Rxjava将Timeout实现为很多不同功能的操作符,比如说超时后用一个备用的Observable继续发射数据等。

     下面我们创建一个Observable,逐渐加大间隔地发射数据,并使用timeout加上超时的限制。

点击(此处)折叠或打开

  1. private Observable<Integer> timeoutObserver() {
  2.         return createObserver().timeout(200, TimeUnit.MILLISECONDS);
  3.     }

  4.     private Observable<Integer> timeoutobserverObserver() {
  5.         return createObserver().timeout(200, TimeUnit.MILLISECONDS, Observable.just(5, 6));
  6.     }

  7.     private Observable<Integer> createObserver() {
  8.         return Observable.create(new Observable.OnSubscribe<Integer>() {
  9.             @Override
  10.             public void call(Subscriber<? super Integer> subscriber) {
  11.                 for (int i = 0; i <= 3; i++) {
  12.                     try {
  13.                         Thread.sleep(i * 100);
  14.                     } catch (InterruptedException e) {
  15.                         e.printStackTrace();
  16.                     }
  17.                     subscriber.onNext(i);
  18.                 }
  19.                 subscriber.onCompleted();
  20.             }
  21.         });
  22.     }
     分别进行订阅

点击(此处)折叠或打开

  1. mLButton.setText("timeout");
  2.         mLButton.setOnClickListener(e -> timeoutObserver().subscribe(new Subscriber<Integer>() {
  3.             @Override
  4.             public void onCompleted() {

  5.             }

  6.             @Override
  7.             public void onError(Throwable e) {
  8.                 log(e);
  9.             }

  10.             @Override
  11.             public void onNext(Integer integer) {
  12.                 log("timeout:" + integer);
  13.             }
  14.         }));
  15.         mRButton.setText("timeoutobserver");
  16.         mRButton.setOnClickListener(e -> timeoutobserverObserver().subscribe(i -> log(i)));
     运行结果如下


     七、Using
     Using操作符创建一个在Observable生命周期内存活的资源,也可以这样理解:我们创建一个资源并使用它,用一个Observable来限制这个资源的使用时间,当这个Observable终止的时候,这个资源就会被销毁。
Using需要使用三个参数,分别是:
     1.创建这个一次性资源的函数
     2.创建Observable的函数
     3.释放资源的函数

     下面我们定义了一个Animal类,并使用Using来控制其创建和释放。

点击(此处)折叠或打开

  1. private Observable<Long> usingObserver() {
  2.         return Observable.using(() -> new Animal(), i -> Observable.timer(5000,TimeUnit.MILLISECONDS), o -> o.relase());
  3.     }

  4.     private class Animal {
  5.         Subscriber subscriber = new Subscriber() {
  6.             @Override
  7.             public void onCompleted() {

  8.             }

  9.             @Override
  10.             public void onError(Throwable e) {

  11.             }

  12.             @Override
  13.             public void onNext(Object o) {
  14.                 log("animal eat");
  15.             }
  16.         };

  17.         public Animal() {
  18.             log("create animal");
  19.             Observable.interval(1000, TimeUnit.MILLISECONDS)
  20.                     .subscribe(subscriber);
  21.         }

  22.         public void relase() {
  23.             log("animal released");
  24.             subscriber.unsubscribe();
  25.         }
  26.     }
     对其进行订阅

点击(此处)折叠或打开

  1. Observable<Long> observable = usingObserver();
  2.         Subscriber subscriber = new Subscriber() {
  3.             @Override
  4.             public void onCompleted() {
  5.                 log("onCompleted");
  6.             }

  7.             @Override
  8.             public void onError(Throwable e) {
  9.                 log("onError");
  10.             }

  11.             @Override
  12.             public void onNext(Object o) {
  13.                 log("onNext"+o);
  14.             }
  15.         };
  16.         mLButton.setText("using");
  17.         mLButton.setOnClickListener(e -> observable.subscribe(subscriber));
  18.         mRButton.setText("unSubscrib");
  19.         mRButton.setOnClickListener(e -> subscriber.unsubscribe());
     运行结果如下。在订阅了几秒之后,对其进行反订阅,Observable就会终止从而触发Animal的释放。


    关于辅助操作符就到这里了,本文中的demo程序见https://github.com/Chaoba/RxJavaDemo
阅读(16500) | 评论(0) | 转发(0) |
给主人留下些什么吧!~~
评论热议
请登录后评论。

登录 注册