拆轮子系列--RxJava理解（二）--subscribeOn-白红宇

本系列文章如下：

上一篇文章主要介绍了RxJava中基本的调用流程以及常用的操作符--map。本文主要介绍RxJava中线程调度的核心操作符之一subscribeOn，本文源码分析基于RxJava2。本文的大纲如下：

一个具体的例子

subscribeOn源码分析

总结

1 . 一个具体的例子

Observable.create(new ObservableOnSubscribe
    
     () {            @Override            public void subscribe(ObservableEmitter
     
       e) throws Exception {                e.onNext("1");                e.onNext("2");                e.onNext("3");                e.onComplete();            }        }).map(new Function
      
       () {            @Override            public Integer apply(String str) throws Exception {               Log.e("TAG", "-map-thread-" + Thread.currentThread().getName() + "-string-" + s);                return Integer.valueOf(s);            }        }).subscribeOn(Schedulers.io())             .subscribe(new Consumer
       
        () {              @Override              public void accept(Integer integer) throws Exception {                       Log.e("TAG", "-sub-" + Thread.currentThread().getName() + "-integer-" + integer);                    }                });    }复制代码

运行下上面的代码，我们先看看结果会是怎么样的：

E/TAG: -map-thread-RxCachedThreadScheduler-1-string-1E/TAG: -sub-RxCachedThreadScheduler-1-integer-1E/TAG: -map-thread-RxCachedThreadScheduler-1-string-2E/TAG: -sub-RxCachedThreadScheduler-1-integer-2E/TAG: -map-thread-RxCachedThreadScheduler-1-string-3E/TAG: -sub-RxCachedThreadScheduler-1-integer-3复制代码

如果我们调用多次的subscribeOn()会是怎么样子的呢？我们改下代码看看：

Observable.create(new ObservableOnSubscribe
    
     () {            @Override            public void subscribe(ObservableEmitter
     
       e) throws Exception {                e.onNext("1");                e.onNext("2");                e.onNext("3");                e.onComplete();            }        })                .subscribeOn(AndroidSchedulers.mainThread())                .map(new Function
      
       () {                    @Override                    public Integer apply(String s) throws Exception {                        Log.e("TAG", "-map-thread-" + Thread.currentThread().getName() + "-string-" + s);                        return Integer.valueOf(s);                    }                })                .subscribeOn(Schedulers.io())                .subscribe(new Consumer
       
        () {                    @Override                    public void accept(Integer integer) throws Exception {                        Log.e("TAG", "-sub-" + Thread.currentThread().getName() + "-integer-" + integer);                    }                });    }复制代码

我们再来看看打印的信息：

E/TAG: -map-thread-main-string-1E/TAG: -sub-main-integer-1E/TAG: -map-thread-main-string-2E/TAG: -sub-main-integer-2E/TAG: -map-thread-main-string-3E/TAG: -sub-main-integer-3复制代码

有没有发现，当我们调用两次的subscribeOn()这个操作符，只有最上面的那个起了作用，后面的subscribeOn()操作符无效。这个是为什么呢？接下来进行源码分析。

2. subscribeOn源码分析

首先，我们来看看subscribeOn()这个操作符到底干了些什么事情，然后再来解释为什么多次调用subscribeOn()这个操作符，只有第一个操作符有效。

@CheckReturnValue@SchedulerSupport(SchedulerSupport.CUSTOM)public final Observable
    
      subscribeOn(Scheduler scheduler) {   ObjectHelper.requireNonNull(scheduler, "scheduler is null");   return RxJavaPlugins.onAssembly(new ObservableSubscribeOn
     
      (this, scheduler));    }复制代码

我们首先看看onAssembly()这个里面到底干了些什么事情：

public static 
    
      Observable
     
       onAssembly(Observable
      
        source) {        Function
       
         f = onObservableAssembly;        if (f != null) {            return apply(f, source);        }        return source;    }复制代码

从源码中我们可以看出，它并没有干什么事情，就是返回了一个Observable的对象，接下来我们看看ObservableSubscribeOn()这部分的源码：

public ObservableSubscribeOn(ObservableSource
    
      source, Scheduler scheduler) {        super(source);        this.scheduler = scheduler;    }    @Override    public void subscribeActual(final Observer
      s) {    final SubscribeOnObserver
     
       parent = new SubscribeOnObserver
      
       (s);    s.onSubscribe(parent);    parent.setDisposable(scheduler.scheduleDirect(new Runnable() {         @Override         public void run() {              source.subscribe(parent);         }     }));  }复制代码

