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，做了兩件事

1、nest() 為Observable<T>生成了一個Observable<Observable<T>> 
2、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的陷阱實在太大，內部類用的風生水起，一不小心，就不知道一個變數的上下文是什麼，需要特別小心。

之前我們分析過subscribeOn這個函式，
現在我們來看下subscribeOn和observeOn這兩個函式到底有什麼異同。

用過rxjava的旁友都知道，subscribeOn和observeOn都是用來切換執行緒用的，可是我什麼時候用subscribeOn，什麼時候用observeOn呢，我們很少知道這兩個區別是啥。

友情提示，如果不想看分析過程的，可以直接跳到下面的總結部分。

subscribeOn

先看下OperatorSubscribeOn的核心程式碼：

public final class OperatorSubscribeOn<T> implements OnSubscribe<T> {

    final Scheduler scheduler;
    final Observable<T> source;

    public OperatorSubscribeOn(Observable<T> source, Scheduler scheduler) {
        this.scheduler = scheduler;
        this.source = source;
    }

    @Override
    public void call(final Subscriber<? super T> subscriber) {
        final Worker inner = scheduler.createWorker();
        subscriber.add(inner);
        
        inner.schedule(new Action0() {
        
            @Override
            public void call() {
               
                Subscriber<T> s = new Subscriber<T>(subscriber) {
                    @Override
                    public void onNext(T t) {
                        subscriber.onNext(t);
                    }
                    
                    @Override
                    public void onError(Throwable e) {
                        try {
                            subscriber.onError(e);
                        } finally {
                            inner.unsubscribe();
                        }
                    }
                    
                    @Override
                    public void onCompleted() {
                        try {
                            subscriber.onCompleted();
                        } finally {
                            inner.unsubscribe();
                        }
                    }
                    
                    ....
                };
                
                source.unsafeSubscribe(s);
            }
        });
    }
}

這裡注意兩點：

1. 因為OperatorSubscribeOn是個OnSubscribe物件，所以在call引數中傳入的subscriber就是我們在外面使用Observable.subscribe(a)傳入的物件a。

2. 這裡source物件指向的是呼叫subscribeOn之前的那個Observable序列。

明確了這兩點，我們就很好的知道了subscribeOn是如何工作，產生神奇的效果了。
其實最最主要的就是一行函式

source.unsafeSubscribe(s);

並且要注意它所在的位置，是在worker的call裡面，說白了，就是把source.subscribe這一行呼叫放在指定的執行緒裡，那麼總結起來的結論就是：

subscribeOn的呼叫，改變了呼叫前序列所執行的執行緒。

observeOn

同樣看下OperatorObserveOn這個類的主要程式碼:

public final class OperatorObserveOn<T> implements Operator<T, T> {

    private final Scheduler scheduler;
    private final boolean delayError;

    /**
     * @param scheduler the scheduler to use
     * @param delayError delay errors until all normal events are emitted in the other thread?
     */
    public OperatorObserveOn(Scheduler scheduler, boolean delayError) {
        this.scheduler = scheduler;
        this.delayError = delayError;
    }

    @Override
    public Subscriber<? super T> call(Subscriber<? super T> child) {
        ....
        ObserveOnSubscriber<T> parent = new ObserveOnSubscriber<T>(scheduler, child, delayError);
        parent.init();
        return parent;
    }

    /** Observe through individual queue per observer. */
    private static final class ObserveOnSubscriber<T> extends Subscriber<T> implements Action0 {
        final Subscriber<? super T> child;
        final Scheduler.Worker recursiveScheduler;
        final NotificationLite<T> on;
        final boolean delayError;
        final Queue<Object> queue;
        
        // the status of the current stream
        volatile boolean finished;

        final AtomicLong requested = new AtomicLong();
        
        final Atomic