RxJava2 Scheduler使用實(shí)例深入解析
前言
歡迎來到大家深入理解 RxJava2 系列第二篇,這里先插上一句,本系列文章用的源碼都是基于 RxJava 2.2.0 正式版。本篇文章將先與大家一起理解 Scheduler 與 Worker ,順著 RxJava2 的源碼捋一下它們的實(shí)現(xiàn)原理。
Scheduler 與 Worker
Scheduler 與 Worker 在 RxJava2 中是一個(gè)非常重要的概念,他們是 RxJava 線程調(diào)度的核心與基石。用過的人肯定都會(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ì)上就是用來調(diào)度 Runnable 的,支持立即、延時(shí)和周期形式的調(diào)用,而 Worker 是任務(wù)的最小單元的載體。在 RxJava2 內(nèi)部的實(shí)現(xiàn)中,通常一個(gè)或者多個(gè) Worker 對應(yīng)一個(gè)ScheduledThreadPoolExecutor
對象,這些暫且不表。
scheduleDirect / schedulePeriodicallyDirect
在 RxJava 1.x 時(shí)代, Scheduler 是沒有scheduleDirect/schedulePeriodicallyDirect
的,只能夠先createWorker
,再通過 Worker 來調(diào)度任務(wù)。這些方法是對 Worker 調(diào)用的簡化,可以認(rèn)為是創(chuàng)建了一個(gè)只能調(diào)度一次任務(wù)的 Worker 并立馬調(diào)度了該任務(wù)。在Scheduler
基類的源碼中,也可以看出默認(rèn)的實(shí)現(xiàn)是直接 createWorker 并創(chuàng)建對應(yīng)的 Task 的(雖然在部分 Scheduler 覆蓋的實(shí)現(xiàn)上并沒有創(chuàng)建 Worker,但是可以認(rèn)為存在虛擬的 Worker)。
createWorker
一個(gè) Scheduler 可以創(chuàng)建多個(gè) Worker,這兩者是一對多的關(guān)系,而 Worker 與 Task 也是一對多的關(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í),可以通過 Worker 來將一系列的 Runnable 統(tǒng)一的調(diào)度和取消,最典型的例子就是observeOn
,下面會(huì)詳細(xì)分析。
Schedulers
RxJava2 默認(rèn)內(nèi)置了幾種 Scheduler 的實(shí)現(xiàn),適用于不同的場景,這些 Scheduler 均在 Schedulers 類中可以直接獲得
方法 | 說明 |
---|---|
Schedulers.computation() | 適用于計(jì)算密集型任務(wù) |
Schedulers.io() | 適用于 IO 密集型任務(wù) |
Schedulers.trampoline() | 在某個(gè)調(diào)用 schedule 的線程執(zhí)行 |
Schedulers.newThread() | 每個(gè) Worker 對應(yīng)一個(gè)新線程 |
Schedulers.single() | 所有 Worker 使用同一個(gè)線程執(zhí)行任務(wù) |
Schedulers.from(Executor) | 使用 Executor 作為任務(wù)執(zhí)行的線程 |
這里我們挑選兩個(gè)最常用的 computation / io 源碼稍作分析。
NewThreadWorker
NewThreadWorker 在 computation / io / newThread 均有涉及,我們先了解一下這個(gè)類。
上面筆者有提到過 Worker 與ScheduledThreadPoolExecutor
的關(guān)系,而這里的NewThreadWorker
與ScheduledThreadPoolExecutor
便是一對一的關(guān)系。在NewThreadWorker
構(gòu)造函數(shù)中會(huì)通過工廠方法創(chuàng)建一個(gè)corePoolSize 為 1 的ScheduledThreadPoolExecutor
對象并持有之。
ScheduledThreadPoolExecutor
從 JDK1.5 開始存在,這個(gè)類繼承于 ThreadPoolExecutor
,可以支持即使、延時(shí)和周期的任務(wù)。但是注意在ScheduledThreadPoolExecutor
中 maximumPoolSize 參數(shù)是無效的,corePoolSize 表示其最大線程數(shù),且它的隊(duì)列是無界的。這里不再細(xì)說該類,否則涉及的就太多了。
有了這個(gè)類,RxJava2 實(shí)現(xiàn) Worker 時(shí)便是站在了巨人的肩膀上,線程調(diào)度可以直接使用該類解決,略微麻煩之處就是封一層Disposable
的邏輯。
具體細(xì)節(jié)讀者可以從源碼一探究竟。
ComputationScheduler
作為計(jì)算密集型的 Scheduler,ComputationScheduler
的線程數(shù)是與 CPU 核心密切相關(guān)的,原因是當(dāng)線程數(shù)遠(yuǎn)遠(yuǎn)超過 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; }
從上面代碼可見MAX_THREADS
大于 0,但是不超過 CPU 核心數(shù),實(shí)際數(shù)值也受用戶設(shè)置的 System Properties 的影響。
FixedSchedulerPool
顧名思義,FixedSchedulerPool
可以認(rèn)為是固定數(shù)目的真正的 Worker 的緩存池。
確定了MAX_THREADS
后,在ComputationScheduler
的構(gòu)造函數(shù),會(huì)創(chuàng)建FixedSchedulerPool
對象,FixedSchedulerPool
內(nèi)部會(huì)直接創(chuàng)建一個(gè)長度為MAX_THREADS
的PoolWorker
數(shù)組。PoolWorker
繼承自NewThreadWorker
,但是沒有任何額外的代碼。
static final class PoolWorker extends NewThreadWorker { PoolWorker(ThreadFactory threadFactory) { super(threadFactory); } }
也就是說當(dāng)FixedSchedulerPool
創(chuàng)建時(shí),已經(jīng)有MAX_THREADS
個(gè) corePoolSize 為 1 的 ScheduledThreadPoolExecutor
隨之創(chuàng)建。
PoolWorker
從使用角度來說,有了FixedSchedulerPool
好像就夠了,我們只需要每次createWorker
時(shí)從池子里取一個(gè)PoolWorker
并返回即可。
但是這里忽略了一個(gè)要點(diǎn),每個(gè) Worker 是獨(dú)立的,每個(gè) Worker 內(nèi)部的任務(wù)是綁定在這個(gè) Worker 中的。如果按照上述的做法,暴露出去PoolWorker
,會(huì)出現(xiàn) 2 個(gè)問題:
- 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)致無法創(chuàng)建任務(wù)
EventLoopWorker
為了解決上述的問題,我們需要在PoolWorker
外再包一層,createWorker
每次都會(huì)創(chuàng)建一個(gè)EventLoopWorker
對象。
EventLoopWorker
其實(shí)是個(gè)代理對象,他會(huì)將 Runnable 代理給FixedSchedulerPool
中取到的PoolWorker
來調(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ù)是無上限的。這是因?yàn)?IO 設(shè)備的速度是遠(yuǎn)遠(yuǎn)低于 CPU 速度的,在等待 IO 操作時(shí), CPU 往往是閑置的,因此應(yīng)該創(chuàng)建更多的線程讓 CPU 盡可能的利用。當(dāng)然并不是說線程越多越好,線程數(shù)目膨脹到一定程度既會(huì)影響 CPU 的效率,也會(huì)消耗大量的內(nèi)存。在IoScheduler
中,每個(gè) Worker 在空置一段時(shí)間后就會(huì)被清除以控制線程的數(shù)目。
CachedWorkerPool
CachedWorkerPool
是一個(gè)變長并定期清理的ThreadWorker
的緩存池,內(nèi)部通過一個(gè)ConcurrentLinkedQueue
維護(hù)。和PoolWorker
類似,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
字段,用來標(biāo)識(shí)這個(gè)ThreadWorker
的超時(shí)時(shí)間。
于此同時(shí),在CachedWorkerPool
初始化時(shí)會(huì)傳入 Worker 的超時(shí)時(shí)間,目前是寫死的 60 秒。這個(gè)超時(shí)時(shí)間表示ThreadWorker
閑置后最大存活時(shí)間(實(shí)際中不保證 60 秒時(shí)被回收)。
EventLoopWorker
IoScheduler
中也存在一個(gè)EventLoopWorker
類,它和ComputationScheduler
中的作用也是類似的:
- 管理自身調(diào)度過的任務(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
這里說個(gè)細(xì)節(jié),因?yàn)?code>CachedWorkerPool是每隔 60 秒清理一次隊(duì)列的,因此ThreadWorker
的存活時(shí)間取決于入隊(duì)的時(shí)機(jī),如果一直沒有被再次取出,其被實(shí)際清理的延遲在 60 - 120 秒之間,有興趣的讀者可以想一想為什么。
示意圖
對比
熟悉線程的讀者朋友們會(huì)發(fā)現(xiàn),ComputationScheduler
與IoScheduler
很像某些參數(shù)下的ThreadPoolExecutor
。
ThreadPoolExecutor 參數(shù) | ComputationScheduler(n) | IoScheduler |
---|---|---|
corePoolSize | n | 0 |
maximumPoolSize | n | Integer.MAX_VALUE |
keepAliveTime | 0 | 60 |
unit | - | TimeUnit.SECONDS |
workQueue | LinkedBlockingQueue | SynchronousQueue |
他們對線程的控制外在的表現(xiàn)很相似。 但是實(shí)際的線程執(zhí)行對象不一樣:
- ThreadPoolExecutor:Thread
- Scheduler:支持立即、延遲、定時(shí)調(diào)度任務(wù)的對象,通常為 ScheduledThreadPoolExecutor(coreSize = 1)
這兩者的對比有助于我們更加深刻地理解 Scheduler 設(shè)計(jì)的內(nèi)在邏輯。
結(jié)語
Scheduler 是 RxJava 線程的核心概念,RxJava 基于此屏蔽了 Thread 相關(guān)的概念,只與 Scheduler / Worker / Runnable 打交道。
以上就是RxJava2 Scheduler使用實(shí)例深入解析的詳細(xì)內(nèi)容,更多關(guān)于RxJava2 Scheduler使用的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
linux環(huán)境下java程序打包成簡單的hello world輸出jar包示例
這篇文章主要介紹了linux環(huán)境下java程序打包成簡單的hello world輸出jar包,結(jié)合簡單hello world輸出程序示例分析了Linux環(huán)境下的java可執(zhí)行jar包文件的生成相關(guān)操作技巧,需要的朋友可以參考下2019-11-11詳解 Java中日期數(shù)據(jù)類型的處理之格式轉(zhuǎn)換的實(shí)例
這篇文章主要介紹了詳解 Java中日期數(shù)據(jù)類型的處理之格式轉(zhuǎn)換的實(shí)例的相關(guān)資料,日期以及時(shí)間格式處理,在Java中時(shí)間格式一般會(huì)涉及到的數(shù)據(jù)類型包括Calendar類和Date類,需要的朋友可以參考下2017-08-08Java中進(jìn)程、協(xié)程與線程的區(qū)別詳解
這篇文章主要介紹了Java中進(jìn)程,線程,協(xié)程的概念、區(qū)別以及使用場景的選擇,早期的操作系統(tǒng)每個(gè)程序就是一個(gè)進(jìn)程,知道一個(gè)程序運(yùn)行完,才能進(jìn)行下一個(gè)進(jìn)程,就是"單進(jìn)程時(shí)代",一切的程序只能串行發(fā)生,需要的朋友可以參考下2023-08-08淺談SpringCloud的微服務(wù)架構(gòu)組件
這篇文章主要介紹了淺談SpringCloud的微服務(wù)架構(gòu)組件,Spring Cloud根據(jù)分布式服務(wù)協(xié)調(diào)治理的需求成立了許多子項(xiàng)目,每個(gè)項(xiàng)目通過特定的組件去實(shí)現(xiàn),需要的朋友可以參考下2023-04-04