準(zhǔn)確來講捷犹，RxAndroid 是隸屬于 ReactiveX 組織的住诸，Jake Wharton 作為參與者赎离，貢獻(xiàn)了大量的代碼（從 git 提交歷史記錄可查詢到）回懦，而且這個(gè)框架短小精悍态贤，不至于像 RxJava 那么龐大舱呻，讓人望而卻步，非常值得一讀悠汽，因此將她歸為【拆 Jake Wharton 系列】之一箱吕，這系列陸續(xù)創(chuàng)作中，歡迎關(guān)注柿冲。
（一） 你將獲得什么
每個(gè)框架有每個(gè)框架的使命茬高，閱讀源碼，可以挖掘相應(yīng)的技術(shù)點(diǎn)假抄，閱讀源碼的樂趣便在于此怎栽。通過閱讀 RxAndroid 源碼和本文，你將獲得：
- RxJava宿饱、RxAndroid 和 Android 的連接
- Rx 世界里鉤子的實(shí)現(xiàn)套路
- 高覆蓋率的單元測試
- Robolectric 對主線程的操縱
- 使用
CountDownLatch來測試線程
（二）RxAndroid 簡介
RxJava 中線程的變換和函數(shù)式編程與 Android 相得益彰熏瞄，但是 RxJava 并非為 Android 量身打造。在線程變換的過程中谬以，Android 有獨(dú)特的 UI 主線程的概念强饮，因此，需要一個(gè)框架來連接 Android 和 RxJava蛉签。RxAndroid充當(dāng)了該角色胡陪。
所以很明確沥寥，RxAndroid 的使命在于提供 Android 主線程的變換，代碼如下：
Observable.just("one", "two", "three", "four", "five")
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(/* an Observer */);
其中, AndroidSchedulers.mainThread() 便是這個(gè)框架提供的能力柠座。
說起 Android 主線程通訊問題邑雅，必不可少的關(guān)聯(lián)到 Handler、Looper妈经、Message淮野、HandlerThread等（可以查看筆者的另一系列文章《Handler 和他的小伙伴們》）。
因此 RxAndroid 提供了更通用的能力,可以指定任意的 Looper 來進(jìn)行任意線程之間的通訊：public static Scheduler from(Looper looper) ,代碼舉例如下：
mHandlerThread = new HandlerThread("HandlerThread");
mHandlerThread.start();
Observable.just("one", "two", "three", "four", "five")
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.from(mHandlerThread.getLooper())
.subscribe(/* an Observer */);
本文基于 RxJava 和 RxAndroid 2.0.1 進(jìn)行源碼分析吹泡，Github地址如下：
https://github.com/ReactiveX/RxAndroid/releases/tag/2.0.1
（三）源碼概覽
RxAndroid 非常簡潔骤星，只有四個(gè)類，為了增加趣味性爆哑，對于其中的三個(gè)核心類洞难，筆者稱之為面子、里子和鉤子揭朝。
面子和里子队贱，互為表里，面子是外在潭袱，靠里子支撐柱嫌；里子是內(nèi)涵，靠面子表現(xiàn)屯换。
關(guān)于面子和里子编丘，軟件世界里有個(gè)更專業(yè)的稱呼，叫門面模式彤悔。表里如一嘉抓，是值得尊敬的品格，現(xiàn)實(shí)世界如此蜗巧，軟件世界也是如此掌眠，而 RxAndroid 更是如此。
以下是這個(gè)框架的精華部分：
- AndroidSchedulers：面子幕屹，即框架的門面蓝丙，它的作用在簡介部分已經(jīng)說明。
- HandlerScheduler：里子望拖，這個(gè)類框架的核心渺尘，由它處理與 Android 主線程通訊的邏輯。
- RxAndroidPlugins：鉤子说敏，提供了行為的擴(kuò)展鸥跟。
- MainThreadDisposable：這個(gè)類是抽象類，提供了資源釋放的生命周期供重寫，它的作用是確保
onDispose()在主線程執(zhí)行医咨，這個(gè)類的解析非本文的重點(diǎn)枫匾。 - 大量的單元測試：UT 同樣是框架的精華部分，這個(gè)框架 UT 非常完善拟淮，值得學(xué)習(xí)干茉。
（四）面子 —— AndroidSchedulers
在上文的簡介中，我們已經(jīng)領(lǐng)略到 AndroidSchedulers 作為門面的簡潔很泊，它僅對外暴露了兩個(gè)方法角虫。這里重點(diǎn)分析一下 AndroidSchedulers.mainThread()。
在此之前委造，我們先看下命名戳鹅。RxJava 關(guān)于線程的取值，同樣也有個(gè)門面類昏兆，相關(guān)的代碼如下：
- Schedulers.io()
- Schedulers.computation()
- Schedulers.newThread()
- Schedulers.single()
Schedulers 和 AndroidSchedulers枫虏，以及 mainThread()和上述方法在命名和實(shí)現(xiàn)上保持了高度的一致性。
接下來回到源碼解析亮垫。mainThread 的實(shí)現(xiàn)并不復(fù)雜模软，但由于埋伏了兩個(gè)鉤子（RxAndroidPlugins）伟骨，代碼便顯得莫名其妙了饮潦，所以我們先忽視所有關(guān)于鉤子的邏輯，將代碼精簡如下：
public static Scheduler mainThread() {
return new HandlerScheduler(new Handler(Looper.getMainLooper()));
}
如此一來携狭，這個(gè)門面類的邏輯就十分簡單了继蜡，同時(shí)也呈現(xiàn)出了這個(gè)框架的里子——HandlerScheduler，并且持有了主線程的 Looper 對象逛腿。
（五）里子 —— HandlerScheduler
見名知意稀并，這是一個(gè)與 Handler 有關(guān)的 Scheduler。與上文一樣单默，我們先來討論下命名的事情碘举。在 RxJava 中，提供的默認(rèn)線程我們都可以找到對應(yīng)的實(shí)現(xiàn)搁廓，分別是：SingleScheduler引颈、ComputationScheduler、IoScheduler 和 NewThreadScheduler境蜕，因此蝙场，這仍然是一個(gè)固定套路。
以上所有的 XxxxxScheduler 都有個(gè)共同的抽象父類粱年，代碼精簡如下售滤，
public abstract class Scheduler {
public abstract Worker createWorker();
public abstract static class Worker implements Disposable {
public abstract Disposable schedule(Runnable run, long delay, TimeUnit unit);
}
}
因此 HandlerScheduler 只要根據(jù)父類的規(guī)范，做相應(yīng)的抽象方法實(shí)現(xiàn)即可，其中 Worker.schedule() 的重寫是關(guān)鍵完箩，由于涉及到 Android 主線程通訊赐俗，該方法的實(shí)現(xiàn)中將使用到 Handler 機(jī)制。所以我們先來簡單回顧下 Handler 怎么進(jìn)行主線程的通訊弊知。
當(dāng)我們需要與主線程通訊時(shí)秃励，發(fā)起的最終實(shí)現(xiàn)都是一致的：
handler.sendMessageAtTime(Message msg, long uptimeMillis)
那么主線程如何處理消息呢？共有兩種方式：
- 重寫 Handler 的 handleMessage()
- 為 Message 指定 callback 屬性（Runnable 類型）吉捶，消息發(fā)出后夺鲜，callback將會(huì)在主線程中回調(diào)。
以上兩種方式呐舔，通過閱讀 Handler 的 dispatchMessage() 源碼可獲知:
/**
* Handle system messages here.
*/
public void dispatchMessage(Message msg) {
if (msg.callback != null) {
handleCallback(msg);
} else {
if (mCallback != null) {
if (mCallback.handleMessage(msg)) {
return;
}
}
handleMessage(msg);
}
}
HandlerScheduler 使用了第2種方式與主線程通訊币励，源碼精簡如下：
@Override
public Worker createWorker() {
return new HandlerWorker(handler);
}
private static final class HandlerWorker extends Worker {
//省略部分代碼
@Override
public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
// 對run對象進(jìn)行代理，增加異常處理和釋放資源的邏輯
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
// 內(nèi)部將執(zhí)行message.callback = scheduled
Message message = Message.obtain(handler, scheduled);
message.obj = this;
handler.sendMessageDelayed(message, Math.max(0L, unit.toMillis(delay)));
return scheduled;
}
}
行文至此珊拼，還有一個(gè)很重要的問題未解決：** Message 對象的回調(diào)函數(shù) callback（Runnable 類型）的具體實(shí)現(xiàn)是什么食呻？ **
通過代碼 debug 可以輕易的獲取到答案，但在此之前澎现，我們先大膽預(yù)測一下：由于主線程執(zhí)行的是 Observer 實(shí)例中的 onNext()仅胞、onCompleted() 和 onError()，因此 Runnable 便是由 Observer 實(shí)例封裝而來剑辫，并且在合適的時(shí)機(jī)執(zhí)行上述三個(gè)方法干旧。
事實(shí)如我們預(yù)測的一樣，Runnable 的具體實(shí)現(xiàn)為 ObservableObserveOn 類中的 ObserveOnObserver 內(nèi)部類對象妹蔽，它是 Runnable 的實(shí)現(xiàn)類椎眯。
HandlerScheduler 實(shí)現(xiàn)了 RxJava 、RxAndroid 和 Android 之間的連接胳岂。
（六）鉤子 —— RxAndroidPlugins
何為鉤子
首先解釋下鉤子的概念编整。一面墻壁光滑明亮，自然是賞心悅目乳丰，但是具備的功能性就弱了些掌测，于是我們在墻壁上安置了一些鉤子，通過鉤子产园，我們可以掛上一些性感衣物或者一副世界名畫汞斧，這面墻壁的功能性便大大增強(qiáng)了。
墻壁便是 RxAndroid淆两，并且內(nèi)置了不少的鉤子断箫，找到這些鉤子后，可以做很多擴(kuò)展的事情秋冰，比如輸出日志仲义、異常處理和單元測試的輔助等。
縱觀 RxJava 和 RxAndroid 源碼，埋伏了大量的鉤子埃撵，這也是造成一些源碼閱讀起來比較費(fèi)解的原因所在赵颅。所有的鉤子的讀寫的邏輯都內(nèi)聚在 Plugins 中，RxJava 中是 RxJavaPlugins暂刘，RxAndroid 中則是 RxAndroidPlugins饺谬，同樣也保持了命名的一致性，而兩個(gè) Plugins 類谣拣，也可以認(rèn)為是所有鉤子的門面類募寨。
發(fā)現(xiàn)鉤子
在講面子這一節(jié)的時(shí)候，筆者對 AndroidSchedulers.mainThread() 源碼做了精簡森缠，實(shí)際上這里埋伏了個(gè)鉤子 onMainThreadHandler：
//MAIN_THREAD 同樣埋伏了鉤子拔鹰，此處不做介紹，最終將返回HandlerScheduler對象
public static Scheduler mainThread() {
return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD);
}
// 以下為 RxAndroidPlugins.java
public static Scheduler onMainThreadScheduler(Scheduler scheduler) {
// 鉤子：onMainThreadHandler
Function<Scheduler, Scheduler> f = onMainThreadHandler;
if (f == null) {
return scheduler;
}
return apply(f, scheduler);
}
// 為鉤子賦值
public static void setMainThreadSchedulerHandler(Function<Scheduler, Scheduler> handler) {
onMainThreadHandler = handler;
}
為了加深理解贵涵，可以對照下面的流程圖查看列肢，大部分的鉤子都是基于同樣的套路來實(shí)現(xiàn)的。
RxAndroid 和 RxJava 2.x 內(nèi)置了大量的鉤子宾茂，而他們都以 get 和 set 的形式對外部提供讀寫瓷马。如下圖：
對于鉤子，我們需要一些具體的實(shí)例來加深理解跨晴，并且希望從框架源碼本身來尋找實(shí)例欧聘，此時(shí)，單元測試將大展身手坟奥。
（七）單元測試是框架最好的說明書
鉤子的 UT 解讀
結(jié)合上一節(jié)树瞭，我們來解析下這個(gè)框架的單元測試，挖掘源碼本身的更多信息量爱谁。
針對鉤子的邏輯，我們一起來看下其中的一個(gè)測試方法：AndroidSchedulersTest 中的 mainThreadCallsThroughToHook()孝偎，方法名其實(shí)已經(jīng)表明的 UT 的測試意圖访敌，即對該場景進(jìn)行測試：通過鉤子（Hook）來執(zhí)行 AndroidSchedulers.mainThread() 方法。
這個(gè)例子告訴我們：
- 如何為鉤子賦值衣盾，并定義擴(kuò)展的行為
- 鉤子中擴(kuò)展的行為觸發(fā)的時(shí)機(jī)
所以說寺旺，單元測試是框架最好的說明書。
高覆蓋率
通過 AS 的 Run Tests with Coverage势决，數(shù)據(jù)表明：這個(gè)框架的單元測試行覆蓋率達(dá)到 91%阻塑，如下圖：
Robolectric 對主線程的操縱
RxAndroid 使用 Robolectric 對 Android 相關(guān)的邏輯進(jìn)行測試。通過 ShadowLooper 可以操縱主線程果复，如下圖所示陈莽，此 UT 位于 HandlerSchedulerTest 的 directScheduleOnceWithDelayPostsWithDelay() 。
這個(gè)例子告訴我們：
-
ShadowLooper.runUiThreadTasks()可以模擬主線程執(zhí)行 -
ShadowLooper.idleMainLooper()可以指定時(shí)間來阻塞主線程 - 除此之外
ShadowLooper還提供了很多好用的 api 來操縱主線程，可以通過 ShadowLooper 的源碼去了解這些 api 的用途走搁。
另外独柑，MainThreadDisposableTest 中的 UT 向我們展示了如何使用 CountDownLatch 來測試線程，有興趣的同學(xué)可以閱讀這部分源碼私植。
總而言之忌栅，一個(gè)優(yōu)秀的框架中的單元測試，既能幫助我們更好的了解框架本身曲稼，也能幫助我們提高單元測試的技巧索绪。
（八）總結(jié)
面子、里子和鉤子組成了 RxAndroid 的全部贫悄，而單元測試在此基礎(chǔ)上起到了錦上添花的作用者春，這依然是一個(gè)麻雀雖小五臟俱全的優(yōu)秀開源框架，每一個(gè)優(yōu)秀的框架清女，都是一本書一部電影钱烟，值得反復(fù)揣摩，用心研究嫡丙。
參考資料
https://github.com/ReactiveX/RxJava/wiki/What%27s-different-in-2.0
https://github.com/ReactiveX/RxAndroid
