本系列文章如下：

• 拆轮子系列--RxJava前奏篇
• 拆轮子系列--RxJava理解(一)--Map解析
• 拆轮子系列--RxJava理解(二)--subscribeOn
• 拆轮子系列--RxJava理解(三)--observeOn

上一篇文章属于RxJava的前奏篇，大体介绍了下为什么要使用RxJava，本篇文章主要从RxJava的基本使用流程入手来对RxJava的源码进行分析，本文的分析源码基于RxJava1来进行分析，RxJava2的使用核心思想区别不大，对于理解不产生什么影响。本文的大纲如下：

• RxJava的基本使用流程分析
• RxJava操作符--map源码解析
• 总结

1. RxJava的基本使用流程分析

首先，先以一个RxJava最普通的使用方式进行开始：

Observable.OnSubscribe<String> onSubscriber1 = new Observable.OnSubscribe<String>() {@Overridepublic void call(Subscriber<? super String> subscriber) {subscriber.onNext("17");subscriber.onCompleted();}};Subscriber<String> subscriber1 = new Subscriber<String>() {@Overridepublic void onCompleted() {}@Overridepublic void onError(Throwable e) {}@Overridepublic void onNext(String s) {}};Observable.create(onSubscriber1).subscribe(subscriber1);}
复制代码

上面的代码其实没有干什么事情，就是发射了一个String字符串。其意义在于，这个是RxJava最基本的一个订阅关系，而我们要分析的就是这个订阅关系是怎么工作的。 分析调用关系，最好就是查看源码，首先，我们看下create()这个方法的源码：

public final static <T> Observable<T> create(OnSubscribe<T> f) {return new Observable<T>(hook.onCreate(f));}
复制代码

从create()方法中我们可以看出，其内部就是创建了一个被观察者，这里我们称为Observable1。查看Observable的构造函数：

protected Observable(OnSubscribe<T> f) {this.onSubscribe = f;}
复制代码

在Observable的构造方法中，主要是做了两件事情：

1. 创建了一个observable1对象
2. 将我们传入的onSubscriber1保存在了observable1的onSubscribe属性中
复制代码

接下里分析subscrible()方法，首先，先看看源码：

public final Subscription subscribe(Subscriber<? super T> subscriber) {return Observable.subscribe(subscriber, this);}private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {...subscriber.onStart();hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);return hook.onSubscribeReturn(subscriber);}...}
复制代码

这里只是关注了几行核心代码，省略掉其余的代码。在subscribe()方法中，很明显的就是调用了hook.onSubscribeStart().call()这方法，接下来我们查看onSubscribeStart()方法：

public <T> OnSubscribe<T> su (Observable<? extends T> observableInstance, final OnSubscribe<T> onSubscribe) {return onSubscribe;}
复制代码

在onSubscribeStart()方法中，直接返回了observable1.onSubscribe。所以，subscribe()方法其实就是做了一件事情，就是调用我们栗子中的onSubscriber1的call()方法，而传入的对象就是栗子中的subscriber1，call()方法所执行的操作就是调用subscriber1对象的onNext()与onCompleted()方法。 我们用一张图总结上面方法的调用关系：

2. RxJava中常用操作符---map

在RxJava中，map操作符是一个很常用也很实用的操作符，现在我们主要分析下map操作的工作原理。首先，我们还是用一个例子进行说明：

Observable.OnSubscribe<String> onSubscriber1 = new Observable.OnSubscribe<String>() {@Overridepublic void call(Subscriber<? super String> subscriber) {subscriber.onNext("17");subscriber.onCompleted();}};Subscriber<Integer> subscriber1 = new Subscriber<Integer>() {@Overridepublic void onCompleted() {}@Overridepublic void onError(Throwable e) {}@Overridepublic void onNext(Integer result) {Log.w("TAG","output---" + result);}};Func1<String, Integer> transformer1 = new Func1<String, Integer>() {@Overridepublic Integer call(String s) {return Integer.parseInt(s);}};Observable.create(onSubscriber1).map(transformer1).subscribe(subscriber1);}
复制代码

上面的例子增加了一个map的操作符，主要的作用就是将String类型的“17”转换成int型进行log输出。 上面的类型转换其实是map操作符很常用的一个功能，现在我们具体分析下这个操作符的底层是怎么实现的：

public final <R> Observable<R> map (Func1<? super T, ? extends R> func) {return lift(new OperatorMap<T, R>(func));}
复制代码

首先还是走create()方法逻辑，创建了一个observable1，并且将传入的参数onSubscriber1保存在observable1的onSubscribe属性中，接下来我们重点关注下map操作符干了什么事情。 从源码中我们发现了两个新的东西，lift()与OperatorMap()。至于func1这个接口，它其实就是继承了Function，里面就只有一个call()方法。这里提及一下，在RxJava中会存在类似于Func+数字的接口，而这些接口中只有一个call()方法，类似的还有Action+数字的接口，它继承Action，里面页存在一个call()方法，与func不同的是，Action接口的call()方法都没有返回值，而Func接口的call()方法都存在返回值。这里我们将它们的源码贴出来进行对比：

