前言

歡迎來(lái)到大家深入理解 RxJava2 系列第二篇，這里先插上一句，本系列文章用的源碼都是基于 RxJava 2.2.0 正式版。本篇文章將先與大家一起理解 Scheduler 與 Worker ，順著 RxJava2 的源碼捋一下它們的實(shí)現(xiàn)原理。

Scheduler 與 Worker

Scheduler 與 Worker 在 RxJava2 中是一個(gè)非常重要的概念，他們是 RxJava 線程調(diào)度的核心與基石。用過(guò)的人肯定都會(huì)了解一些，但是想必了解 Worker 的讀者們就不多了。很多人會(huì)疑惑，既然有了 Scheduler 可以直接調(diào)度 Runnable，為何又強(qiáng)加一個(gè) Worker 的概念，諸位稍安勿躁，跟著筆者的思路一起走下去。

定義

筆者這里展示一下 Scheduler 最核心的定義部分：

public abstract class Scheduler {
    @NonNull
    public abstract Worker createWorker();
    public Disposable scheduleDirect(@NonNull Runnable run) {
        ...
    }
    public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
        ...
    }
    @NonNull
    public Disposable schedulePeriodicallyDirect(@NonNull Runnable run, long initialDelay, long period, @NonNull TimeUnit unit) {
        ...
    }
    public abstract static class Worker implements Disposable {
        @NonNull
        public Disposable schedule(@NonNull Runnable run) {
            ...
        }
        @NonNull
        public abstract Disposable schedule(@NonNull Runnable run, long delay, @NonNull TimeUnit unit);
        @NonNull
        public Disposable schedulePeriodically(@NonNull Runnable run, final long initialDelay, final long period, @NonNull final TimeUnit unit) {
            ...
        }
    }
}

從上面的定義可以看出，Scheduler 本質(zhì)上就是用來(lái)調(diào)度 Runnable 的，支持立即、延時(shí)和周期形式的調(diào)用，而 Worker 是任務(wù)的最小單元的載體。在 RxJava2 內(nèi)部的實(shí)現(xiàn)中，通常一個(gè)或者多個(gè) Worker 對(duì)應(yīng)一個(gè)ScheduledThreadPoolExecutor對(duì)象，這些暫且不表。

scheduleDirect / schedulePeriodicallyDirect

在 RxJava 1.x 時(shí)代， Scheduler 是沒(méi)有scheduleDirect/schedulePeriodicallyDirect的，只能夠先createWorker，再通過(guò) Worker 來(lái)調(diào)度任務(wù)。這些方法是對(duì) Worker 調(diào)用的簡(jiǎn)化，可以認(rèn)為是創(chuàng)建了一個(gè)只能調(diào)度一次任務(wù)的 Worker 并立馬調(diào)度了該任務(wù)。在Scheduler基類(lèi)的源碼中，也可以看出默認(rèn)的實(shí)現(xiàn)是直接 createWorker 并創(chuàng)建對(duì)應(yīng)的 Task 的（雖然在部分 Scheduler 覆蓋的實(shí)現(xiàn)上并沒(méi)有創(chuàng)建 Worker，但是可以認(rèn)為存在虛擬的 Worker）。

createWorker

一個(gè) Scheduler 可以創(chuàng)建多個(gè) Worker，這兩者是一對(duì)多的關(guān)系，而 Worker 與 Task 也是一對(duì)多的關(guān)系。

如下圖所示：

Worke 的存在為了確保兩件事：

• 同一個(gè) Worker 創(chuàng)建的 Task 都會(huì)確保串行，且立即執(zhí)行的任務(wù)符合先進(jìn)先出原則。
• Worker 綁定了調(diào)用了他的方法的 Runnable，當(dāng)該 Worker 取消時(shí)，基于他的 Task 均被取消

因此當(dāng)有操作符需要使用 Scheduler 時(shí)，可以通過(guò) Worker 來(lái)將一系列的 Runnable 統(tǒng)一的調(diào)度和取消，最典型的例子就是observeOn，下面會(huì)詳細(xì)分析。

Schedulers

RxJava2 默認(rèn)內(nèi)置了幾種 Scheduler 的實(shí)現(xiàn)，適用于不同的場(chǎng)景，這些 Scheduler 均在 Schedulers 類(lèi)中可以直接獲得

這里我們挑選兩個(gè)最常用的 computation / io 源碼稍作分析。

NewThreadWorker

NewThreadWorker 在 computation / io / newThread 均有涉及，我們先了解一下這個(gè)類(lèi)。

