點贊再看,養成習慣,公衆號搜一搜【一角錢技術】關注更多原創技術文章。本文 GitHub org_hejianhui/JavaStudy 已收錄,有個人系列文章。html
線程池的具體實現有兩種,分別是ThreadPoolExecutor 默認線程池和ScheduledThreadPoolExecutor 定時線程池,上一篇已經分析過ThreadPoolExecutor原理與使用了,本篇咱們來重點分析下ScheduledThreadPoolExecutor的原理與使用。java
ScheduledThreadPoolExecutor
與 ThreadPoolExecutor 線程池的概念有些區別,它是一個支持任務週期性調度的線程池。git
ScheduledThreadPoolExecutor 繼承 ThreadPoolExecutor,同時經過實現 ScheduledExecutorSerivce 來擴展基礎線程池的功能,使其擁有了調度能力。其整個調度的核心在於內部類 DelayedWorkQueue ,一個有序的延時隊列。github
定時線程池類的類結構圖以下: 編程
ScheduledThreadPoolExecutor 的出現,很好的彌補了傳統 Timer 的不足,具體對比看下錶:數組
Timer | ScheduledThreadPoolExecutor | |
---|---|---|
線程 | 單線程 | 多線程 |
多任務 | 任務之間相互影響 | 任務之間不影響 |
調度時間 | 絕對時間 | 相對時間 |
異常 | 單任務異常,後續任務受影響 | 無影響 |
它用來處理延時任務或定時任務 它接收SchduledFutureTask類型的任務,是線程池調度任務的最小單位,有三種提交任務的方式:markdown
它採用 DelayedWorkQueue
存儲等待的任務數據結構
PriorityQueue
,它會根據 time 的前後時間排序,若 time 相同則根據 sequenceNumber
排序;由於前面講阻塞隊列實現的時候,已經對DelayedWorkQueue進行了說明,更多內容請查看《阻塞隊列 — DelayedWorkQueue源碼分析》多線程
工做線程的執行過程:併發
take方法是何時調用的呢? 在ThreadPoolExecutor中,getTask方法,工做線程會循環地從workQueue中取任務。但定時任務卻不一樣,由於若是一旦getTask方法取出了任務就開始執行了,而這時可能尚未到執行的時間,因此在take方法中,要保證只有在到指定的執行時間的時候任務才能夠被取走。
PS:對於以上原理的理解,能夠經過下面的源碼分析加深印象。
ScheduledThreadPoolExecutor有四個構造形式:
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
public ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), threadFactory);
}
public ScheduledThreadPoolExecutor(int corePoolSize, RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), handler);
}
public ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), threadFactory, handler);
}
複製代碼
固然咱們也可使用工具類Executors的newScheduledThreadPool的方法,快速建立。注意這裏使用的DelayedWorkQueue。
ScheduledThreadPoolExecutor沒有提供帶有最大線程數的構造函數的,默認是Integer.MAX_VALUE,說明其能夠無限制的開啓任意線程執行任務,在大量任務系統,應注意這一點,避免內存溢出。
核心方法主要介紹ScheduledThreadPoolExecutor的調度方法,其餘方法與 ThreadPoolExecutor 一致。調度方法均由 ScheduledExecutorService 接口定義:
public interface ScheduledExecutorService extends ExecutorService {
// 特定時間延時後執行一次Runnable
public ScheduledFuture<?> schedule(Runnable command,
long delay, TimeUnit unit);
// 特定時間延時後執行一次Callable
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);
// 固定週期執行任務(與任務執行時間無關,週期是固定的)
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit);
// 固定延時執行任務(與任務執行時間有關,延時從上一次任務完成後開始)
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit);
}
複製代碼
咱們再來看一下接口的實現,具體是怎麼來實現線程池任務的提交。由於最終都回調用 delayedExecute 提交任務。因此,咱們這裏只分析schedule方法,該方法是指任務在指定延遲時間到達後觸發,只會執行一次。源代碼以下:
public ScheduledFuture<?> schedule(Runnable command,
long delay,
TimeUnit unit) {
//參數校驗
if (command == null || unit == null)
throw new NullPointerException();
//這裏是一個嵌套結構,首先把用戶提交的任務包裝成ScheduledFutureTask
//而後在調用decorateTask進行包裝,該方法是留給用戶去擴展的,默認是個空方法
RunnableScheduledFuture<?> t = decorateTask(command,
new ScheduledFutureTask<Void>(command, null,
triggerTime(delay, unit)));
//包裝好任務之後,就進行提交了
delayedExecute(t);
return t;
}
複製代碼
delayedExecute 任務提交方法:
private void delayedExecute(RunnableScheduledFuture<?> task) {
//若是線程池已經關閉,則使用拒絕策略把提交任務拒絕掉
if (isShutdown())
reject(task);
else {
//與ThreadPoolExecutor不一樣,這裏直接把任務加入延遲隊列
super.getQueue().add(task);//使用用的DelayedWorkQueue
//若是當前狀態沒法執行任務,則取消
if (isShutdown() &&
!canRunInCurrentRunState(task.isPeriodic()) &&
remove(task))
task.cancel(false);
else
//這裏是增長一個worker線程,避免提交的任務沒有worker去執行
//緣由就是該類沒有像ThreadPoolExecutor同樣,woker滿了才放入隊列
ensurePrestart();
}
}
複製代碼
咱們能夠看到提交到線程池的任務都包裝成了 ScheduledFutureTask,繼續往下咱們再來研究下。 造方
從ScheduledFutureTask類的定義能夠看出,ScheduledFutureTask類是ScheduledThreadPoolExecutor類的私有內部類,繼承了FutureTask類,並實現了RunnableScheduledFuture接口。也就是說,ScheduledFutureTask具備FutureTask類的全部功能,並實現了RunnableScheduledFuture接口的全部方法。ScheduledFutureTask類的定義以下所示:
private class ScheduledFutureTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V> 複製代碼
ScheduledFutureTask類繼承圖以下:
SchduledFutureTask接收的參數(成員變量):
// 任務開始的時間
private long time;
// 任務添加到ScheduledThreadPoolExecutor中被分配的惟一序列號
private final long sequenceNumber;
// 任務執行的時間間隔
private final long period;
//ScheduledFutureTask對象,實際指向當前對象自己
RunnableScheduledFuture<V> outerTask = this;
//當前任務在延遲隊列中的索引,可以更加方便的取消當前任務
int heapIndex;
複製代碼
解析:
ScheduledFutureTask類繼承了FutureTask類,並實現了RunnableScheduledFuture接口。在ScheduledFutureTask類中提供了以下構造方法。
ScheduledFutureTask(Runnable r, V result, long ns) {
super(r, result);
this.time = ns;
this.period = 0;
this.sequenceNumber = sequencer.getAndIncrement();
}
ScheduledFutureTask(Runnable r, V result, long ns, long period) {
super(r, result);
this.time = ns;
this.period = period;
this.sequenceNumber = sequencer.getAndIncrement();
}
ScheduledFutureTask(Callable<V> callable, long ns) {
super(callable);
this.time = ns;
this.period = 0;
this.sequenceNumber = sequencer.getAndIncrement();
}
複製代碼
FutureTask的構造方法以下:
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
複製代碼
經過源碼能夠看到,在ScheduledFutureTask類的構造方法中,首先會調用FutureTask類的構造方法爲FutureTask類的callable和state成員變量賦值,接下來爲ScheduledFutureTask類的time、period和sequenceNumber成員變量賦值。理解起來比較簡單。
咱們先來看getDelay方法的源碼,以下所示:
//獲取下次執行任務的時間距離當前時間的納秒數
public long getDelay(TimeUnit unit) {
return unit.convert(time - now(), NANOSECONDS);
}
複製代碼
getDelay方法比較簡單,主要用來獲取下次執行任務的時間距離當前系統時間的納秒數。
ScheduledFutureTask類在類的結構上實現了Comparable接口,compareTo方法主要是對Comparable接口定義的compareTo方法的實現。源碼以下所示:
public int compareTo(Delayed other) {
if (other == this)
return 0;
if (other instanceof ScheduledFutureTask) {
ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
long diff = time - x.time;
if (diff < 0)
return -1;
else if (diff > 0)
return 1;
else if (sequenceNumber < x.sequenceNumber)
return -1;
else
return 1;
}
long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
}
複製代碼
這段代碼看上去好像是對各類數值類型數據的比較,本質上是對延遲隊列中的任務進行排序。排序規則爲:
isPeriodic方法的源代碼以下所示:
//判斷是不是週期性任務
public boolean isPeriodic() {
return period != 0;
}
複製代碼
這個方法主要是用來判斷當前任務是不是週期性任務。這裏只要判斷運行任務的執行週期不等於0就能肯定爲週期性任務了。由於不管period的值是大於0仍是小於0,當前任務都是週期性任務。
setNextRunTime方法的做用主要是設置當前任務下次執行的時間,源碼以下所示:
private void setNextRunTime() {
long p = period;
//固定頻率,上次執行任務的時間加上任務的執行週期
if (p > 0)
time += p;
//相對固定的延遲執行,當前系統時間加上任務的執行週期
else
time = triggerTime(-p);
}
複製代碼
這裏再一次證實了使用isPeriodic方法判斷當前任務是否爲週期性任務時,只要判斷period的值是否不等於0就能夠了。
這裏咱們看到在setNextRunTime方法中,調用了ScheduledThreadPoolExecutor類的triggerTime方法。接下來咱們看下triggerTime方法的源碼。
triggerTime方法用於獲取延遲隊列中的任務下一次執行的具體時間。源碼以下所示。
private long triggerTime(long delay, TimeUnit unit) {
return triggerTime(unit.toNanos((delay < 0) ? 0 : delay));
}
long triggerTime(long delay) {
return now() +
((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
}
複製代碼
這兩個triggerTime方法的代碼比較簡單,就是獲取下一次執行任務的具體時間。有一點須要注意的是:delay < (Long.MAX_VALUE >> 1判斷delay的值是否小於Long.MAX_VALUE的一半,若是小於Long.MAX_VALUE值的一半,則直接返回delay,不然須要處理溢出的狀況。
咱們看到在triggerTime方法中處理防止溢出的邏輯使用了ScheduledThreadPoolExecutor類的overflowFree方法,接下來,咱們就看看ScheduledThreadPoolExecutor類的overflowFree方法的實現。
overflowFree方法的源代碼以下所示:
private long overflowFree(long delay) {
//獲取隊列中的節點
Delayed head = (Delayed) super.getQueue().peek();
//獲取的節點不爲空,則進行後續處理
if (head != null) {
//從隊列節點中獲取延遲時間
long headDelay = head.getDelay(NANOSECONDS);
//若是從隊列中獲取的延遲時間小於0,而且傳遞的delay
//值減去從隊列節點中獲取延遲時間小於0
if (headDelay < 0 && (delay - headDelay < 0))
//將delay的值設置爲Long.MAX_VALUE + headDelay
delay = Long.MAX_VALUE + headDelay;
}
//返回延遲時間
return delay;
}
複製代碼
經過對overflowFree方法的源碼分析,能夠看出overflowFree方法本質上就是爲了限制隊列中的全部節點的延遲時間在Long.MAX_VALUE值以內,防止在compareTo方法中溢出。
cancel方法的做用主要是取消當前任務的執行,源碼以下所示:
public boolean cancel(boolean mayInterruptIfRunning) {
//取消任務,返回任務是否取消的標識
boolean cancelled = super.cancel(mayInterruptIfRunning);
//若是任務已經取消
//而且須要將任務從延遲隊列中刪除
//而且任務在延遲隊列中的索引大於或者等於0
if (cancelled && removeOnCancel && heapIndex >= 0)
//將當前任務從延遲隊列中刪除
remove(this);
//返回是否成功取消任務的標識
return cancelled;
}
複製代碼
這段代碼理解起來相對比較簡單,首先調用取消任務的方法,並返回任務是否已經取消的標識。若是任務已經取消,而且須要移除任務,同時,任務在延遲隊列中的索引大於或者等於0,則將當前任務從延遲隊列中移除。最後返回任務是否成功取消的標識。
run方法能夠說是ScheduledFutureTask類的核心方法,是對Runnable接口的實現,源碼以下所示:
public void run() {
//當前任務是不是週期性任務
boolean periodic = isPeriodic();
//線程池當前運行狀態下不能執行週期性任務
if (!canRunInCurrentRunState(periodic))
//取消任務的執行
cancel(false);
//若是不是週期性任務
else if (!periodic)
//則直接調用FutureTask類的run方法執行任務
ScheduledFutureTask.super.run();
//若是是週期性任務,則調用FutureTask類的runAndReset方法執行任務
//若是任務執行成功
else if (ScheduledFutureTask.super.runAndReset()) {
//設置下次執行任務的時間
setNextRunTime();
//重複執行任務
reExecutePeriodic(outerTask);
}
}
複製代碼
整理一下方法的邏輯:
這裏,調用了FutureTask類的run方法和runAndReset方法,而且調用了ScheduledThreadPoolExecutor類的reExecutePeriodic方法。接下來,咱們分別看下這些方法的實現。
FutureTask類的run方法源碼以下所示:
public void run() {
//狀態若是不是NEW,說明任務或者已經執行過,或者已經被取消,直接返回
//狀態若是是NEW,則嘗試把當前執行線程保存在runner字段中
//若是賦值失敗則直接返回
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
//執行任務
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
//任務異常
setException(ex);
}
if (ran)
//任務正常執行完畢
set(result);
}
} finally {
runner = null;
int s = state;
//若是任務被中斷,執行中斷處理
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
複製代碼
代碼的總體邏輯爲:
方法的源碼以下所示:
protected boolean runAndReset() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return false;
boolean ran = false;
int s = state;
try {
Callable<V> c = callable;
if (c != null && s == NEW) {
try {
c.call(); // don't set result
ran = true;
} catch (Throwable ex) {
setException(ex);
}
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
return ran && s == NEW;
}
複製代碼
FutureTask類的runAndReset方法與run方法的邏輯基本相同,只是runAndReset方法會重置當前任務的執行狀態。
reExecutePeriodic重複執行任務方法,源代碼以下所示:
void reExecutePeriodic(RunnableScheduledFuture<?> task) {
//線程池當前狀態下可以執行任務
if (canRunInCurrentRunState(true)) {
//與ThreadPoolExecutor不一樣,這裏直接把任務加入延遲隊列
super.getQueue().add(task);//使用用的DelayedWorkQueue
//線程池當前狀態下不能執行任務,而且成功移除任務
if (!canRunInCurrentRunState(true) && remove(task))
//取消任務
task.cancel(false);
else
//這裏是增長一個worker線程,避免提交的任務沒有worker去執行
//緣由就是該類沒有像ThreadPoolExecutor同樣,woker滿了才放入隊列
ensurePrestart();
}
}
複製代碼
整體來講reExecutePeriodic方法的邏輯比較簡單,須要注意的是:調用reExecutePeriodic方法的時候已經執行過一次任務,因此,並不會觸發線程池的拒絕策略;傳入reExecutePeriodic方法的任務必定是週期性的任務。
ScheduledThreadPoolExecutor之因此要本身實現阻塞的工做隊列,是由於 ScheduleThreadPoolExecutor 要求的工做隊列有些特殊。
DelayedWorkQueue是一個基於堆的數據結構,相似於DelayQueue和PriorityQueue。在執行定時任務的時候,每一個任務的執行時間都不一樣,因此DelayedWorkQueue的工做就是按照執行時間的升序來排列,執行時間距離當前時間越近的任務在隊列的前面(注意:這裏的順序並非絕對的,堆中的排序只保證了子節點的下次執行時間要比父節點的下次執行時間要大,而葉子節點之間並不必定是順序的)。
堆結構以下圖: 可見,DelayedWorkQueue是一個基於最小堆結構的隊列。堆結構可使用數組表示,能夠轉換成以下的數組: 在這種結構中,能夠發現有以下特性: 假設「第一個元素」 在數組中的索引爲 0 的話,則父結點和子結點的位置關係以下:
爲何要使用DelayedWorkQueue呢?
由於前面講阻塞隊列實現的時候,已經對DelayedWorkQueue進行了說明,更多內容請查看《阻塞隊列 — DelayedWorkQueue源碼分析》
DelayedWorkQueue
;schedule
、 scheduleAtFixedRate
、 scheduleWithFixedDelay
,同時注意他們之間的區別;ScheduledFutureTask
繼承者FutureTask,實現了任務的異步執行而且能夠獲取返回結果。同時實現了Delayed接口,能夠經過getDelay方法獲取將要執行的時間間隔;runAndReset
方法,每次執行完不設置結果和狀態。整體來講,ScheduedThreadPoolExecutor的重點是要理解下次執行時間的計算,以及優先隊列的出隊、入隊和刪除的過程,這兩個是理解ScheduedThreadPoolExecutor的關鍵。
PS:以上代碼提交在 Github :github.com/Niuh-Study/…
PS:這裏有一個技術交流羣(扣扣羣:1158819530),方便你們一塊兒交流,持續學習,共同進步,有須要的能夠加一下。
文章持續更新,能夠公衆號搜一搜「 一角錢技術 」第一時間閱讀, 本文 GitHub org_hejianhui/JavaStudy 已經收錄,歡迎 Star。