原文作者：小付　

原文地址：ScheduledThreadPoolExecutor原理分析

目錄

一、簡單使用

二、類UML圖

三、處理流程

四、任務提交方式

五、SchduledFutureTask之run方法實現

---

一、簡單使用

這里先學會簡單使用再深入探討。

? ? ? ? ScheduledThreadPoolExecutor ?scheduled = new ScheduledThreadPoolExecutor(2);scheduled.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {loge("time:");}}, 0, 40, TimeUnit.MILLISECONDS);//0表示首次執行任務的延遲時間，40表示每次執行任務的間隔時間，TimeUnit.MILLISECONDS執行的時間間隔數值單位

?

• 間隔單位毫秒：TimeUnit.MILLISECONDS?
• 間隔單位秒：TimeUnit.SECONDS?
• 間隔單位分鐘：TimeUnit.MINUTES?
• 間隔單位小時：TimeUnit.HOURS?
• 間隔單位天：TimeUnit.DAYS

二、類UML圖

定時線程池類ScheduledThreadPoolExecutor用來處理延時任務或定時任務，它的類結構圖如下：

三、處理流程

它的處理流程如下：主線程提交的定時任務或延時任務采用DelayQueue存儲。

DelayQueue內部封裝了一個PriorityQueue（優先隊列，是一種堆結構），它會根據time的先后時間排序，若 time相同則根據sequenceNumber排序；DelayQueue也是一個無界隊列；

工作線程會從DelayQueue取已經到期的任務去執行；

執行結束后重新設置任務的到期時間，再次放回DelayQueue

PriorityQueue堆結構如下圖：

可見，DelayedQueue是一個基于最小堆結構的隊列。堆結構可以使用數 組表示，可以轉換成如下的數組：

為什么要使用DelayedWorkQueue呢？定時任務執行時需要取出最近要執行的任務，所以任務在隊列中每次出隊時一定要是當 前隊列中執行時間最靠前的，所以自然要使用優先級隊列。DelayedQueue是一個優先級隊列，它可以保證每次出隊的任務都是當前隊列中 執行時間最靠前的，由于它是基于堆結構的隊列，堆結構在執行插入和刪除操作時的最壞時 間復雜度是 O(logN)。PriorityQueue內部的比較邏輯實現代碼如下：

public int compareTo(Delayed other) {if (other == this) // compare zero if same objectreturn 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;elsereturn 1;}long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;}

四、任務提交方式

它接收SchduledFutureTask類型的任務，是線程池調度任務的最小單位，有三種提交任務的方式：

1、schedule()方法

首先是schedule方法，該方法是指任務在指定延遲時間到達后觸發，只會執行一次。

public ScheduledFuture<?> schedule(Runnable command,long delay,TimeUnit unit) {//參數校驗if (command == null || unit == null)throw new NullPointerException();//這里是一個嵌套結構，首先把用戶提交的任務包裝成ScheduledFutureTask 8 //然后在調用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);//如果當前狀態無法執行任務，則取消if (isShutdown() &&!canRunInCurrentRunState(task.isPeriodic()) &&remove(task))task.cancel(false);else//這里是增加一個worker線程，避免提交的任務沒有worker去執行//原因就是該類沒有像ThreadPoolExecutor一樣，woker滿了才放入隊列ensurePrestart();}}

2、scheduleAtFixedRate()

schedule和scheduleAtFixedRate的區別在于，如果指定開始執行的時間在當前系統運行時間之前，scheduleAtFixedRate會把已經過去的時間也作為周期執行，而schedule不會把過去的時間算上。scheduleAtFixedRate()源碼如下：

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit) {if (command == null || unit == null)throw new NullPointerException();if (period <= 0)throw new IllegalArgumentException();ScheduledFutureTask<Void> sft =new ScheduledFutureTask<Void>(command,null,triggerTime(initialDelay, unit),unit.toNanos(period));RunnableScheduledFuture<Void> t = decorateTask(command, sft);sft.outerTask = t;delayedExecute(t);return t;}

SchduledFutureTask：從上面提交任務方法的源碼的第六行ScheduledFutureTask<Void> sft =new ScheduledFutureTask<Void>(command,null,triggerTime(initialDelay, unit),?unit.toNanos(period));可以看到scheduleAtFixedRate()方法將command封裝為ScheduledFutureTask。下面再來看一下SchduledFutureTask接收的參數(成員變量)：

?- private long time：任務開始的時間?- private final long sequenceNumber?：任務的序號?- private final long period：任務執行的時間間隔

五、SchduledFutureTask之run方法實現

run方法是調度task的核心，task的執行實際上是run方法的執行。

?? ?public void run() {boolean periodic = isPeriodic();//如果當前線程池已經不支持執行任務，則取消if (!canRunInCurrentRunState(periodic))cancel(false);//如果不需要周期性執行，則直接執行run方法然后結束else if (!periodic)ScheduledFutureTask.super.run();//如果需要周期執行，則在執行完任務以后，設置下一次執行時間else if (ScheduledFutureTask.super.runAndReset()) {// 計算下次執行該任務的時間setNextRunTime();//重復執行任務reExecutePeriodic(outerTask);}}}

如果當前線程池運行狀態不可以執行任務，取消該任務，然后直接返回，否則執行 步驟2；

如果不是周期性任務，調用FutureTask中的run方法執行，會設置執行結果，然后 直接返回，否則執行步驟3；

如果是周期性任務，調用FutureTask中的runAndReset方法執行，不會設置執行 結果，然后直接返回，否則執行步驟4和步驟5；

計算下次執行該任務的具體時間；

重復執行任務。

reExecutePeriodic方法

void reExecutePeriodic(RunnableScheduledFuture<?> task) {if (canRunInCurrentRunState(true)) {super.getQueue().add(task);if (!canRunInCurrentRunState(true) && remove(task))task.cancel(false);elseensurePrestart();}}

該方法和delayedExecute方法類似，不同的是：

由于調用reExecutePeriodic方法時已經執行過一次周期性任務了，所以不會 reject當前任務；

傳入的任務一定是周期性任務。

一切偉大的行動和思想，都有一個微不足道的開始。

總結

以上是生活随笔為你收集整理的Java并发编程—ScheduledThreadPoolExecutor原理分析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。