上面筆者有提到過(guò) Worker 與ScheduledThreadPoolExecutor 的關(guān)系，而這里的NewThreadWorker與ScheduledThreadPoolExecutor便是一對(duì)一的關(guān)系。在NewThreadWorker構(gòu)造函數(shù)中會(huì)通過(guò)工廠方法創(chuàng)建一個(gè)corePoolSize 為 1 的ScheduledThreadPoolExecutor對(duì)象并持有之。

ScheduledThreadPoolExecutor 從 JDK1.5 開(kāi)始存在，這個(gè)類(lèi)繼承于 ThreadPoolExecutor，可以支持即使、延時(shí)和周期的任務(wù)。但是注意在ScheduledThreadPoolExecutor中 maximumPoolSize 參數(shù)是無(wú)效的，corePoolSize 表示其最大線程數(shù)，且它的隊(duì)列是無(wú)界的。這里不再細(xì)說(shuō)該類(lèi)，否則涉及的就太多了。

有了這個(gè)類(lèi)，RxJava2 實(shí)現(xiàn) Worker 時(shí)便是站在了巨人的肩膀上，線程調(diào)度可以直接使用該類(lèi)解決，略微麻煩之處就是封一層Disposable的邏輯。

具體細(xì)節(jié)讀者可以從源碼一探究竟。

ComputationScheduler

作為計(jì)算密集型的 Scheduler，ComputationScheduler的線程數(shù)是與 CPU 核心密切相關(guān)的，原因是當(dāng)線程數(shù)遠(yuǎn)遠(yuǎn)超過(guò) CPU 核心數(shù)目時(shí)，CPU 的時(shí)間更多的損耗在了線程的上下文切換，因此比較通用的方式是保持最大線程數(shù)和 CPU 核心數(shù)一致。

最大線程數(shù)目

MAX_THREADS = cap(Runtime.getRuntime().availableProcessors(), Integer.getInteger(KEY_MAX_THREADS, 0));
static int cap(int cpuCount, int paramThreads) {
    return paramThreads <= 0 || paramThreads > cpuCount ? cpuCount : paramThreads;
}

從上面代碼可見(jiàn)MAX_THREADS 大于 0，但是不超過(guò) CPU 核心數(shù)，實(shí)際數(shù)值也受用戶設(shè)置的 System Properties 的影響。

FixedSchedulerPool

顧名思義，FixedSchedulerPool 可以認(rèn)為是固定數(shù)目的真正的 Worker 的緩存池。

確定了MAX_THREADS后，在ComputationScheduler的構(gòu)造函數(shù)，會(huì)創(chuàng)建FixedSchedulerPool對(duì)象，FixedSchedulerPool 內(nèi)部會(huì)直接創(chuàng)建一個(gè)長(zhǎng)度為MAX_THREADS的PoolWorker數(shù)組。PoolWorker繼承自NewThreadWorker，但是沒(méi)有任何額外的代碼。

static final class PoolWorker extends NewThreadWorker {
    PoolWorker(ThreadFactory threadFactory) {
        super(threadFactory);
    }
}

也就是說(shuō)當(dāng)FixedSchedulerPool創(chuàng)建時(shí)，已經(jīng)有MAX_THREADS個(gè) corePoolSize 為 1 的 ScheduledThreadPoolExecutor隨之創(chuàng)建。

PoolWorker

從使用角度來(lái)說(shuō)，有了FixedSchedulerPool 好像就夠了，我們只需要每次createWorker時(shí)從池子里取一個(gè)PoolWorker并返回即可。

但是這里忽略了一個(gè)要點(diǎn)，每個(gè) Worker 是獨(dú)立的，每個(gè) Worker 內(nèi)部的任務(wù)是綁定在這個(gè) Worker 中的。如果按照上述的做法，暴露出去PoolWorker，會(huì)出現(xiàn) 2 個(gè)問(wèn)題：

• createWorker 會(huì)可能會(huì)返回相同的 Worker，導(dǎo)致這個(gè) Worker 被 dispose 后，其內(nèi)部所有的任務(wù)會(huì)被一并取消，而違背了不同 Worker 之間的任務(wù)的獨(dú)立性
• PoolWorker也就是NewThreadWorker 被 dispose 后，其關(guān)聯(lián)的ScheduledThreadPoolExecutor被 shutdown，后續(xù)再次獲取該 Worker 也會(huì)導(dǎo)致無(wú)法創(chuàng)建任務(wù)

EventLoopWorker

為了解決上述的問(wèn)題，我們需要在PoolWorker外再包一層，createWorker每次都會(huì)創(chuàng)建一個(gè)EventLoopWorker對(duì)象。

