This post is from a low-reputation account and contains an unverified outbound link. Be cautious before clicking external links.

RxJava2.0操作符之--辅助操作符

hiquanta(3)
Published in
#steemit
Words
410
Reading
2 min
9y
Listen to this post
~3 min

Delay

延迟一段指定的时间再发射来自Observable的发射物

Observable.just(1, 2, 3, 4).delay(5, TimeUnit.SECONDS).subscribe(RxUtils.getObserver());
try {
    Thread.sleep(Integer.MAX_VALUE);
} catch (InterruptedException e1) {
    e1.printStackTrace();
}

Do

注册一个动作作为原始Observable生命周期事件的一种占位符,相当于注册回调

其中RxJava中的实现为doXXX 很简单 看名字就知道回调时机和怎么用

  //doOnEach
//        Observable.just(1,2,3,4,5).doOnEach(RxUtils.getObserver()).subscribe();
  /**
 * onNext:1 Thread:Thread[main,5,main] onNext:2 Thread:Thread[main,5,main] onNext:3 Thread:Thread[main,5,main] onNext:4 Thread:Thread[main,5,main] onNext:5 Thread:Thread[main,5,main] onComplete Thread:Thread[main,5,main] */  //doOnNext
//        Observable.just(1,2,3,4,5).doOnNext(new Consumer() {
//            public void accept(@NonNull Integer integer) throws Exception {
//                if(integer>3)
//                    throw new RuntimeException("error");
//            }
//        }).subscribe(RxUtils.getObserver());
  /**
 * onSubscribe Thread:Thread[main,5,main] onNext:1 Thread:Thread[main,5,main] onNext:2 Thread:Thread[main,5,main] onNext:3 Thread:Thread[main,5,main] onError:java.lang.RuntimeException: error Thread:Thread[main,5,main] */  //doOnSubscribe
//        Observable observable = Observable.just(1, 2, 3, 4, 5, 6).doOnSubscribe(new Consumer() {
//            public void accept(@NonNull Disposable disposable) throws Exception {
//                System.out.println("published");
//            }
//        });
//        observable.subscribe(RxUtils.getObserver());
//        observable.subscribe(RxUtils.getObserver());

  /**
 * published onSubscribe Thread:Thread[main,5,main] onNext:1 Thread:Thread[main,5,main] onNext:2 Thread:Thread[main,5,main] onNext:3 Thread:Thread[main,5,main] onNext:4 Thread:Thread[main,5,main] onNext:5 Thread:Thread[main,5,main] onNext:6 Thread:Thread[main,5,main] onComplete Thread:Thread[main,5,main] published onSubscribe Thread:Thread[main,5,main] onNext:1 Thread:Thread[main,5,main] onNext:2 Thread:Thread[main,5,main] onNext:3 Thread:Thread[main,5,main] onNext:4 Thread:Thread[main,5,main] onNext:5 Thread:Thread[main,5,main] onNext:6 Thread:Thread[main,5,main] onComplete Thread:Thread[main,5,main] */  //doOnUnsubscribe 2.0 has removed instead of
  DisposableSubscriber disposableSubscriber = new DisposableSubscriber() {
            public void onNext(Long along) {
                System.out.println(along);
  }

            public void onError(Throwable throwable) {
                System.out.println(throwable);
  }

            public void onComplete() {
                System.out.println("onComplete");
  }
        };

  Flowable.interval(1000, TimeUnit.MILLISECONDS).doOnCancel(new Action() {
            public void run() throws Exception {
                System.out.println("doOnCancel");
  }
        }).subscribe(disposableSubscriber);

  disposableSubscriber.dispose();

Materialize/Dematerialize

Materialize将数据项和事件通知都当做数据项发射，Dematerialize刚好相反。

Dematerialize操作符是Materialize的逆向过程，它将Materialize转换的结果还原成它原本的形式。

    Observable.just(1,2,3,4).materialize().subscribe(RxUtils.>getObserver());
    Observable.just(1,2,3,4).materialize().dematerialize().subscribe(RxUtils.getObserver());

ObserveOn_SubscribOn

ObserveOn:指定一个观察者在哪个调度器上观察这个Observable

SubscribOn:用来指定Observable在哪个线程上运行

  //  Observable.just(1,2,3,4,5).observeOn( Schedulers.newThread()).subscribeOn(Schedulers.computation()).subscribe(RxUtils.getObserver());
  Observable.just(1,2,3,4,5).subscribeOn( Schedulers.newThread()).subscribeOn(Schedulers.computation()).subscribe(RxUtils.getObserver());
 try {
      Thread.sleep(Integer.MAX_VALUE);
  } catch (InterruptedException e1) {
      e1.printStackTrace();
  }

