謎之RxJava （三）—— 線程切換

【謎之RxJava （二） —— Magic Lift】

Rxjava -- 一個異步庫

RxJava最迷人的是什麼？
答案就是把異步序列寫到一個工做流裏！和javascript的Promise/A一模一樣。
OK，在java中作異步的事情在咱們傳統理解過來可不方便，並且，若是要讓異步按照咱們的工做流來，就更困難了。

可是在RxJava中，咱們只要調用調用
subscribOn()和observeOn()就能切換咱們的工做線程，是否是讓小夥伴都驚呆了？

而後結合RxJava的Operator，寫異步的時候，想切換線程就是一行代碼的事情，整個workflow還很是清晰：

Observable.create()
// do something on io thread
.work() // work.. work..
.subscribeOn(Schedulers.io())
// observeOn android main thread
.observeOn(AndroidSchedulers.mainThread())
.subscribe();

咱們不再用去寫什麼見鬼的new Thread和Handler了，在這麼幾行代碼裏，咱們實現了在io線程上作咱們的工做(work)，在main線程上，更新UI

Subscribe On

先看下subscribeOn幹了什麼

public final Observable<T> subscribeOn(Scheduler scheduler) {
    if (this instanceof ScalarSynchronousObservable) {
        return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);
    }
    return nest().lift(new OperatorSubscribeOn<T>(scheduler));
}

啊，原來也是個lift，就是從一個Observable生成另一個Observable咯，這個nest是幹嗎用？

public final Observable<Observable<T>> nest() {
    return just(this);
}

這裏返回類型告訴咱們，它是產生一個Observable<Observable<T>>
講到這裏，會有點暈，先記着這個，而後咱們看OperatorSubscribeOn這個操做符,

構造函數是

public OperatorSubscribeOn(Scheduler scheduler) {
    this.scheduler = scheduler;
}

OK，這裏保存了scheduler對象，而後就是咱們前一章說過的轉換方法。

@Override
public Subscriber<? super Observable<T>> call(final Subscriber<? super T> subscriber) {
    final Worker inner = scheduler.createWorker();
    subscriber.add(inner);
    return new Subscriber<Observable<T>>(subscriber) {

        @Override
        public void onCompleted() {
            // ignore because this is a nested Observable and we expect only 1 Observable<T> emitted to onNext
        }

        @Override
        public void onError(Throwable e) {
            subscriber.onError(e);
        }

        @Override
        public void onNext(final Observable<T> o) {
            inner.schedule(new Action0() {

                @Override
                public void call() {
                    final Thread t = Thread.currentThread();
                    o.unsafeSubscribe(new Subscriber<T>(subscriber) {

                        @Override
                        public void onCompleted() {
                            subscriber.onCompleted();
                        }

                        @Override
                        public void onError(Throwable e) {
                            subscriber.onError(e);
                        }

                        @Override
                        public void onNext(T t) {
                            subscriber.onNext(t);
                        }

                        @Override
                        public void setProducer(final Producer producer) {
                            subscriber.setProducer(new Producer() {

                                @Override
                                public void request(final long n) {
                                    if (Thread.currentThread() == t) {
                                        // don't schedule if we're already on the thread (primarily for first setProducer call)
                                        // see unit test 'testSetProducerSynchronousRequest' for more context on this
                                        producer.request(n);
                                    } else {
                                        inner.schedule(new Action0() {

                                            @Override
                                            public void call() {
                                                producer.request(n);
                                            }
                                        });
                                    }
                                }

                            });
                        }

                    });
                }
            });
        }

    };
}

讓人糾結的類模板

看完這段又臭又長的，先深呼吸一口氣，咱們慢慢分析下。
首先要注意RxJava裏面最讓人頭疼的模板問題，那麼OperatorMap這個類的聲明是

public final class OperatorMap<T, R> implements Operator<R, T>

而Operator這個接口繼承Func1

public interface Func1<T, R> extends Function {
    R call(T t);
}

咱們這裏不要記T和R，記住傳入左邊的模板是形參，傳入右邊的模板是返回值。

好了，那麼這裏的call就是從一個T轉換成一個Observable<T>的過程了。