EventLoopWorker 其實(shí)是個(gè)代理對(duì)象，他會(huì)將 Runnable 代理給FixedSchedulerPool中取到的PoolWorker來(lái)調(diào)度，并且他會(huì)負(fù)責(zé)管理經(jīng)由他創(chuàng)建的任務(wù)，當(dāng)自身被取消時(shí)，會(huì)將創(chuàng)建的任務(wù)統(tǒng)統(tǒng)取消。

示意圖

IoScheduler

與 ComputationScheduler 恰恰相反，IO 密集型的 Scheduler 線程數(shù)是無(wú)上限的。這是因?yàn)?IO 設(shè)備的速度是遠(yuǎn)遠(yuǎn)低于 CPU 速度的，在等待 IO 操作時(shí)， CPU 往往是閑置的，因此應(yīng)該創(chuàng)建更多的線程讓 CPU 盡可能的利用。當(dāng)然并不是說(shuō)線程越多越好，線程數(shù)目膨脹到一定程度既會(huì)影響 CPU 的效率，也會(huì)消耗大量的內(nèi)存。在IoScheduler中，每個(gè) Worker 在空置一段時(shí)間后就會(huì)被清除以控制線程的數(shù)目。

CachedWorkerPool

CachedWorkerPool是一個(gè)變長(zhǎng)并定期清理的ThreadWorker的緩存池，內(nèi)部通過(guò)一個(gè)ConcurrentLinkedQueue維護(hù)。和PoolWorker類(lèi)似，ThreadWorker也是繼承自NewThreadWorker：

static final class ThreadWorker extends NewThreadWorker {
    private long expirationTime;
    ThreadWorker(ThreadFactory threadFactory) {
        super(threadFactory);
        this.expirationTime = 0L;
    }
    public long getExpirationTime() {
        return expirationTime;
    }
    public void setExpirationTime(long expirationTime) {
        this.expirationTime = expirationTime;
    }
}

僅僅是增加了一個(gè)expirationTime字段，用來(lái)標(biāo)識(shí)這個(gè)ThreadWorker的超時(shí)時(shí)間。

于此同時(shí)，在CachedWorkerPool初始化時(shí)會(huì)傳入 Worker 的超時(shí)時(shí)間，目前是寫(xiě)死的 60 秒。這個(gè)超時(shí)時(shí)間表示ThreadWorker閑置后最大存活時(shí)間（實(shí)際中不保證 60 秒時(shí)被回收）。

EventLoopWorker

IoScheduler中也存在一個(gè)EventLoopWorker類(lèi)，它和ComputationScheduler中的作用也是類(lèi)似的：

• 管理自身調(diào)度過(guò)的任務(wù)
• 管理ThreadWorker，使其可被回收再次使用

Worker 的管理

• 創(chuàng)建：在閑置隊(duì)列中查找ThreadWorker，如果存在則取出，否則new``一個(gè)新的ThreadWorker，最后在外面包一層EventLoopWorker```并返回。
• 回收：當(dāng)EventLoopWorker dispose 后，會(huì)更新內(nèi)部的ThreadWorker超時(shí)時(shí)間，并促使CachedWorkerPool將ThreadWorker加入閑置隊(duì)列
• 清理：CachedWorkerPool在初始化時(shí)啟動(dòng)定時(shí)任務(wù)，每隔 60 秒清理隊(duì)列中超時(shí)的ThreadWorker

這里說(shuō)個(gè)細(xì)節(jié)，因?yàn)?code>CachedWorkerPool是每隔 60 秒清理一次隊(duì)列的，因此ThreadWorker的存活時(shí)間取決于入隊(duì)的時(shí)機(jī)，如果一直沒(méi)有被再次取出，其被實(shí)際清理的延遲在 60 - 120 秒之間，有興趣的讀者可以想一想為什么。

示意圖

對(duì)比

熟悉線程的讀者朋友們會(huì)發(fā)現(xiàn)，ComputationScheduler與IoScheduler很像某些參數(shù)下的ThreadPoolExecutor。

他們對(duì)線程的控制外在的表現(xiàn)很相似。 但是實(shí)際的線程執(zhí)行對(duì)象不一樣：

• ThreadPoolExecutor：Thread
• Scheduler：支持立即、延遲、定時(shí)調(diào)度任務(wù)的對(duì)象，通常為 ScheduledThreadPoolExecutor(coreSize = 1)

這兩者的對(duì)比有助于我們更加深刻地理解 Scheduler 設(shè)計(jì)的內(nèi)在邏輯。

結(jié)語(yǔ)

Scheduler 是 RxJava 線程的核心概念，RxJava 基于此屏蔽了 Thread 相關(guān)的概念，只與 Scheduler / Worker / Runnable 打交道。

以上就是RxJava2 Scheduler使用實(shí)例深入解析的詳細(xì)內(nèi)容，更多關(guān)于RxJava2 Scheduler使用的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！