Android進階：5、RxJava2源碼解析 2

上一篇文章Android進階：4、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方法，完成一個鏈式的調用

八.寫在最後
若是喜歡個人文章，想與一羣資深開發者一塊兒交流學習的話，歡迎加入個人合做羣Android Senior Engineer技術交流羣。有flutter—性能優化—移動架構—資深UI工程師 —NDK相關專業人員和視頻教學資料
羣號：925019412