RxJava日常使用总结（六）辅助操作

link之家

链接快照平台

输入网页链接，自动生成快照
标签化管理网页链接

delay操作符

延迟一段指定的时间再发射来自Observable的发射物。就是推迟指定发射Observable的事件。

Observable
.fromIterable(mItems)
.delay(1, TimeUnit.SECONDS)
.as(RxLifecycleUtil.bindLifecycle(this))
.subscribe();
复制代码

do操作符

注册一个动作作为原始Observable生命周期事件的一种占位符。

Observable发射的事件，我们可以在注册subscribe中处理，但有时候Observable是提供出去的，在哪里注册都是未知的，不可能每个subscribe的地方都写一套。这时候do操作符就用处了，do系列的操作符相当于事件的钩子，在执行时调用使用do系列的操作的回调。

doOnEach操作符

doOnEach操作符让你可以注册一个回调，它产生的Observable每发射一项数据就会调用它一次。参数是Notification。

Observable.fromIterable(mItems)
    .doOnEach(new Consumer<Notification<Object>>() {
        @Override
        public void accept(Notification<Object> notification) throws Exception {
            //...
    .as(RxLifecycleUtil.bindLifecycle(this))
    .subscribe();
复制代码

doOnNext操作符

doOnNext操作符类似于doOnEach(Action1) ，但是它的Action不是接受一个Notification参数，而是接受发射的数据项。

Observable.fromIterable(mItems)
    .doOnNext(new Consumer<Object>() {
        @Override
        public void accept(Object o) throws Exception {
            L.d(o.toString());
    .as(RxLifecycleUtil.bindLifecycle(this))
    .subscribe();
复制代码

doOnSubscribe操作符

doOnSubscribe操作符注册一个动作，当观察者订阅它生成的Observable它就会被调用。

注册时就调用，例如接口请求的Observable，注册时就弹出等待框。

Observable.fromIterable(mItems)
    .doOnSubscribe(new Consumer<Disposable>() {
        @Override
        public void accept(Disposable disposable) throws Exception {
        	  //准备开始
            showLoading();
    .as(RxLifecycleUtil.bindLifecycle(this))
    .subscribe();
复制代码

doOnComplete操作符

注册一个完成回调

Observable.fromIterable(mItems)
        .doOnComplete(new Action() {
            @Override
            public void run() throws Exception {
                //完成...
        .as(RxLifecycleUtil.bindLifecycle(this))
        .subscribe();
复制代码

doOnError操作符

注册一个出错回调

Observable.fromIterable(mItems)
    .doOnError(new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            //异常了
    .as(RxLifecycleUtil.bindLifecycle(this))
    .subscribe();
复制代码

doTerminate操作符

doTerminate操作符注册一个动作，当它产生的Observable终止之前会被调用，无论是正常还是异常终止。

Observable.fromIterable(mItems)
        .doOnTerminate(new Action() {
            @Override
            public void run() throws Exception {
                //...准备结束了
        .as(RxLifecycleUtil.bindLifecycle(this))
        .subscribe();
复制代码

doAfterTerminate操作符

doAfterTerminate操作符注册一个动作，当它产生的Observable终止之后会被调用，无论是正常还是异常终止。

结束时调用，接口请求成功或失败后，隐藏弹窗

Observable.fromIterable(mItems)
        .doAfterTerminate(new Action() {
            @Override
            public void run() throws Exception {
            	   //已经结束
                hideLoading();
        .as(RxLifecycleUtil.bindLifecycle(this))
        .subscribe();
复制代码

observeOn和subscribeOn操作符

ObserveOn：指定一个观察者在哪个调度器上观察这个Observable。可以说就是任务结束时进行回调的线程，而Android一般都是主线程。

SubscribeOn：指定Observable自身在哪个调度器上执行。可以说是耗时操作指定的线程，一般为IO线程或computation计算线程。

Android的Handler的Scheduler一般都使用RxAnroid上提供的，怎么引用就不说了，看Github。例如

Observable.fromIterable(mItems)
		  //Observable执行在子线程，所以在子线程遍历
        .subscribeOn(Schedulers.io())
         //回调线程，在主线程
        .observeOn(AndroidSchedulers.mainThread())
        .as(RxLifecycleUtil.bindLifecycle(this))
        .subscribe();
复制代码

timeout操作符

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

一般用来定时检查，Observable指定时间内时候发送过事件，超过时间发送一个onError()，异常对象为TimeoutException。

例如封装WebSocket的Observable，指定时间内没有发出事件，发出onError()事件，再配合retry重试，尝试重新连接WebSocket。

Observable
        .create(new WebSocketOnSubscribe(url))
        //如果数据源指定之间内没有发出消息，会发送一个超时异常，配合retry吃掉这个异常后，重试
        .timeout(timeout, timeUnit)
        .retry()
        //将回调都放置到主线程回调，外部调用方直接观察，实现响应回调方法做UI处理
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread());
复制代码

timestamp操作符

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

例如WebSocket实践中，后端要求我们定时多少秒发送一个WebSocket心跳消息。里面需要带一个时间戳，就可以使用timestamp操作符，包装数据为Timed对象。（内部其实就是map操作符，将数据包装在Timed对象，Timed对象中有个time字段为当前时间戳）

* 发送心跳包 public Observable < Boolean > sendHeartBeatMsg ( ) { return getRxWebSocket (). heartBeat ( getUrl (), AskTeacherConstant . CONSULTING_ROOM_PING_MSG_INTERVAL_TIME , TimeUnit . SECONDS , new HeartBeatGenerateCallback () { @Override public String onGenerateHeartBeatMsg ( long timestamp ) { return GsonUtil . toJson ( new HeartBeatMsgRequestModel ( WssCommandTypeEnum . HEART_BEAT . getCode (), String . valueOf (timestamp / 1000 ))); @Override public Observable < Boolean > heartBeat ( String url, int period, TimeUnit unit,HeartBeatGenerateCallback heartBeatGenerateCallback ) { return Observable . interval (period, unit) //timestamp操作符，给每个事件加一个时间戳 . timestamp () . retry () . flatMap ( new Function < Timed < Long >, ObservableSource < Boolean >>() { @Override public ObservableSource < Boolean > apply ( Timed < Long > timed) throws Exception { long timestamp = timed. time (); String heartBeatMsg = heartBeatGenerateCallback. onGenerateHeartBeatMsg (timestamp); Logger . d ( TAG , "发送心跳消息: " + heartBeatMsg); return send (url, heartBeatMsg); 复制代码

serialize操作符

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

Observable发射事件的线程是多个不同子线程（异步）进行发射，就可能造成事件混乱。可能onNext()、onError、onComplete顺序不是正确的，使用serialize操作符能使事件按同步顺序返回。

Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
        emitter.onNext("wally");
        emitter.onNext("wally");
        emitter.onComplete();
        emitter.onNext("wally");
.serialize()
.subscribe();
//wally
//wally
//wally
//onComplete
复制代码