Serialize

强制一个Observable连续调用并保证行为正确

这个说明目前没有从文档中找到能够说明此功能的代码，自己的测试代码也有点问题

如果你又好的方法，可以留言告诉我

Subscribe

操作来自Observable的发射物和通知

一直在用这个方法，很简单，就不贴代码了

1.x 2.x 方法差异如下

[RxJava 1․x BlockingObservable.forEach forEach subscribe](http://reactivex.io/documentation/operators/subscribe.html#collapseRxJava 1․x)

[RxJava 2․x blockingForEach blockingSubscribe forEachWhile safeSubscribe subscribe](http://reactivex.io/documentation/operators/subscribe.html#collapseRxJava 2․x)

TimeInterval

将一个发射数据的Observable转换为发射那些数据发射时间间隔的对象

Observable.interval(3, TimeUnit.SECONDS)
            .timeInterval()
            .subscribe(RxUtils.>getObserver());
 try {
        Thread.sleep(Integer.MAX_VALUE);
  } catch (InterruptedException e1) {
        e1.printStackTrace();
  }
}

onSubscribe
Thread:Thread[main,5,main]
onNext:Timed[time=3007, unit=MILLISECONDS, value=0]
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:Timed[time=3000, unit=MILLISECONDS, value=1]
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:Timed[time=3001, unit=MILLISECONDS, value=2]
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:Timed[time=3000, unit=MILLISECONDS, value=3]
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:Timed[time=3000, unit=MILLISECONDS, value=4]
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:Timed[time=2999, unit=MILLISECONDS, value=5]
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:Timed[time=3000, unit=MILLISECONDS, value=6]
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:Timed[time=3001, unit=MILLISECONDS, value=7]
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:Timed[time=3000, unit=MILLISECONDS, value=8]
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:Timed[time=3000, unit=MILLISECONDS, value=9]
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:Timed[time=3000, unit=MILLISECONDS, value=10]
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:Timed[time=3000, unit=MILLISECONDS, value=11]
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:Timed[time=2999, unit=MILLISECONDS, value=12]

Timeout

对原始Observable的一个镜像，如果过了一个指定的时长仍没有发射数据，它会发一个错误通知

    Observable.interval(3, TimeUnit.SECONDS)
            .timeout(2,TimeUnit.SECONDS)
            .subscribe(RxUtils.getObserver());
    try {
        Thread.sleep(Integer.MAX_VALUE);
    } catch (InterruptedException e1) {
        e1.printStackTrace();
    }

    onSubscribe
    Thread:Thread[main,5,main]
    onError:java.util.concurrent.TimeoutException
    Thread:Thread[RxComputationThreadPool-1,5,main]

Timestamp

给Observable发射的数据项附加一个时间戳

  Observable.interval(3, TimeUnit.SECONDS)
          .timestamp()
          .subscribe(RxUtils.>getObserver());
  try {
      Thread.sleep(Integer.MAX_VALUE);
  } catch (InterruptedException e1) {
      e1.printStackTrace();
  }

onSubscribe
Thread:Thread[main,5,main]
onNext:Timed[time=1500432643307, unit=MILLISECONDS, value=0]
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:Timed[time=1500432646306, unit=MILLISECONDS, value=1]
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:Timed[time=1500432649306, unit=MILLISECONDS, value=2]
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:Timed[time=1500432652306, unit=MILLISECONDS, value=3]
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:Timed[time=1500432655307, unit=MILLISECONDS, value=4]

Using

创建一个只在Observable生命周期内存在的一次性资源

Observable.interval(3, TimeUnit.SECONDS)
        .using(new Callable() {
            public Long call() throws Exception {
                return Long.valueOf(2);
  }
        }, new Function, ObservableSource>() {
            public ObservableSource apply(@NonNull Long aLong) throws Exception {
                return new ObservableSource() {
                    public void subscribe(@NonNull Observersuper Long> observer) {
                        observer.onNext(Long.valueOf(2));
  }
                };
  }
        }, new Consumer() {
            public void accept(@NonNull Long aLong) throws Exception {
                        System.out.println(aLong);
  }
        })
        .subscribe(RxUtils.getObserver());
try {
    Thread.sleep(Integer.MAX_VALUE);
} catch (InterruptedException e1) {
    e1.printStackTrace();
}

    onNext:2
   Thread:Thread[main,5,main]
steemit
cn
news
java
cn-programming
via steemit
This post was published via steemit. Ecency is not the originator or editor of this content and displays it for discovery purposes only.
5
RxJava2.0操作符之--辅助操作符 | Ecency