Rxjava observeOn()和subscribeOn()初探

Posted 2020-09-20

Rxjava这么强大的类库怎么可能没有多线程切换呢？

其中observeOn()与subscribeOn()就是实现这样的作用的。本文主要讲解observeOn()与subscribeOn()的用法，不去探究其中的原理。

0. 默认情况

在默认情况下，其不做任何线程处理，Observable和Observer处于同一线程，没有做任何线程切换，依次执行，如下图所示：

可以写一个demo测试之：

Observable<String> source = Observable.just("Alpha","Beta","Gamma");
        source.subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onNext(Integer integer) {
                    Log.i("TAG", "Received " + integer + " on thread:" +         
                    Thread.currentThread().getName());
            }
        });

1. subscribeOn()的作用

该方法是指明数据产生的线程，即Observable发射数据所在的线程，如果之后不做任何处理，操作符operator（如map，flatmap等）也在subscribeOn指定的线程做数据处理。

 

 

 

 

 

 

 

 

 

多次使用subscribeOn()并不能频繁地切换线程，只有距离数据源最近的一个subscribeOn()唯一确定数据源发射数据的线程。如代码所示：

Observable<String> source = Observable.just("Alpha","Beta","Gamma");
source
.subscribeOn(Schedulers.computation())
.subscribeOn(Schedulers.newThread())
.subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onNext(Integer integer) {
                    Log.i("TAG", "onNext: " + "on thread:" + Thread.currentThread().getName());
            }
        });
    }

其中只有subscribeOn(Schedulers.computation())对数据源source起作用，该source在Schedulers.computation()指定的线程发射数据。如果后面没有使用observeOn()，操作符operator都会在Schedulers.computation()所指定的线程做数据变换。

2. observeOn()的作用

在android开发中，我们经常面临这样的场景，在工作者线程中产生数据，在UI线程中更新相应的View，subscribeOn()指定了数据发射的线程，但我们更新UI的操作，不可能在发射数据的线程运行，这会造成ANR的问题。此时就必须通过observeOn()方法做线程的切换：

Observable<String> source = Observable.just("Alpha","Beta","Gamma");
        source.map(new Func1<String, Integer>() {
            @Override
            public Integer call(String s) {
                Log.i("TAG", "call: " + Thread.currentThread().getName());
                return s.length();
            }
        }).observeOn(Schedulers.newThread()).subscribeOn(Schedulers.computation()).
                subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onNext(Integer integer) {
                    Log.i("TAG", "onNext: " + "on thread:" + Thread.currentThread().getName());
            }
        });

上述代码的运行结果如下：

TAG: call: RxComputationScheduler-1
TAG: onNext: on thread:RxNewThreadScheduler-1
TAG: call: RxComputationScheduler-1
TAG: onNext: on thread:RxNewThreadScheduler-1
TAG: call: RxComputationScheduler-1
TAG: onNext: on thread:RxNewThreadScheduler-1

可以看到，在observeOn()之前的操作，都运行在subscribeOn(Schedulers.computation())指定的线程，即RxComputationScheduler-1线程；而使用了observeOn()之后，在它之后的操作都

运行在了observeOn(Schedulers.newThread())指定的线程。

所以，给出一个结论observeOn()只对其之后的操作起作用；observeOn()可以使用多次，每次使用对其之后的operator起作用，对之前的操作没有影响。

上图很好地诠释了ObserveOn的作用。

3. backpressure的问题

由于ObserveOn的作用，数据流在多个线程中不断的传输，可能存在速度不匹配的情况。如下图所示，当底部的数据流发射速度快于顶部数据流的处理速度，若产生异常，可能导致一部分数据未被顶部的subscriber处理。

 

废话太多，说不清楚，看下代码吧：

Observable<String> source = Observable.just("Alpha","Beta","Gamma");
        source.map(new Func1<String, Integer>() {
            @Override
            public Integer call(String s) {
                Log.i("TAG", "call: " + Thread.currentThread().getName());
                if (s.equals("Gamma"))
                    throw new RuntimeException();
                return s.length();
            }
        })
         .doOnError(new Action1<Throwable>() {
             @Override
             public void call(Throwable throwable) {
                  Log.i("TAG", "doOnError: " + Thread.currentThread().getName());
             }
         })
        .observeOn(Schedulers.newThread())
        .subscribeOn(Schedulers.computation())
        .subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {
                Log.i("TAG", "onError: " + "on thread:" + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Integer integer) {
                    Log.i("TAG", "onNext: " + "on thread:" + Thread.currentThread().getName());
            }
        });
    }

看一下运行结果：

TAG: call: RxComputationScheduler-1
TAG: call: RxComputationScheduler-1
TAG: call: RxComputationScheduler-1
TAG: doOnError: RxComputationScheduler-1
TAG: onError: on thread:RxNewThreadScheduler-1

我们就可以看到，当发射的数据为Gamma时抛出异常，之前发射的数据"Alpha","Beta"还未被subsriber的onNext方法处理，这就是backpressure问题。

4. onErrorResumeNext

onErrorResumeNext是错误恢复处理方法，当我们数据链中某个操作符抛出异常，此时会中断整个数据链，但我们想尝试恢复一下，这时可以使用

onErrorResumeNext。比如Android在过于频繁登录时，系统会弹出一个dialog（弹窗），让用户输入验证码，该逻辑就可以放在onErrorResumeNext中处理。

我们先看一段代码：

Observable<String> source = Observable.just("Alpha","Beta","Gamma");
        source.map(new Func1<String, Integer>() {
            @Override
            public Integer call(String s) {
                Log.i("TAG", "call: " + Thread.currentThread().getName());
                if (s.equals("Gamma"))
                    throw new RuntimeException();
                return s.length();
            }
        })
                .observeOn(Schedulers.newThread())
                .subscribeOn(Schedulers.computation())
                .onErrorResumeNext(new Func1<Throwable, Observable<? extends Integer>>() {
            @Override
            public Observable<? extends Integer> call(Throwable throwable) {
                return Observable.just(1000).map(new Func1<Integer, Integer>() {
                    @Override
                    public Integer call(Integer integer) {
                        Log.i("TAG", "call:onErrorResumeNext: "  +  Thread.currentThread().getName());
                        return integer;
                    }
                })
                .subscribeOn(Schedulers.computation());
            }
        })
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i("TAG", "onError: " + "on thread:" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i("TAG", "onNext: " + "on thread:" + Thread.currentThread().getName());
                    }
                });

看上段代码中onErrorResumeNext的部分，为什么在onErrorResumeNext可以再次使用subscribeOn()，我的猜测（并没有看源码）可能是该段代码产生了新的数据源，所以可以使用subsribeOn()指定数据源发射数据的线程。

它的运行结果：

TAG: call: RxComputationScheduler-1
TAG: call: RxComputationScheduler-1
TAG: call: RxComputationScheduler-1
TAG: onNext: on thread:RxNewThreadScheduler-1
TAG: onNext: on thread:RxNewThreadScheduler-1
TAG: call:onErrorResumeNext: RxComputationScheduler-2
TAG: onNext: on thread:RxComputationScheduler-2

看运行结果在onErrorResumeNext方法中使用了subscribeOn()，线程切换到了RxComputationScheduler-2，在之后没有observeOn的情况下，最后一个onNext也运行在了RxComputationScheduler-2。很神奇！！！！！