浅析 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 线程

以上，水平有限，大佬轻喷。