Rxjava 2.1 订阅流程源码解析
一、前提说明
本文是在 Rxjava 2.1 的基础上进行的，目前只对 Rxjava 进行解析，未搭配 Retrofit 食用，如果想看 Rxjava + Retrofit 源码解析，请移步 Retrofit 2.1 + Rxjava 源码解析（一）。
二、Rxjava 使用栗子
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.e("TAG", "第二被调用，发射事件，subscribe(): ");
//发射器
emitter.onNext("1");
emitter.onNext("2");
emitter.onNext("3");
emitter.onComplete();
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.e("TAG", "最先被调用，onSubscribe(): ");
}
@Override
public void onNext(String s) {
Log.e("TAG", "onNext(): " + s);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
Log.e("TAG", "onComplete(): ");
}
});
在这里我们可以看到首先创建了一个 Observable 对象，然后调用订阅方法 subscribe() 就可以完成观察者模式的订阅。
三、Observable 和 ObservableOnSubscribe
Observable 是一个抽象类， ObservableOnSubscribe 是接口，我们看看它们里面有哪些方法：
public interface ObservableOnSubscribe<T> {
/**
* Called for each Observer that subscribes.
* @param e the safe emitter instance, never null
* @throws Exception on error
*/
void subscribe(@NonNull ObservableEmitter<T> e) throws Exception;
}
Observable 的接口方法就是我们常用的 api，如 subscribe()，combineLatest()，interval() 等等。
我们先看看 Observable.create(ObservableOnSubscribe) 里面做了什么：
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
//过滤掉无关紧要的代码
return new ObservableCreate<T>(source);
}
可以看到构造了 ObservableCreate 类的一个对象，猜测这个对象必然是 Observable 的子类，必然继承了 Observable 类。
public final class ObservableCreate<T> extends Observable<T> {
final ObservableOnSubscribe<T> source;
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}
......
}
好了，这里可以看到创建了一个 ObservableCreate 对象，它包含一个 ObservableOnSubscribe 对象。
四、subscribe()
我们从 ObservableCreate 累中没有看到 subscribe()，这么说肯定是在 Observable 类中：
public abstract class Observable<T> implements ObservableSource<T> {
@SchedulerSupport(SchedulerSupport.NONE)
@Override
public final void subscribe(Observer<? super T> observer) {
//过滤掉无关紧要的代码
subscribeActual(observer);
}
protected abstract void subscribeActual(Observer<? super T> observer);
......
}
我们回到 ObservableCreate 类中，查看 subscribeActual() ：
public final class ObservableCreate<T> extends Observable<T> {
final ObservableOnSubscribe<T> source;
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}
@Override
protected void subscribeActual(Observer<? super T> observer) {
//创建一个发射器
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
//这里可以看到一旦订阅后，首先回调的方法是 observer#onSubscribe（）
//可以在这个方法中做一些预处理
observer.onSubscribe(parent);
try {
//接着会调用 ObservableOnSubscribe#subscribe()
//这样就可以用发射器发送事件了
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
//发送 Error 事件
parent.onError(ex);
}
}
static final class CreateEmitter<T> extends AtomicReference<Disposable>
implements ObservableEmitter<T>, Disposable {
private static final long serialVersionUID = -3434801548987643227L;
final Observer<? super T> observer;
CreateEmitter(Observer<? super T> observer) {
this.observer = observer;
}
@Override
public void onNext(T t) {
if (!isDisposed()) {
observer.onNext(t);
}
}
@Override
public void onError(Throwable t) {
if (!tryOnError(t)) {
RxJavaPlugins.onError(t);
}
}
@Override
public boolean tryOnError(Throwable t) {
if (!isDisposed()) {
try {
observer.onError(t);
} finally {
dispose();
}
return true;
}
return false;
}
@Override
public void onComplete() {
if (!isDisposed()) {
try {
observer.onComplete();
} finally {
dispose();
}
}
}
@Override
public void dispose() {
DisposableHelper.dispose(this);
}
}
......
}
从源码中可以看到，一旦调用了 Observable.subscribe(Observer)，首先会调用 observer.onSubscribe(disposable);，然后会调用 ObservableOnSubscribe.subscribe(observableEmitter)，之后在我们的栗子中会调用发射器的发送 Next 事件和 Complete 事件。
从发射器 CreateEmitter 的源码可以看到，调用 CreateEmitter.onNext(T t) 和 CreateEmitter.onComplete() 会调用 Observer.onNext(t) 和 Observer.onComplete()，这个就是观察者模式，被观察者状态发生变化，通知观察者。
五、订阅流程小结
基本订阅流程按照我们上面的梳理就很清晰了。我们首先创建了一个被观察者 Observable（具体来说是一个 ObservableCreate 对象），然后订阅观察者 Observer，这样就将 Observer 传进了 Observable 里面，当 Observable 的状态发生改变或者 Observable 发送事件的时候，就能唤起观察者进行响应了。
