目前RxJava+Retrofit+OkHttp是android中非常流行的Http通讯的解决方案，我也在工作中用这几个库结合我们自己服务器返回数据的格式封装了一套网络请求框架。这篇文章就工作中使用到的RxJava库的源码进行解析。
先放上典型的RxJava请求的代码
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("a");
emitter.onNext("b");
emitter.onNext("c");
emitter.onComplete();
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String o) {
Log.d("Rxjava", o);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
为了方便分析，我们还是把他改成我们熟悉的样式吧。
普通写法的RxJava调用
Observable o1 = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("a");
emitter.onNext("b");
emitter.onNext("c");
emitter.onComplete();
}
});
Observable o2 = o1.subscribeOn(Schedulers.io());
Observable o3 = o2.observeOn(AndroidSchedulers.mainThread());
o3.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String o) {
Log.d("Rxjava", o);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
好了，这下看着容易入手多了~
开始前我们要先了解几个接口和abstract类
Observable： 虚类，表示可观察对象
ObservableSource： 接口，表示可观察的资源，Observable类实现了该接口
ObservableOnSubscribe： 接口，我们需要实现该接口的void subscribe(@NonNull ObservableEmitter<T> emitter)方法来实现onNext，onError等逻辑调用，并且在Observable的create方法中传递过去来创建一个Observable对象，创建的Observable对象会持有该实现。
Observer： 接口，表示观察者
我们先看一下o1，o2，o3这几个Observable对象的创建过程。
这里使用了装饰器模式的思想，o1是实体对象，o1中的ObservableOnSubscribe对象的subscribe 方法，以及Observer的各个方法最终在这里被调用；o2持有o1对象引用，为ObservableOnSubscribe的subscribe 方法执行添加了多线程支持；o3持有o2对象，为Observer的各个方法执行提供了多线程（例子中为主线程）支持。
o1的创建方法Observable.create
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
第一行是检查传入对象是不是null，RxJavaPlugins里面的方法是对要返回的Observable对象进行了一层包装，方便我们对其进行诸如log记录等拦截操作，可以忽略。
我们主要看新创建的这个ObservableCreate对象，如下图，其继承自Observable类，并且持有我们传递进去的ObservableOnSubscribe。
再看o2是如何被创建的
Observable o2 = o1.subscribeOn(Schedulers.io());
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}
和o1的创建过程类似，subscrebeOn方法创建了一个ObservableSubscribeOn对象，该对象继承自Observable类，该对象持有o1的引用，以及Schedulers.io()返回的Scheduler对象。
Scheduler对象是用来进行线程切换的。
再看o3的创建过程
Observable o3 = o2.observeOn(AndroidSchedulers.mainThread());
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
}
observeOn方法返回了一个ObservableObserveOn对象，该对象继承自Observable类，该对象持有o2的引用，以及AndroidSchedulers.mainThread()返回的Scheduler对象。
下面就是o3.subscribe方法了，o3的真正类型是ObservableObserveOn，它的subscribe方法就是使用的超类Observable的
public final void subscribe(Observer<? super T> observer) {
...
observer = RxJavaPlugins.onSubscribe(this, observer);
ObjectHelper.requireNonNull(observer, "The RxJavaPlugins.onSubscribe hook returned a null Observer. Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");
subscribeActual(observer);
...
}
可以看到subscribe方法实际上调用了subscribeActual方法。再跟进去
@Override
protected void subscribeActual(Observer<? super T> observer) {
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}
这里的scheduler就是我们构建o3对象的时候传入的AndroidSchedulers.mainThread()方法返回的Scheduler对象，我们跟进去发现这并不是一个TrampolineScheduler，所以会从else分支里面走。
在这里首先使用传入的Observer对象构建了一个ObserveOnObserver对象，ObserveOnObserver类是o3的实际类型ObservableObserveOn的内部类，它实现了Observer接口，相当于是我们传入的Observer的代理，只是在调用相关方法的时候使用ObservableObserveOn对象内的Scheduler进行了线程切换。
随后调用了source的subscribe方法并传入新创建的ObserveOnObserver对象，这里的source就是我们的o2。
如上所述，对象o2的实际类型是ObservableSubscribeOn类，跟踪其subscribe方法，和o3一样，其最终也是调用的subscribeActual方法。
@Override
public void subscribeActual(final Observer<? super T> observer) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);
observer.onSubscribe(parent);
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
在这里，首先用传入的Observer对象构建了一个SubscribeOnObserver，SubscribeOnObserver类是ObservableSubscribeOn类的内部类，它实现了Observer接口，并对传入的Observer进行了简单包装，除此之外添加了取消自身执行的Disposable接口。
接着，使用创建的SubscribeOnObserver对象构建了一个SubscribeTask对象，它是一个Runnable对象。
SubscribeTask对象创建之后，使用了scheduler对象对其进行线程控制，该scheduler对象就是我们创建o2对象时传入的Schedulers.io()返回的scheduler对象，它会让该runnable在io线程中执行。
来看看SubscribeTask对象的具体情况，
final class SubscribeTask implements Runnable {
private final SubscribeOnObserver<T> parent;
SubscribeTask(SubscribeOnObserver<T> parent) {
this.parent = parent;
}
@Override
public void run() {
source.subscribe(parent);
}
}
它的run方法里面调用了source.subscribe，这里的source就是我们的o1对象。
上面说过，o1对象的实际类型是ObservableCreate类，其subscribe也同样调用的subscribeActual方法，来看看。
@Override
protected void subscribeActual(Observer<? super T> observer) {
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);
try {
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
它首先用传入的Observer对象构建了一个CreateEmitter对象，CreateEmitter类是ObservableCreate类的内部类，实现了ObservableEmitter接口。ObservableEmitter和Observer是对应的关系，都有onNext，onError，onComplete等方法，只是一个负责发射，一个负责接收。
创建的CreateEmitter对象可以看做是传入的Observer对象的代理。
CreateEmitter(Observer<? super T> observer) {
this.observer = observer;
}
@Override
public void onNext(T t) {
if (t == null) {
onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
return;
}
if (!isDisposed()) {
observer.onNext(t);
}
}
可以看到其内部的onNext方法基本上只是简单地调用了传入的observer的onNext方法。
回到subscribeActual方法，构建完CreateEmitter对象之后，调用了source.subscribe，这里的source是一个ObservableOnSubscribe，就是我们创建o1对象是传入的ObservableOnSubscribe对象。
这个subscribe被调用之后，其中的各个onNext方法就被依次执行了。
Over
