拆 Jake Wharton 系列之 RxAndroid

準(zhǔn)確來講，RxAndroid 是隸屬于 ReactiveX 組織的，Jake Wharton 作為參與者，貢獻(xiàn)了大量的代碼（從 git 提交歷史記錄可查詢到），而且這個框架短小精悍，不至于像 RxJava 那么龐大，讓人望而卻步，非常值得一讀，因此將她歸為【拆 Jake Wharton 系列】之一，這系列陸續(xù)創(chuàng)作中，歡迎關(guān)注。

（一） 你將獲得什么

每個框架有每個框架的使命，閱讀源碼，可以挖掘相應(yīng)的技術(shù)點(diǎn)，閱讀源碼的樂趣便在于此。通過閱讀 RxAndroid 源碼和本文，你將獲得：

  1. RxJava、RxAndroid 和 Android 的連接
  2. Rx 世界里鉤子的實現(xiàn)套路
  3. 高覆蓋率的單元測試
  4. Robolectric 對主線程的操縱
  5. 使用 CountDownLatch 來測試線程

（二）RxAndroid 簡介

RxJava 中線程的變換和函數(shù)式編程與 Android 相得益彰，但是 RxJava 并非為 Android 量身打造。在線程變換的過程中，Android 有獨(dú)特的 UI 主線程的概念，因此，需要一個框架來連接 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() 便是這個框架提供的能力。

說起 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 非常簡潔，只有四個類，為了增加趣味性，對于其中的三個核心類，筆者稱之為面子、里子和鉤子。

面子和里子，互為表里，面子是外在，靠里子支撐；里子是內(nèi)涵，靠面子表現(xiàn)。

關(guān)于面子和里子，軟件世界里有個更專業(yè)的稱呼，叫門面模式。表里如一，是值得尊敬的品格，現(xiàn)實世界如此，軟件世界也是如此，而 RxAndroid 更是如此。

以下是這個框架的精華部分：

  1. AndroidSchedulers：面子，即框架的門面，它的作用在簡介部分已經(jīng)說明。
  2. HandlerScheduler：里子，這個類框架的核心，由它處理與 Android 主線程通訊的邏輯。
  3. RxAndroidPlugins：鉤子，提供了行為的擴(kuò)展。
  4. MainThreadDisposable：這個類是抽象類，提供了資源釋放的生命周期供重寫，它的作用是確保 onDispose() 在主線程執(zhí)行，這個類的解析非本文的重點(diǎn)。
  5. 大量的單元測試：UT 同樣是框架的精華部分，這個框架 UT 非常完善，值得學(xué)習(xí)。

（四）面子 —— AndroidSchedulers

在上文的簡介中，我們已經(jīng)領(lǐng)略到 AndroidSchedulers 作為門面的簡潔，它僅對外暴露了兩個方法。這里重點(diǎn)分析一下 AndroidSchedulers.mainThread()。

在此之前，我們先看下命名。RxJava 關(guān)于線程的取值，同樣也有個門面類，相關(guān)的代碼如下：

  1. Schedulers.io()
  2. Schedulers.computation()
  3. Schedulers.newThread()
  4. Schedulers.single()

Schedulers 和 AndroidSchedulers，以及 mainThread()和上述方法在命名和實現(xiàn)上保持了高度的一致性。

接下來回到源碼解析。mainThread 的實現(xiàn)并不復(fù)雜，但由于埋伏了兩個鉤子（RxAndroidPlugins），代碼便顯得莫名其妙了，所以我們先忽視所有關(guān)于鉤子的邏輯，將代碼精簡如下：

public static Scheduler mainThread() {
        return new HandlerScheduler(new Handler(Looper.getMainLooper()));
    }

如此一來，這個門面類的邏輯就十分簡單了，同時也呈現(xiàn)出了這個框架的里子——HandlerScheduler，并且持有了主線程的 Looper 對象。

（五）里子 —— HandlerScheduler

見名知意，這是一個與 Handler 有關(guān)的 Scheduler。與上文一樣，我們先來討論下命名的事情。在 RxJava 中，提供的默認(rèn)線程我們都可以找到對應(yīng)的實現(xiàn)，分別是：SingleScheduler、ComputationScheduler、IoScheduler 和 NewThreadScheduler，因此，這仍然是一個固定套路。

以上所有的 XxxxxScheduler 都有個共同的抽象父類，代碼精簡如下，

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)的抽象方法實現(xiàn)即可，其中 Worker.schedule() 的重寫是關(guān)鍵，由于涉及到 Android 主線程通訊，該方法的實現(xiàn)中將使用到 Handler 機(jī)制。所以我們先來簡單回顧下 Handler 怎么進(jìn)行主線程的通訊。

當(dāng)我們需要與主線程通訊時，發(fā)起的最終實現(xiàn)都是一致的：

handler.sendMessageAtTime(Message msg, long uptimeMillis)

那么主線程如何處理消息呢？共有兩種方式：

  1. 重寫 Handler 的 handleMessage()
  2. 為 Message 指定 callback 屬性（Runnable 類型），消息發(fā)出后，callback將會在主線程中回調(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;
    }

}

