RxAndroid深入理解

注：本文的分析基于RxAndroid 1.2.1

现在项目里面大多都已经使用了rxjava, 因此对于很多rxjava的扩展库，也都可以使用在项目里了。

RxAndroid 已经成为标配了，基本只要使用了Rxjava, 你肯定能看见RxAndroid的身影。

使用场景：

Observable.just("one", "two", "three", "four", "five")
        .subscribeOn(Schedulers.newThread())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(/* an Observer */);

很简单，就是指定到主线程上，即UI线程上，所以subscribe里的内容就会执行在主线程上。

或者其他线程的Looper

Looper backgroundLooper = // ...
Observable.just("one", "two", "three", "four", "five")
        .observeOn(AndroidSchedulers.from(backgroundLooper))
        .subscribe(/* an Observer */)

这种使用方式，就是把后台线程所绑定的Looper作为AndroidSchedulers的参数， 那么这时实际上等同于 ：

...
observeOn(Scheduler.io()); 

因此这种方式下，subscribe 里的代码是执行在backgroundLooper所绑定的线程。

代码结构：

Paste_Image.png

实现分析：

既然使用的入口是AndroidScheduler 那就重这里开始分析。

首先看看AndroidScheduler的创建：

Paste_Image.png

搭眼一看，饿汉式的单例， 并且这个单例是使用AtomicReference的，原子操作，保证多线程情况下，线程获取的值都是最新的，正确的。实例化用到了RxAndroidPlugins， 看下：

Paste_Image.png

同样是AtomicReference ，同样是单例。

Paste_Image.png

又引用到RxAndroidSechedulersHook

Paste_Image.png

这是个什么鬼，搞了半天最后getMainThreadScheduler() 返回null. 再回头看看前面AndroidScheduler实例化的地方。

Paste_Image.png

soga, 空的话直接返回一个LooperScheduler, 否则直接使用。

这就是问题了，为什么要绕这么大一圈？

我觉得秘密在RxAndroidSchedulersHook里的一个方法，如下：

Paste_Image.png

那么这个方法在哪里被调用了呢？ 这个问题先放着，后面会讲到。
先来看看注释，大概的意思是，该方法在把action 交给Scheduler处理之前被执行， 可以用于
包装/装饰/日志记录（翻译过来好别扭！）， 默认值只是传递作用，直接返回。

如果你仔细看看Rxjava RxJavaSchedulersHook 类，你会发现如出一辙， 一模一样！！！
连注释都长一样！
.
.

这里是把UI线程的Looper作为参数创建了一个LoopScheduler.

再进去看看LooperScheduler:

Paste_Image.png

LooperScheduler 继承自Scheduler， 因此必须实现createWorker方法。

Worker是个什么东东？

Paste_Image.png

实际上是Scheduler里 的一个静态抽象类，实现了Subscription接口，因此也就有unSubscriber方法。
可以取消订阅，但是Worker并没有实现取消订阅方法，而是交给子类去实现，比如这里的HandlerWorker就实现了Worker抽象类，下面是截取的核心部分：

Paste_Image.png

HandlerWorker实现了schedule方法，一个是有延时的，一个是无延时的。 从代码看首先判断了是否被取消订阅了， 取消了则直接返回，不做任何处理，否者接着走。
哎呦喂！ 哎呦喂！ 这不是RxAndroidSchedulersHook 里的那个我们之前放着的方法吗？
在这里被调用了。

接着走这里又跳出来个ScheduledAction， 贴出代码！

Paste_Image.png Paste_Image.png

从代码中不难看出， 在LooperScheduler 中创建的handler， 交给HandlerWorker后又交给了ScheduledAction。 该类实现了Runnable 接口，run方法里面调用了action.call() ;看来这里才是真正调用的地方，然后如果发生异常就调用RxjavaPlugin 触发错误处理。

ScheduledAction同样实现了Subscription接口， 取消也很简单，handler removeCallbacks(this).。

看完了ScheduledAction是不是感觉一目了然了，接着上面HandlerWorker的schedule方法讲，
后面从Message池中获取了一个Message并且把message的Callback参数设置为ScheduledAction的实例， 然后后面handler.sendMessageDelayed(message, unit.toMillis(delayTime)); 延时发送这条消息。 看到这你应该明白了，就是利用handler把消息发送到了looper所在的线程！ 因此你传递进去的Looper绑定在什么线程， 我们的action(订阅者)就在哪个线程执行！

而我们平时都是使用AndroidScheduler.mainThread(); 默认传递的就是UI线程的Looper!

有的小伙伴说，不对吧，我没看到接收处理消息的地方啊! 楼主你个坑！
实际上这就是handler你了解的不够细：

                                  (Handler.java)

Paste_Image.png

看到了吗？ 这里是handler接收到Looper分配的消息的处理， 先是判断了msg有没有callback, 有的话就不会调用handleMessage(msg)了，而是调用了handlerCallback(msg); 从而调用了runnable的run方法，也就是ScheduledAction的run()方法.

Paste_Image.png

.

RxAndroid中还有一个类没讲 MainThreadSubscription， 下篇文章会讲下RxBinding，里面会讲到，这里就不多说了。

http://www.jianshu.com/p/ded241051b61

说了这么多废话，希望对你有所帮助！ 如果有任何你认为不对的地方，请及时反馈给我，谢谢

.

点个赞， 谢谢！