简介
这篇RxJava源码分析非常精简，仅通过最简单的调用方式对RxJava的整个调用流程做深入剖析，未贴大量源码，需要结合源码，对比流程分析，跟上思路。
先看下简单使用
Observable.create(new ObservableOnSubscribe<Object>() {
@Override
public void subscribe(ObservableEmitter<Object> emitter) throws Exception {
emitter.onNext("发射一条消息");
}
}).subscribe(new Observer<Object>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Object value) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
调用Observable的create时，返回一个ObservableCreate并传入ObservableOnSubscribe（即source）
@SchedulerSupport(SchedulerSupport.NONE)
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
链式调用Observable的订阅方法subscribe，传入观察者Observer
@SchedulerSupport(SchedulerSupport.NONE)
@Override
public final void subscribe(Observer<? super T> observer) {
ObjectHelper.requireNonNull(observer, "observer is null");
try {
observer = RxJavaPlugins.onSubscribe(this, observer);
ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
subscribeActual(observer);
} catch (NullPointerException e) { // NOPMD
throw e;
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
// can't call onError because no way to know if a Disposable has been set or not
// can't call onSubscribe because the call might have set a Subscription already
RxJavaPlugins.onError(e);
NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
npe.initCause(e);
throw npe;
}
}
再调用ObservableCreate的subscribeActual方法，可以看到创建了一个发射器其入参是观察者，然后立马调用了观察者的onSubscribe方法，最终再调用我们最初传入的source的subscribe方法。
整体流程详尽分析
Observable
.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(99);
}
})
.map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
if (integer == 99){
return "一切正常";
}
return "";
}
})
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String value) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
通过这段调用代码，对源码调用流程进行深入分析，精心绘制流程图如下：
image.png
整体流程概述，大概可分为三个阶段：
- 创建阶段（从上向下创建）：从Observable.create开始，创建一个ObservableCreate(传入一个发射消息源source即为ObservableOnSubscribe)返回-->然后再通过返回的ObservableCreate的.map创建一个ObserverableMap(并将自己传入作为ObserverableMap的source)，返回一个ObserverableMap。
- 订阅阶段（从下向上订阅）：这时候来到最后的.subscribe(传入终点观察者Observer)(即ObserverableMap的subscribe())时，会进而调用它的subscribeActual()，其内部则又会创建一个MapObserver(并传入终点观察者Observer)，调用创建阶段传入的source的.subscribe(传入MapObserver)，而这个source即为ObservableCreate --> 接着ObservableCreate也一样调用它的subscribeActual()，内部创建一个发射器CreateEmitter(并传入MapObserver)，再调用之前传入的source的.subscribe(传入发射器)，这个source其实就是最初通过Observable.create传入ObservableOnSubscribe
- 发射阶段（从上向下分发）：又回到了开始调用ObservableOnSubscribe.subscribe()，内部调用了emitter.onNext();这个CreateEmitter就是刚刚我们订阅时在ObservableCreate中创建的，它的onNext中又会调用我们创建发射器时传入的observer(即为MapObserver)的onNext()，而MapObserver.onNext()内部会通过创建ObserverableMap时传入的function进行转换，最后再调用真正的observer.onNext()，即最终的Observer。
如果上面明白了，加入线程切换其实就非常简单了，根据整体调用流向加入中间层。
.subscribeOn(Schedulers.io())
//Schedulers.io()返回一个线程池
.observeOn(AndroidSchedulers.mainThread())
//AndroidSchedulers.mainThread()返回一个持有主线程Handler
如上.subscribeOn()也只是包裹了一层Observerable而已，当向上订阅调用上层的source.subscribe()时，会把这句放入子线程执行，所以从此后向上订阅以及再向下分发的部分都是在子线程；
而.observeOn同样也只是包装了一层Observerable，当向下发射调用到observer.onNext时会通过Handler切换到主线程，后续的分发就都切换到主线程了。
