浅析 RxJava 2.x 线程调度
RxJava 的大名不需要多做介绍，我自己也在项目中使用RxJava也有一段时间了，抱着学习的心态，一直想更深入一点的去了解它的实现原理，当然RxJava东西还是蛮多的，不可能一篇文章就能说清楚，so，这肯定是一个系列，在这一篇，我先尝试从 RxJava的线程控制 来切入，管中窥豹吧，我尽力，看官们随意。
最简单的使用
从我们日常使用的代码入手，可能能更容易理解一点。
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("Hello rxJava");
emitter.onComplete();
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.d("rxjava", s);
}
});
为了说明原理，排除干扰，这里用了最简单的逻辑。产生一个字符串“Hello rxJava”, 然后在监听中打印log，subscribeOn(Schedulers.io()) 设置数据在io线程产生，observeOn(AndroidSchedulers.mainThread()) 设置数据在主线程被监听，使用没问题，接下来我们来一步步探索实现原理。
怎么理解RxJava中的Observable、Observer、Scheduler
看字面就知道，Observable是被观察者，Observer是观察者，Scheduler是调度器，在RxJava实现中，Observable是一个抽象类，实现了ObservableSource接口。
Observable.png
ObservableSource 接口就一个方法 subscribe，参数接收一个Observer，所以使用的时候，就是通过subscribe方法注册观察者。下面是RxJava的实现：
ObservableSource.png
这里提一下，Observable 暴露了一个 钩子方法 subscribeActual(Observer observer),所有的Observable 子类都可以在这个方法里面实现自己的特殊逻辑。
Observable subscribe实现.png
线程调度 核心问题就2个，数据的发射的调用在哪个线程， 数据的监听在哪个线程，Scheduler 就是 干这个的（基本原理 就是工作线程的话就是 起线程池来处理任务，主线程就是利用handler 来 处理数据回调）。
捋一捋示例代码的调用逻辑
我们来捋一捋上面代码的调用逻辑，不难但是比较繁琐，这里只列出过程，具体实现可以看源码 -。-
- Observerable.create, 创建了一个ObservableCreate对象，这个类当然也继承了Observable，它的subscribeActual方法实现：创建一个CreateEmitter对象发射数据（调用Obserser对象onNext方法）
- Observerable.subscribeOn（Scheduler scheduler), 创建一个ObservableSubscribeOn对象，继承于Observable，它的subscribeActual方法实现：新建一个SuscribeOnObserver对象装饰原有的Observer，然后将新的Observer封装到SubscribeTask对象中(一个Runnable), 然后扔给scheduler调度
- Observerable.observeOn（Scheduler scheduler), 创建一个ObservableObserveOn对象，继承于Observable，它的subscribeActual方法实现：新建一个ObserveOnObserver对象装饰原有的Observer
有点懵逼，看图更直观
示例调用流程.png
简单来说，我们在这里只需要知道三类对象
- Observable 被观察者，在我们的图中，从下往上，运用装饰者模式，层层装饰增强
- Observer 观察者，我们的图中，从上往下，同样运用装饰者模式，层层装饰增强
- Scheduler 调度器，改变我们方法调用所在的线程, subscribeOn 影响Observable生成数据所在线程，observeOn 影响Observer监听所在线程。由于Observable的调用是 自下往上，所以 subscribe的线程 是在它后面调用的最近的一个 SubscribeOn方法 所设置的线程（可以理解为，只有第一个设置的SubscribeOn生效），而Observer的调用时自上往下的，所以 observe的线程 是在它前面调用的最近的一个 ObserveOn方法 所设置的线程。
举个实际一点的例子，假设有这样一个需求:获取后台配置的广告图，接口给的地址只是基础url，需要自行根据手机尺寸拼上长宽字段，然后下载图片到本地, 最后显示到ImageView。
ApiService.getAdvImageUrl()
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.map(new Function<String, String>() {
@Override
public String apply(String s) throws Exception {
return s + "/w=800/h=1080";
}
}).observeOn(Schedulers.newThread())
.map(new Function<String, Drawable>() {
@Override
public Drawable apply(String s) throws Exception {
return downloadImageToLocal(s);
}
}).observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Drawable>() {
@Override
public void accept(Drawable drawable) throws Exception {
//TODO
imageView.setImageDrawable(drawable);
}});
根据我们上面总结的线程切换规则
- 调用接口获取广告图片地址，在 io 线程
- 拼接图片长宽字段，在 main 线程
- 下载图片， 新建了一个 线程 执行
- 设置图片到ImageView， 在 main 线程
以上，水平有限，大佬轻喷。
