常用辅助操作符

delay、do系列、subscribeOn、observeOn、timeout等

实例与功能介绍

1. delay操作符

代码实例

        Observable observable = Observable.just(1, 2, 3);
        observable = observable.delay(5, TimeUnit.SECONDS);

功能：可以让源Observable在发送每项数据之前都暂停一段指定的时间段。

2. do系列操作符

代码实例

        Observable observable = Observable.just(1, 2, 3);
        observable = observable.doOnNext(new Action1<Integer>() {
            @Override
            public void call(Integer integer) {
                LogShowUtil.addLog("RxJava","开始执行: "+integer.intValue(),true);
            }
        });

结果

开始执行: 1
结果: 1
开始执行: 2
结果: 2
开始执行: 3
结果: 3

功能：Do系列操作符就是为原始的Observable的生命周期注册一个回调，当Observable的某个时间发生时就会调用这些回调。

• doOnEach:Observable调用onNext()、onError()和onCompleted()时都会回调这个方法
• doOnNext:只在执行onNext()时会回调
• doOnSubscribe:订阅时会回调
• doOnUnsubscribe:取消订阅时会回调
• doOnCompleted:Observable正常终止时会回调
• doOnError:Observable异常终止时会回调
• doOnTerminate:Observable终止(包含正常和异常终止)之前时会回调
• finallyDo:Observable终止(包含正常和异常终止)之后时会回调

3. subscribeOn操作符

代码实例

        Observable observable = Observable.just(1, 2, 3);
        observable = observable.subscribeOn(Schedulers.newThread());

功能：指定Observable在哪个线程上执行

4. observeOn操作符

代码实例

        Observable Aobservable = Observable.create(new Observable.OnSubscribe<String>(){
            @Override
            public void call(Subscriber<? super String> subscriber) {
                LogShowUtil.addLog("RxJava","发送线程: "+Thread.currentThread().getName(),true);
                subscriber.onNext("杨");
                subscriber.onCompleted();
            }
        });
        Observable Bobservable = Aobservable.subscribeOn(Schedulers.newThread());
        Observable Cobservable = Bobservable.observeOn(AndroidSchedulers.mainThread());

功能：指定Observer在哪个线程上执行

5. timeout操作符

代码实例 1

        Observable observable = Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                for (int i=0; i<10;i++){
                    try {
                        Thread.sleep(i*100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    subscriber.onNext(i);
                }
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.newThread());

        observable = observable.timeout(200,TimeUnit.MILLISECONDS);

结果

结果: 0
结果: 1
结果: 2
异常

代码实例 2

        Observable observable = Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                for (int i=0; i<10;i++){
                    try {
                        Thread.sleep(i*100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    subscriber.onNext(i);
                }
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.newThread());

        observable = observable.timeout(200,TimeUnit.MILLISECONDS,Observable.just(10,11));

结果

结果: 0
结果: 1
结果: 2
结果: 11
结果: 12

功能：设置Observable过了指定的一段时间仍然没有发送任何数据就异常跳转。实例1是直接抛出异常，实例2是跳转执行其他Observable。