/*** Represents a function with one argument.*/
public interface Func1<T, R> extends Function {R call(T t);
}/*** A one-argument action.*/
public interface Action1<T> extends Action {void call(T t);
}
复制代码

好了，言归正传，我们继续分析map。我们先看看OperatorMap到底做了什么操作：

public final class OperatorMap<T, R> implements Operator<R, T> {...}
复制代码

从源码中我们看到OperatorMap实现了Operator，那么这个Operator到底是个什么东西？看下源码：

   /*** Operator function for lifting into an Observable.*/public interface Operator<R, T> extends Func1<Subscriber<? super R>, Subscriber<? super T>> {...}
复制代码

好了，到这里也就明白了，Operator继承了Func1与Subscriber这个抽象类，前面提到过，func接口中只有一个带返回值的call()方法。也就是说，OperatorMap这个类其实就是重写了Func接口中的call()方法，至于做了什么事情，我们继续分析，先看下OperatorMap的构造函数：

public OperatorMap(Func1<? super T, ? extends R> transformer) {this.transformer = transformer;}
复制代码

看过源码之后，发现这里其实啥也没干，就是将传入进来的transformer保存到OperatorMap中的属性中。等等，我们之前调用map()操作符的时候传入了一个transformer1，那么这个transformer1就是保存在OperatorMap的transformer属性中。接下来我们看看OperatorMap中重写了func接口的call()方法，到底做什么事情呢，继续查看源码：

@Override
public Subscriber<? super T> call(final Subscriber<? super R> o) {return new Subscriber<T>(o) {@Overridepublic void onCompleted() {o.onCompleted();}@Overridepublic void onError(Throwable e) {o.onError(e);}@Overridepublic void onNext(T t) {try {o.onNext(transformer.call(t));} catch (Throwable e) {Exceptions.throwOrReport(e, this, t);}}};}
复制代码

从源码中发现，这个call()方法就是接受一个subscriber类型的参数，然后返回另外一个Subscriber类型的对象。其实我们可以把OperatorMap看成一个订阅者，这样就可以监听Observable发出来的事件。在call()方法中，其实最重要的就是o.onNext(transformer.call(t));这个方法。 局部总结下OperatorMap中做的事情：

• 1. 重写了Func接口的call()方法，接收一个subscriber类型的参数，返回另一个subscriber类型的对象
• 1. 将传入的transformer1对象保存在operatorMap的属性中
• 1. 在重写的call()方法中，调用了transformer.call(t)，就是调用了transformer1中的Integer.parseInt(s)，简单的说，就是传入的subscriber调用了其onNext()方法，执行了Integer.parseInt(s)并返回了一个新创建的Subscriber对象

接下来，继续分析下map操作符中的另一个比较重要的东西--lift(...)。老样子看看源码：

public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {return new Observable<R>(new OnSubscribe<R>() {@Overridepublic void call(Subscriber<? super R> o) {try {Subscriber<? super T> st = hook.onLift(operator).call(o);try {st.onStart();onSubscribe.call(st);} ...}});}
复制代码

截取了lift()中的重要代码，我们可以看到，lift()方法接收了一个Operator类型的参数，返回一个新创建的Observable类型对象。

3. 总结

分析了这么多，我们结合上面的例子，将map整个调用流程梳理一遍：

• 1. 首先，Observable.create(onSubscriber1)。这个里面做的操作就是创建了一个Observable对象observable1，并且将onSubscriber1对象保存在observable1对象的onSubscribe属性中
• 1. 执行了map(transformer1)操作。这里面的操作稍微复杂点。1) 先将transformer1对象保存在OperatorMap中的transformer属性中；2) 调用lift()方法，返回一个observable2；3) 在调用map()之后会调用subscribe()方法，也就是调用了observable2.subscribe()，同理，就是调用了observable2.onSubscribe.call()方法；4）call()方法首先做的事情就是传递一个subscriber1，返回一个subscribe对象，这里标记为subscribe_map；5）回到lift()方法继续执行，接着调用onSubscribe.call(st)，这里的st就是上一步返回的subscribe_map对象，也就是onSubscribe.call(subscribe_map); 6) onSubscribe对象就是我们第一步保存的onSubscriber1，所以现在的执行逻辑是onSubscriber1.call(subscribe_map)
• 1. 结合上面例子的情况，onSubscriber1.call(subscribe_map)的调用过程就是，先调用subscribe_map的call()，subscribe_map的call()操作会执行transformer1.call()，也就是执行了Integer.parseInt(“17”)，返回17; 接着调用onSubscriber1.call()，执行subscriber1.onNext()，输出17。 结合例子的情况，整个调用过程我们用一个流程图来进行直观阐述： 到此，添加了map操作符的这个例子基本上梳理完毕了，map操作符只是RxJava中一个常用的操作符，RxJava有很多灵活的操作符，这里就不一一列举了，操作符的合理使用往往能提升我们的开发效率，让我们的程序变得简洁。