作者博客
http://www.cherylgood.cn/
前言
基于RxJava2.1.1
我们在前面的 RxJava2.0使用详解（一）初步分析了RxJava从创建到执行的流程。RxJava2.0使用详解（二) 中分析了RxJava的随意终止Reactive流的能力的来源；也明白了RxJava的onComplete();与onError(t);只有一个会被执行的秘密。
本次我们将探索RxJava2.x线程切换的实现原理。做到知其然，知其所以然。
Ok，开始我们的探索之旅吧！
从Demo到源码
本次我们将在上次的demo基础了做点改动。
版本1:不存在线程切换
observable.subscribe(observer);
输出结果：
版本2:切换线程（切换操作是如此的潇洒）
输出结果：
结果分析（因为我用的是@RunWith(AndroidJUnit4.class)执行代码，所以在工作线程是AndroidJUnitRunner）：
现在我们现象，后面根据现象分析原因。
没线程切换的版本：
1、在那里调用subscribe，则都在当前线程执行。
存在版本切换的版本：
1、被观察者的onSubscribe在调用subscribe的线程中执行
2、被观察者的subscribe在RxJava2的RxCachedThreadScheduler-1中运行。
3、onNext工作在主线程
OK，现象看完了，我们开始看本质吧！但是，从哪入手呢？还是老办法，哪里触发的行为就哪里下手(￣∇￣)
OK，我们先来探索切换Observable工作线程的subscribeOn方法入手
看到了RxJavaPlugins.onAssembly，前面分析过，为hook服务，现在看成是返回传入的Obserable即可。这里的this为我们的observable,scheduler就是我们传入的Schedulers.io();我们继续看ObservableSubscribeOn；
其继承AbstractObservableWithUpstream
AbstractObservableWithUpstream继承自Observable，其作用是通过source字段保存上游的Observable,因为Observable是ObservableSource接口的实现类，所以我们可以认为Observable和ObservableSource在本文中是相等的：
也就是说ObservableSubscribeOn是对Observble进行了一次wrapper操作
OK，我们继续回来看ObservableSubscribeOn的源码
Ok，我们开看下SubscribeTask
1、parent就是我们包装后的observer,其内部保存了下游的observer
2、source即通过ObservableSubscribeOnwrapper后存储我们上游的obserabler
所以run里面的source.subscribe(parent);即为wrapper的observer订阅了上游的observable,触发了上游observable的subscribeActual，开始执行数据的分发
上层obserable -》wrapper产生的observer -》真实的observser
思路梳理（重要）
Ok，分析到这里思路基本清晰了
1、在执行subscribeOn时，在Observable和observer中插入了一个(wrapper)obserabler和(wrapper)Observer
原Observabler->【(Wrapper)Observabler||(Wrapper)Observer】->（原）Observer
2、Observer.subscribe触发->(Wrapper)Observabler.subscribeActual()内部调用->parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));,->scheduler在指定线程调用(完成线程切换)->SubscribeTask.run,run内部调用->原Observabler.subscribe((Wrapper)Observer)触发->(原)Observabler.subscribeActual()开始数据分发
Ok,此时分发给的是(Wrapper)Observer，那应该是(Wrapper)Observer分发给了(原）Observer,我们看下是不是
OK，(Wrapper)Observer对(原)Observer进行了wrapper，我们看下源码：
Ok，确实是(Wrapper)Observer分发给了(原）Observer。
到这里的时候，整个流程基本OK了，但是，我们在5和11处说了,调度Worker也会加入Disposable进行管理，我还是要一探究竟。
Ok，有了对SubscribeOnObserver分析的铺垫，我们现在可以分析第5处parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));的代码了，我们先看scheduler.scheduleDirect()这句
OK，其返回一个Disposable,我们看下这个Disposable是否真的是调度的线程的。
Ok，现在终点转移到DisposeTask,我们把run给了DisposeTask，然后worker调度task开始执行run
OK,那么我们可以猜测w.schedule(task, delay, unit)执行后应该是调度了task的某个方法，然后task开始执行Runnable的run
是不是真的呢？我们来看下new DisposeTask(decoratedRun, w)做了什么
结论：
scheduler.scheduleDirect无延迟调用->worker->worker调度->DisposeTask->DisposeTask.run执行->decoratedRun.run();
decoratedRun即我们外部的SubscribeTask
总结
我们从subscribeOn入手分析了Observable线程切换的流程。其基本是通过中间插入包装类，也就是装饰者模式的体现，巧妙的实现了线程的切换。
其内部也对Disposed做了处理，保证Disposed的传递。
装饰者模式的使用贯穿了RxJava2的各处(个人理解），再次体会了设计模式的魅力。
由于本篇过长，observeOn订阅者线程的切换就再分一篇吧。
相关推荐