图解RxJava2(二) rxjava原理讲解

点击上方蓝字关注公众号

码个蛋第248次推文

今天有黄焖鸡。

作者：HuYounger

博客：http://rkhcy.github.io/

文章目录

• 概述

• 例子

• 源码分析

• 多次subscribeOn

• 最后

0

概述

接着《图解RxJava2(一)》这一片讲讲在 RxJava 中可以通过 subscribeOn/observeOn 很方便地完成上下游指定线程的切换，日常开发除了一些常用的Rx 操作符外，这两个方法也是打交道最多的。最初学习 RxJava 的时候总是死记硬背：subscribeOn 用于指定上游线程，observeOn 用于指定下游线程，多次用 subscribeOn 指定上游线程只有第一次有效，多次用 observeOn 指定下次线程，每次都有效…很久不用之后，总是把这两个方法搞混，那么这两个方法内部是怎么实现的呢？本篇先分析subscribeOn 方法。

1

例子

先回顾上篇文章的流程，饭店(Observable)开张前提要有厨师(ObservableOnSubscribe)，接着改名叫沙县小吃(ObservableCreate)，饭店接客(Observable.subscribe(observer))，创建服务员(CreateEmitter)把顾客和厨师关联起来，之后厨师每做一道菜都通过服务员端给顾客，整个流程如下：

我们都知道 Andriod 有主线程，在未指定线程切换操作的情况下，上图的流程是跑在主线程中，另外主线程中往往还存在其他任务需要执行，所以结合线程来看应该是这样的

上图给人一种感觉，好像厨师的菜是「秒做」出来的，然而我们都知道现实生活中厨师做菜是需要时间的，在安卓中，主线程执行耗时操作会阻塞后续的任务，还有可能引起 ANR，所以厨师做菜的操作不能放在主线程中 。下面让上游睡5秒模拟耗时操作

上游：

下游：

建立联系，以及执行其他任务(这里只是打了个 log )

打印如下：

可以看到，由于上游耗时，导致主线程中「其他任务」被阻塞了，因此需要新建一个子线程来处理上游的耗时任务，使用 RxJava 的 subscribeOn 就能轻松实现，修改代码：

打印如下：

此时「其他任务」不会被阻塞。从上面的 log 可以看到，创建了 RxNewThreadScheduler-1 的子线程来执行上游的耗时任务，并且此时下游除 onSubscribe 外，所有方法都执行在子线程中，它是怎么做到的？(通常情况下游会调用 observeOn(AndroidSchedulers.mainThread) 来更新UI，下篇分析)。

2

源码分析

上面的代码简短优雅，其实做了很多事情。基于上篇的分析，在执行完 Observable.create 和 new Observer 后此时主线程应该是下面的样子

Schedulers.newThread

Scheduler 翻译为调度器，RxJava2 中 Scheduler 的一些常用子类如下：

Schedulers.newThread 会初始化 NewThreadScheduler ；

上面的注释已经解释得很清楚了，在初始化 NewThreadScheduler 的时候会创建 RxThreadFactory，并指明了该线程工厂之后生产线程的名称和默认优先级；RxThreadFactory 是 ThreadFactory 的子类，也没多少代码

RxThreadFactory 中的 newThread 方法用来生产新线程。Schedulers.newThread 到此就完成了它的工作，总结下来就是：

1.创建线程调度器 NewThreadScheduler；

2.创建线程工厂 RxThreadFactory ；

到目前为止这些操作都是在主线程中执行的，子线程还未被创建。

subscribeOn(Scheduler scheduler)

该方法返回 Observable ，创建了 ObservableSubscribeOn ，名字起得又很容易让人头晕…这里就不画关系图了，只关心它的属性即可，它是 Observable(饭店) 的子类，结合我们举的例子，就给它起名黄焖鸡饭店；this 就是上面传过来的沙县小吃(ObservableCreate) ；初始化如下：

目前为止这些操作都是在主线程中执行，子线程还未创建

subscribe(Observer observer)