行文至此，還有一個很重要的問題未解決：** Message 對象的回調(diào)函數(shù) callback（Runnable 類型）的具體實現(xiàn)是什么？ **

通過代碼 debug 可以輕易的獲取到答案，但在此之前，我們先大膽預(yù)測一下：由于主線程執(zhí)行的是 Observer 實例中的 onNext()、onCompleted() 和 onError()，因此 Runnable 便是由 Observer 實例封裝而來，并且在合適的時機(jī)執(zhí)行上述三個方法。

事實如我們預(yù)測的一樣，Runnable 的具體實現(xiàn)為 ObservableObserveOn 類中的 ObserveOnObserver 內(nèi)部類對象，它是 Runnable 的實現(xiàn)類。

HandlerScheduler 實現(xiàn)了 RxJava 、RxAndroid 和 Android 之間的連接。

（六）鉤子 —— RxAndroidPlugins

何為鉤子

首先解釋下鉤子的概念。一面墻壁光滑明亮，自然是賞心悅目，但是具備的功能性就弱了些，于是我們在墻壁上安置了一些鉤子，通過鉤子，我們可以掛上一些性感衣物或者一副世界名畫，這面墻壁的功能性便大大增強(qiáng)了。

墻壁便是 RxAndroid，并且內(nèi)置了不少的鉤子，找到這些鉤子后，可以做很多擴(kuò)展的事情，比如輸出日志、異常處理和單元測試的輔助等。

縱觀 RxJava 和 RxAndroid 源碼，埋伏了大量的鉤子，這也是造成一些源碼閱讀起來比較費(fèi)解的原因所在。所有的鉤子的讀寫的邏輯都內(nèi)聚在 Plugins 中，RxJava 中是 RxJavaPlugins，RxAndroid 中則是 RxAndroidPlugins，同樣也保持了命名的一致性，而兩個 Plugins 類，也可以認(rèn)為是所有鉤子的門面類。

發(fā)現(xiàn)鉤子

在講面子這一節(jié)的時候，筆者對 AndroidSchedulers.mainThread() 源碼做了精簡，實際上這里埋伏了個鉤子 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;
}

為了加深理解，可以對照下面的流程圖查看，大部分的鉤子都是基于同樣的套路來實現(xiàn)的。

鉤子的實現(xiàn)套路.png

RxAndroid 和 RxJava 2.x 內(nèi)置了大量的鉤子，而他們都以 get 和 set 的形式對外部提供讀寫。如下圖：

RxJava 中提供的鉤子

對于鉤子，我們需要一些具體的實例來加深理解，并且希望從框架源碼本身來尋找實例，此時，單元測試將大展身手。

（七）單元測試是框架最好的說明書

鉤子的 UT 解讀

結(jié)合上一節(jié)，我們來解析下這個框架的單元測試，挖掘源碼本身的更多信息量。

針對鉤子的邏輯，我們一起來看下其中的一個測試方法：AndroidSchedulersTest 中的 mainThreadCallsThroughToHook()，方法名其實已經(jīng)表明的 UT 的測試意圖，即對該場景進(jìn)行測試：通過鉤子（Hook）來執(zhí)行 AndroidSchedulers.mainThread() 方法。

鉤子的單元測試

這個例子告訴我們：

  • 如何為鉤子賦值，并定義擴(kuò)展的行為
  • 鉤子中擴(kuò)展的行為觸發(fā)的時機(jī)

所以說，單元測試是框架最好的說明書。

高覆蓋率

通過 AS 的 Run Tests with Coverage，數(shù)據(jù)表明：這個框架的單元測試行覆蓋率達(dá)到 91%，如下圖：


高覆蓋率

Robolectric 對主線程的操縱

RxAndroid 使用 Robolectric 對 Android 相關(guān)的邏輯進(jìn)行測試。通過 ShadowLooper 可以操縱主線程，如下圖所示，此 UT 位于 HandlerSchedulerTest 的 directScheduleOnceWithDelayPostsWithDelay() 。

Robolectric對主線程的操作

這個例子告訴我們：

  • ShadowLooper.runUiThreadTasks() 可以模擬主線程執(zhí)行
  • ShadowLooper.idleMainLooper() 可以指定時間來阻塞主線程
  • 除此之外 ShadowLooper 還提供了很多好用的 api 來操縱主線程，可以通過 ShadowLooper 的源碼去了解這些 api 的用途。

另外，MainThreadDisposableTest 中的 UT 向我們展示了如何使用 CountDownLatch 來測試線程，有興趣的同學(xué)可以閱讀這部分源碼。

總而言之，一個優(yōu)秀的框架中的單元測試，既能幫助我們更好的了解框架本身，也能幫助我們提高單元測試的技巧。

（八）總結(jié)

面子、里子和鉤子組成了 RxAndroid 的全部，而單元測試在此基礎(chǔ)上起到了錦上添花的作用，這依然是一個麻雀雖小五臟俱全的優(yōu)秀開源框架，每一個優(yōu)秀的框架，都是一本書一部電影，值得反復(fù)揣摩，用心研究。

參考資料

https://github.com/ReactiveX/RxJava/wiki/What%27s-different-in-2.0
https://github.com/ReactiveX/RxAndroid

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容