迷之RxJava （三）—— 線程切換
來自： http://segmentfault.com/a/1190000004051191
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);
}
});
}
}
});
}
});
}
});
}
};
}</pre>
讓人糾結的類模板
看完這段又臭又長的，先深呼吸一口氣，我們慢慢分析下。
首先要注意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的陷阱實在太大，內部類用的風生水起，一不小心，就不知道一個變量的上下文是什么，需要特別小心。
迷之RxJava（四）—— Retrofit和RxJava的基情
本文在不停更新中，如果有不明白的地方（可能會有很多），請大家給出意見，拍磚請輕點= =