通过上篇学习可知，subscribe(observer) 内部会调用 subscribeActual(observer) ，该方法是个抽象方法，具体实现在 Observable(饭店) 的子类，现在是 ObservableSubscribeOn(黄焖鸡饭店)。

注释1 又冒出来一个 SubscribeOnObserver，同样只关心它的属性，SubscribeOnObserver 是AtomicReference的子类(保证原子性)，同时实现了 Observer(也是个顾客) 和 Disposable(保证一次性操作) 接口；为了方便理解，假设之前传的顾客叫小明，这里的顾客叫小红，小红会持有小明的引用(actual)，之后一系列的方法实际上会调用到小明的方法。

注释2 执行顾客小明的 onSubscribe 方法，我们发现到目前为止还没有创建过子线程，所以解释了上面 log 下游 onSubscribe 打印线程名为 main。

注释3 分为下面3步

步骤① SubscribeTask 是

ObservableSubscribeOn(黄焖鸡饭店) 的内部类，实现了 Runnable 接口

如果 run 方法被触发，那么执行顺序是：

Observable.subscribe —> Observable.subscribeActual —> ObservableCreate.subscribeActual，绕了一圈又回到上篇的那个流程。为了方便理解，SubscribeTask 就是黄焖鸡饭店(ObservableSubscribeOn)的「任务」也就是沙县小吃的「做菜」(ObservableCreate.subscribeActual)。所以现在万事具备，只差子线程了。

步骤② Scheduler.scheduleDirect

注意这个方法的注释，该方法调度的时候不保证顺序，所以平时在配合使用 subscribeOn(子线程)/observeOn(主线程) 会出现上下游输出顺序不确定的情况(比如有时候上游生产了3个后才逐个发送给下游，有时上游生产了2个，就开始发送给下游)，这也是多线程的一个特点。当然这里不会出现这个情况，因为从输出来看，此时上下游都在一个子线程里。貌似跑远了…继续分析

前面创建 NewThreadScheduler 的时候说 createWorker 方法很重要，这里派上用场了：

NewThreadWorker 内部维护一个线程池 ScheduledExecutorService ， 主要作用是提供延时调度和周期性调度，默认线程池大小为1，线程池里的线程通过我们传的线程工厂创建。

之后把 NewThreadWorker 和步骤①中的任务包装成 DisposeTask，又是一个Runnable

最后会执行 NewThreadWorker.schedule 方法

到这里终于看到任务(ObservableCreate.subscribeActual)执行在子线程中。

步骤③ parent.setDisposable 设置可中断。至此流程如下

之后所有的事情都是在子线程中进行的，上篇已经分析过了

后续还有：服务员端菜(CreateEmitter.onNext) —> 顾客小红拿到菜(SubscribeOnObserver.onNext) —> 顾客小明拿到菜(Observer.onNext)，模拟如下：

多次subscribeOn

上面我先把任务从一个线程切换到另一个线程，但是只有最先指定的有效(可以用 io 线程更容易看出差别)，这是为啥呢？通过上面的分析我们知道，subscribeOn 每次会返回一个 Observable ，为了方便理解，把先指定返回的Observable 叫黄焖鸡1号店，后指定返回的 Observable 叫黄焖鸡2号店，第一个 subscribeOn 执行：

黄焖鸡1号店创建的时候会持有沙县小吃的引用，接着第二个 subscribeOn 执行：

黄焖鸡2号店创建的时候会持有黄焖鸡1号店的引用，接着执行 subscribe(observer) 方法，会先调用黄焖鸡2号店的 subscribeActual 方法：

接着调用黄焖鸡2号店的 subscribeActual 方法 :

可以看到此时黄焖鸡1号店的 Worker 和小红是创建在子线程2的，并在子线程2中把当前线程切到了新的线程，后面的操作就和上面一样了，这就是为啥多次通过 subscribeOn 指定线程，只有最先指定的有效。

3

最后

多次用 subscribeOn 指定上游线程真的只有第一次有效吗？其实不然，具体可以看Dávid Karnok 的这篇博客，其中涉及到一些 Rx 操作符操作，本篇只是介绍 subscribeOn 的使用和原理，就不引入其他内容，mark 下日后再捡起来看。