本文共 1860 字，大约阅读时间需要 6 分钟。

0．前言

看此篇之前最好知道RxJava的使用，关于框架的使用推荐。由于RxJava内部源码实现有点复杂，既然用拆轮子的方式来分析源码比较难啃，不如换种方式，以造轮子的方式，将源码中与性能、兼容性、扩展性有关的代码剔除，只留下核心代码，加上我个人的理解，带大家揭秘RxJava的实现原理（本文不涉及框架的使用介绍）。

1．观察者

Subsribler在RxJava里面是一个抽象类，它实现了Observer接口。

2．被观察者

Observable（被观察者）拥有很多工厂方法和各式各样的操作符。每个Observable里面都维护了一个OnSubscribe对象，并通过subscribe()里面的call(Subscriber<? super T> subscriber)方法与观察者产生联系。

3．RxJava的雏形

通过上面写的观察者和被观察者，即可写出一个没有操作符和线程切换功能的简易版Rxjava。

通过Observable.create将OnSubscribe的匿名类传给Observable，在subscribe()时回调OnSubscribe接口中的call方法，同时call方法参数即为subscribe的参数，即观察者，因此继续回调subscriber.onNext()即可完成观察者里的逻辑。

结果如下：

4．RxJava里的操作符

RxJava之所以强大好用，与其拥有丰富灵活的操作符是分不开的。那么我们就试着为这个框架添加一个最常用的操作符：map。先看代码：

//测试类中的代码

结果如下：

其实RxJava每调用一次操作符的方法，就相当于在上层数据源和下层观察者之间桥接了一个新的Observable。桥接的Observable内部会实例化新的OnSuscribe和Subscriber。

新建的OnSuscribe的call方法负责持有目标Subscriber，此时就可以回调subscriber的方法来完成观察的行为了。但是这是还没有数据源，想要获得数据源必须调用源Observable.OnSubscribe的subscribe方法，传入一个新的Subscriber，这样就可以在它的onNext()方法中获得数据源，并经过传入的接口处理后，发送给最终的Subscriber。

总体来说就是源Observable.OnSubscribe将Event往下发送给桥接Observable.Subscriber，最终桥接Observable.Subscriber将Event做相应处理后转发给目标Subscriber。

5．RxJava里的线程切换

RxJava中最激动人心的功能是异步处理，能够自如地切换线程。

利用subscribeOn() 结合observeOn() 来实现线程控制，让事件的产生和消费发生在不同的线程。 observeOn() 可以多次调用，Subscriber的执行线程与最后一次observeOn()的调用有关。但subscribeOn() 多次调用只有第一个subscribeOn() 起作用。

这是因为 observeOn() 作用的是Subscriber，而subscribeOn() 作用的是OnSubscribe，这时事件还没开始发送，因此subscribeOn()的线程控制可以从事件发出的开端就造成影响。

线程调度除了桥接Observable以外，RxJava还用到一个很关键的类Scheduler（调度器）。

Scheduler核心代码如下：

具体的Scheduler的实现类就不看了，但我们需要知道，能做到线程切换的关键是Worker的schedule方法，因为它会把传过来的任务放入线程池，并在新线程中执行。

5.1  实现observeOn

observeOn是作用于下层Subscriber的，需要让下层Subscriber的事件处理方法放到新线程中执行。为此，在Observable类里面，添加如下代码：

//测试代码如下

结果如下：

5.2  实现subscribeOn

subscribeOn是作用于上层OnSubscribe的，可以让OnSubscribe的call方法在新线程中执行。

因此，在Observable类里面，添加如下代码：

测试代码如下：

结果如下：

6  总结

相信看RxJava这个简易版的设计对大家的启示，比网上的一些源码解析清晰的多，希望可以抛砖引玉，大家有兴趣可以把flatMap等其他操作符来自己实现一下。