Android進階：五、RxJava2原始碼解析 2
上一篇文章Android進階：四、RxJava2 原始碼解析 1 裡我們講到Rxjava2 從建立一個事件到事件被觀察的過程原理，這篇文章我們講Rxjava2中鏈式呼叫的原理。本文不講用法，仍然需要讀者熟悉Rxjava基本的用法。
一.Rxjava2 的基本用法
Rxjava是解決非同步問題的，它的鏈式呼叫讓程式碼看起來非常流暢優雅。現在我們帶上執行緒切換以及鏈式呼叫來看看。下面程式碼是示例：
Observable
.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("a");
}
})
.subscribeOn(Schedulers.io())
.unsubscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.map(new Function<String, Integer>() {
@Override
public Integer apply(String s) throws Exception {
return 1;
}
})
.subscribe(new Observer<Object>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Object o) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
我們建立一個事件（觀察者），想輸出一個字串 "a"。這個事件發生在IO執行緒，結束也在IO執行緒，事件的狀態回調發生在主執行緒。示例的用法大家應該都能懂，我們主要討論這個鏈式的原理流程。為什麼這麼說呢？因為這個鏈式跟一般的鏈式不太一樣。
二.create方法
這個方法我們之前看過，返回一個ObservableCreate物件，ObservableCreate繼承自Observable，裡面的source存著我們建立的ObservableOnSubscribe匿名物件。
三.subscribeOn方法
這是Obserbvable的方法，先看原始碼：
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
Function<? super Observable, ? extends Observable> f = onObservableAssembly;
if (f != null) {
return apply(f, source);
}
return source;
}
程式碼結構跟create的差不多，在鉤子函式裡直接返回我們建立的物件ObservableSubscribeOn<T>(this, scheduler)，並傳入當前的Observable也就是ObservableCreate物件。所以我們看一下這個類的程式碼：
public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
super(source);
this.scheduler = scheduler;
}
}
這個類繼承自AbstractObservableWithUpstream類，建構函式的引數是ObservableSource，所以這裡我們需要介紹兩個類：
-
ObservableSource
ObservableSource是一個介面，所有的Observable都實現了這個介面，它裡面只有：
void subscribe(@NonNull Observer<? super T> observer);
這一個方法。很明顯這個方法是為了讓Observer訂閱Observable的，或者說為了Observable把事件狀態傳遞給Observer的。
-
AbstractObservableWithUpstream
這個類繼承了Observbable
abstract class AbstractObservableWithUpstream<T, U> extends Observable<U> {
protected final ObservableSource<T> source;
AbstractObservableWithUpstream(ObservableSource<T> source) {
this.source = source;
}
}
從原始碼可以看出這個類有變數source，它在建構函式裡傳入值，儲存ObservableSource物件。
所以當我們呼叫Observable的subscribeOn方法的時候會建立一個ObservableSubscribeOn物件，並用變數source儲存當前的Observable物件，然後返回ObservableSubscribeOn物件。
四.unsubscribeOn方法
public final Observable<T> unsubscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new ObservableUnsubscribeOn<T>(this, scheduler));
}
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
Function<? super Observable, ? extends Observable> f = onObservableAssembly;
if (f != null) {
return apply(f, source);
}
return source;
}
這個方法跟上面的方法是一個模子刻的。所以我們主要看ObservableUnsubscribeOn這個類就好。
public final class ObservableUnsubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
public ObservableUnsubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
super(source);
this.scheduler = scheduler;
}
}
這個類跟剛才的ObservableSubscribeOn也幾乎一模一樣，繼承自AbstractObservableWithUpstream類，使用source存了當前Observable物件。而此時的Observbvable物件是上一個方法建立的物件，也就是ObservableSubscribeOn物件。
五.observeOn方法和map方法
由於這些方法的內容基本一樣我就省略程式碼的解釋。
observeOn方法是建立了ObservableObserveOn物件，並儲存上一個方法建立的Observable。map方法是建立ObservableMap物件，並儲存上一個方法建立的Observable
所以總結一下可知：鏈式呼叫這些方法的時候，都會建立一個相關的物件，然後用變數source儲存上一個方法建立的Observable子類物件。
六.subscribe方法
上次文章講到，這個方法內部會呼叫一個抽象方法，subscribeActual方法，作為真實的訂閱。而這個方法的邏輯需要看子類如何實現。
而第一次呼叫該這個subscribe方法的物件是ObservableMap物件。所以我們看看它內部如何實現的。
ObservableMap的subscribeActual方法實現：
public void subscribeActual(Observer<? super U> t) {
source.subscribe(new MapObserver<T, U>(t, function));
}
內部呼叫了source的subscribe方法。此時ObservableMap物件裡存的source是上一個方法建立的observable，也就是ObservableObserveOn物件。所以我們要看看ObservableObserveOn是如何實現subscribeActual方法的：
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));
}
}
同理他最終也是呼叫了上一個Observable的subscribe。
於是我們知道當我們呼叫subscribe方法的時候，會遞迴式的呼叫source儲存的上一個方法建立的Observable的subscribeActual方法，一直到ObsservableCreate的subscribeActual的方法，把事件狀態傳遞給觀察者。這個上一篇文章已經講過。
七.總結
我們常見的普通的鏈式呼叫一般都會返回當前同一個物件。和普通的鏈式呼叫不同當我們呼叫Rxjava2的鏈式呼叫時，他們會返回自己對應的Observable子類物件，每個物件都不一樣，然後在subscribeActual方法中遞迴式的呼叫每個物件的subscribeActual方法，完成一個鏈式的呼叫。