在上面的代码中，真正的执行操作是在subscribeActual()方法中，该方法中首先执行的操作就是通过代理的方式使用SubscribeOnObserver将Observer进行封装，设置Disposable来将subscribe进行线程切换到scheduler的线程中。简单点来说，subscribeOn这个操作符就是将上一层的ObservableSource(就是上一层的Observable)放到一个新的线程去发射元素。 ok，这里也就解释了为什么调用多个subscribeOn()时，只有最上面的subscribeOn()起作用，那么根据上面的分析，当我们在操作的时候加入多个map()操作符的时候，情况会是怎么样子的？如下示例：

Observable.create(new ObservableOnSubscribe
    
     () {            @Override            public void subscribe(ObservableEmitter
     
       e) throws Exception {                e.onNext("1");                e.onNext("2");                e.onComplete();            }        })                .map(new Function
      
       () {                    @Override                    public Integer apply(String s) throws Exception {                        Log.e("TAG", "-map-thread-" + Thread.currentThread().getName() + "-string-" + s);                        return Integer.valueOf(s);                    }                })                .subscribeOn(Schedulers.io())                .map(new Function
       
        () {                @Override                public Long apply(Integer integer) throws Exception {                        Log.e("TAG", "-map-thread-" + Thread.currentThread().getName() + "-Integer-" + integer);                        return Long.valueOf(integer);                    }                })                .subscribe(new Consumer
        
         () {                @Override                public void accept(Long longValue) throws Exception {                        Log.e("TAG", "-sub-" + Thread.currentThread().getName() + "-longValue-" + longValue);                    }                });复制代码

结果如下:

E/TAG: -map-thread-RxCachedThreadScheduler-1-string-1E/TAG: -map-thread-RxCachedThreadScheduler-1-Integer-1E/TAG: -sub-RxCachedThreadScheduler-1-longValue-1E/TAG: -map-thread-RxCachedThreadScheduler-1-string-2E/TAG: -map-thread-RxCachedThreadScheduler-1-Integer-2E/TAG: -sub-RxCachedThreadScheduler-1-longValue-2复制代码

从结果中我们发现，不管执行了多少次的map()操作，每个Observable只有第一个subscribeOn()方法起作用，而且多次的map()操作不会影响线程的变换。

3. 总结

首先，用一张图来展示subscribeOn()这个操作符基本运行的流程：

subscribeOn这个操作符主要的执行逻辑现在总结如下：

1. 代理Observable做的事情就是将Observable1发射到scheduler指定的线程中进行subscribe()，并且调用下一层传上来的observer1来接收事件；

1. 当我们多次调用subscribeOn()方法后，其实此时的事件并没有发射出去，当执行到第一个subscribeOn()方法时，后面subscribeOn()切换线程就会被第一个subscribeOn()切换线程的方法截断，因此，多次调用subscribeOn()对整个流程是不产生影响的，从而这里可以解释前面的问题，当使用多个subscribeOn()的时候，也只有第一个subscribeOn()起作用。

再举个浅显的栗子来说明下多次调用subscribeOn的情况：

new Thread(new Runnable() {            @Override            public void run() {                new Thread(new Runnable() {                    @Override                    public void run() {                        new Thread(new Runnable() {                            @Override                            public void run() {                                Log.e("TAG", "which thread?=" + Thread.currentThread().getName());                            }                        }, "线程3").start();                    }                }, "线程2").start();            }        }, "线程1").start();复制代码

上面代码执行的结果：

E/TAG: which thread?=线程3复制代码

1. 当我们使用subscribeOn()且使用多个map()操作符的时候，subscribeOn会将上一层Observable切换到一个指定的线程进行事件的发射，而上一层Observable拿到的observer已经是由下层observer一路封装上来的，也就是该observer经历了所有的转换。因此，不管执行了多少次的map()操作，不管该map()操作是执行在subscribeOn()之前还是之后，其最后的发射事件会在subscribeOn()指定的线程执行。

RxJava中线程调度方法subscribeOn()方法这里就介绍完毕，下一篇继续分析RxJava中另一个线程调度方法observableOn()。

如果文章中有什么疏漏或者错误的地方，还望各位指正，你们的监督是我最大的动力，谢谢！