Rxjava -- 一個(gè)異步庫(kù)
RxJava最迷人的是什么夜涕？
答案就是把異步序列寫到一個(gè)工作流里朱嘴！和javascript的Promise/A如出一轍哄尔。
OK振愿，在java中做異步的事情在我們傳統(tǒng)理解過(guò)來(lái)可不方便，而且红选，如果要讓異步按照我們的工作流來(lái)澜公，就更困難了。
但是在RxJava中喇肋，我們只要調(diào)用調(diào)用
subscribOn()和observeOn()就能切換我們的工作線程坟乾，是不是讓小伙伴都驚呆了？
然后結(jié)合RxJava的Operator蝶防，寫異步的時(shí)候甚侣，想切換線程就是一行代碼的事情，整個(gè)workflow還非常清晰：
Observable.create()
// do something on io thread
.work() // work.. work..
.subscribeOn(Schedulers.io())
// observeOn android main thread
.observeOn(AndroidSchedulers.mainThread())
.subscribe();
我們?cè)僖膊挥萌懯裁匆?jiàn)鬼的new Thread和Handler了慧脱，在這么幾行代碼里渺绒，我們實(shí)現(xiàn)了在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));
}
啊宗兼，原來(lái)也是個(gè)lift躏鱼，就是從一個(gè)Observable生成另外一個(gè)Observable咯，這個(gè)nest是干嘛用殷绍？
public final Observable<Observable<T>> nest() {
return just(this);
}
這里返回類型告訴我們染苛，它是產(chǎn)生一個(gè)Observable<Observable<T>>
講到這里，會(huì)有點(diǎn)暈主到，先記著這個(gè)茶行，然后我們看OperatorSubscribeOn這個(gè)操作符,
構(gòu)造函數(shù)是
public OperatorSubscribeOn(Scheduler scheduler) {
this.scheduler = scheduler;
}
OK，這里保存了scheduler對(duì)象登钥，然后就是我們前一章說(shuō)過(guò)的轉(zhuǎn)換方法畔师。
@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);
}
});
}
}
});
}
});
}
});
}
};
}
讓人糾結(jié)的類模板
看完這段又臭又長(zhǎng)的，先深呼吸一口氣牧牢，我們慢慢分析下看锉。
首先要注意RxJava里面最讓人頭疼的模板問(wèn)題，那么OperatorMap這個(gè)類的聲明是
public final class OperatorMap<T, R> implements Operator<R, T>
而Operator這個(gè)接口繼承Func1
public interface Func1<T, R> extends Function {
R call(T t);
}
我們這里不要記T和R塔鳍，記住傳入左邊的模板是形參伯铣，傳入右邊的模板是返回值。
好了轮纫，那么這里的
call就是從一個(gè)T轉(zhuǎn)換成一個(gè)Observable<T>的過(guò)程了腔寡。
總結(jié)一下，我們這一次調(diào)用subscribeOn掌唾，做了兩件事
1放前、
nest()為Observable<T>生成了一個(gè)Observable<Observable<T>>
2、lift()對(duì)Observalbe<Observalbe<T>>進(jìn)行一個(gè)變化郑兴，變回Observable<T>
因?yàn)?code>lift是一個(gè)模板函數(shù)犀斋，它的返回值的類型是參照它的形參來(lái)，而他的形參是Operator<T, Observable<T>> 這個(gè)結(jié)論非常重要Ｇ榱叽粹！
OK，到這里我們已經(jīng)存儲(chǔ)了所有的序列却舀，等著我們調(diào)用了虫几。
調(diào)用鏈
首先，記錄我們?cè)谡{(diào)用這條指令之前的Observable<T>挽拔，記為Observable$1
然后辆脸，經(jīng)過(guò)lift生成的Observable<T>記為Observable$2
好了，現(xiàn)在我們拿到的依然是Observable<T>這個(gè)對(duì)象螃诅，但是它不是原始的Observable$1啡氢，要深深記住這一點(diǎn)状囱，它是由lift生成的Observable$2，這時(shí)候進(jìn)行subscribe倘是，那看到首先調(diào)用的就是OnSubscribe.call方法亭枷，好，直接進(jìn)入lift當(dāng)中生成的那個(gè)地方搀崭。
我們知道這一層lift的operator就是剛剛的OperatorSubscribOn叨粘，那么調(diào)用它的call方法，生成的是一個(gè)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) {
...
}
好瘤睹，還記得我們調(diào)用過(guò)nest么升敲？，這里的onSubscribe可是nest上下文中的噢轰传，每一次驴党，到這個(gè)地方，這個(gè)onSubscribe就是上一層Observable的onSubscribe绸吸，即Observable<Observable<T>>的onSubscribe鼻弧，相當(dāng)于棧彈出了一層设江。它的call直接在Subscriber的onNext中給出了最開(kāi)始的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);
}
});
}
});
}
}
對(duì)，就是它叉存，這里要注意码俩，這里的subscriber就是我們?cè)?code>lift中，傳入的o
Subscriber<? super T> st = hook.onLift(operator).call(o);
對(duì)歼捏，就是它稿存，其實(shí)它就是SafeSubscriber。
回過(guò)頭瞳秽，看看剛剛的onNext()方法瓣履，inner.schedule() 這個(gè)函數(shù)，我們可以認(rèn)為就是postRun()類似的方法练俐，而onNext()中傳入的o是我們之前生成的Observable$1袖迎，是從Observable.just封裝出來(lái)的Observable<Observable<T>>中產(chǎn)生的，這里調(diào)用了Observable$1.unsafeSubscribe方法腺晾，我們暫時(shí)不關(guān)心它和subscribe有什么不同燕锥，但是我們知道最終功能是一樣的就好了。
注意它運(yùn)行時(shí)的線程Ｃ醪酢归形！在
inner這個(gè)Worker上！于是它的運(yùn)行線程已經(jīng)被改了１怯伞暇榴！
好厚棵，這里的unsafeSubscribe調(diào)用的方法就是調(diào)用原先Observable$1.onSubscribe中的call方法：
這個(gè)Observable$1就是我們之前自己定義的Observable了。
綜上所述蔼紧，如果我們需要我們的Observable$1在一個(gè)別的線程上運(yùn)行的時(shí)候窟感，只需要在后面跟一個(gè)subscribeOn即可。結(jié)合扔物線大大的圖如下：
總結(jié)
這里邏輯著實(shí)不好理解歉井。如果還沒(méi)有理解的朋友柿祈，可以按照我前文說(shuō)的順序，細(xì)致的看下來(lái)哩至，我把邏輯過(guò)一遍之后躏嚎，發(fā)現(xiàn)lift的陷阱實(shí)在太大，內(nèi)部類用的風(fēng)生水起菩貌，一不小心卢佣，就不知道一個(gè)變量的上下文是什么，需要特別小心箭阶。
本文在不停更新中虚茶，如果有不明白的地方（可能會(huì)有很多），請(qǐng)大家給出意見(jiàn)仇参，拍磚請(qǐng)輕點(diǎn)= =