總結一下，咱們這一次調用subscribeOn，作了兩件事

一、nest() 爲Observable<T>生成了一個Observable<Observable<T>>
二、lift() 對Observalbe<Observalbe<T>>進行一個變化，變回Observable<T>

由於lift是一個模板函數，它的返回值的類型是參照它的形參來，而他的形參是Operator<T, Observable<T>> 這個結論很是重要！！
OK，到這裏咱們已經存儲了全部的序列，等着咱們調用了。

調用鏈

首先，記錄咱們在調用這條指令以前的Observable<T>，記爲Observable$1
而後，通過lift生成的Observable<T>記爲Observable$2

好了，如今咱們拿到的依然是Observable<T>這個對象，可是它不是原始的Observable$1，要深深記住這一點，它是由lift生成的Observable$2，這時候進行subscribe，那看到首先調用的就是OnSubscribe.call方法，好，直接進入lift當中生成的那個地方。

咱們知道這一層lift的operator就是剛剛的OperatorSubscribOn，那麼調用它的call方法，生成的是一個Subscriber<Observable<T>>

Subscriber<? super T> st = hook.onLift(operator).call(o);
try {
    // new Subscriber created and being subscribed with so 'onStart' it
    st.onStart();
    onSubscribe.call(st);
} catch (Throwable e) {
...
}

好，還記得咱們調用過nest麼？，這裏的onSubscribe但是nest上下文中的噢，每一次，到這個地方，這個onSubscribe就是上一層Observable的onSubscribe，即Observable<Observable<T>>的onSubscribe，至關於棧彈出了一層。它的call直接在Subscriber的onNext中給出了最開始的Observable<T>，咱們這裏就要看下剛剛在OperatorSubscribeOn中生成的Subscriber

new Subscriber<Observable<T>>(subscriber) {

    @Override
    public void onCompleted() {
        // ignore because this is a nested Observable and we expect only 1 Observable<T> emitted to onNext
    }
    
    @Override
    public void onError(Throwable e) {
        subscriber.onError(e);
    }
    
    @Override
    public void onNext(final Observable<T> o) {
        inner.schedule(new Action0() {
    
            @Override
            public void call() {
                final Thread t = Thread.currentThread();
                o.unsafeSubscribe(new Subscriber<T>(subscriber) {
    
                    @Override
                    public void onCompleted() {
                        subscriber.onCompleted();
                    }
    
                    @Override
                    public void onError(Throwable e) {
                        subscriber.onError(e);
                    }
    
                    @Override
                    public void onNext(T t) {
                        subscriber.onNext(t);
                    }
                });
            }
        });
    }
}

對，就是它，這裏要注意，這裏的subscriber就是咱們在lift中，傳入的o

Subscriber<? super T> st = hook.onLift(operator).call(o);

對，就是它，其實它就是SafeSubscriber。

回過頭，看看剛剛的onNext()方法，inner.schedule() 這個函數，咱們能夠認爲就是postRun()相似的方法，而onNext()中傳入的o是咱們以前生成的Observable$1，是從Observable.just封裝出來的Observable<Observable<T>>中產生的，這裏調用了Observable$1.unsafeSubscribe方法，咱們暫時不關心它和subscribe有什麼不一樣，可是咱們知道最終功能是同樣的就行了。

注意它運行時的線程！！在inner這個Worker上！因而它的運行線程已經被改了！！

好，這裏的unsafeSubscribe調用的方法就是調用原先Observable$1.onSubscribe中的call方法：
這個Observable$1就是咱們以前本身定義的Observable了。

綜上所述，若是咱們須要咱們的Observable$1在一個別的線程上運行的時候，只須要在後面跟一個subscribeOn便可。結合扔物線大大的圖以下：

總結

這裏邏輯着實很差理解。若是尚未理解的朋友，能夠按照我前文說的順序，細緻的看下來，我把邏輯過一遍以後，發現lift的陷阱實在太大，內部類用的風生水起，一不當心，就不知道一個變量的上下文是什麼，須要特別當心。

迷之RxJava（四）—— Retrofit和RxJava的基情

本文在不停更新中，若是有不明白的地方（可能會有不少），請你們給出意見，拍磚請輕點= =