【1】前言 本篇幅是對 線程池底層原理詳解與源碼分析 的補充，預設你已經看完了上一篇對ThreadPoolExecutor類有了足夠的瞭解。 【2】ScheduledThreadPoolExecutor的介紹 1.ScheduledThreadPoolExecutor繼承自ThreadPoolExe ...
【1】前言
本篇幅是對 線程池底層原理詳解與源碼分析 的補充，預設你已經看完了上一篇對ThreadPoolExecutor類有了足夠的瞭解。
【2】ScheduledThreadPoolExecutor的介紹
1.ScheduledThreadPoolExecutor繼承自ThreadPoolExecutor。它主要用來在給定的延遲之後運行任務，或者定期執行任務。ScheduledThreadPoolExecutor可以在構造函數中指定多個對應的後臺線程數。
2.構造函數展示
public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE,DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,new DelayedWorkQueue()); } public ScheduledThreadPoolExecutor(int corePoolSize,ThreadFactory threadFactory) { super(corePoolSize, Integer.MAX_VALUE,DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,new DelayedWorkQueue(), threadFactory); } public ScheduledThreadPoolExecutor(int corePoolSize,RejectedExecutionHandler handler) { super(corePoolSize, Integer.MAX_VALUE,DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,new DelayedWorkQueue(), handler); } public ScheduledThreadPoolExecutor(int corePoolSize,ThreadFactory threadFactory,RejectedExecutionHandler handler) { super(corePoolSize, Integer.MAX_VALUE,DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,new DelayedWorkQueue(), threadFactory, handler); }
3.通過構造函數我們可以看到，它的線程池本身就是調用ThreadPoolExecutor類的構造方法，因此也繼承了ThreadPoolExecutor類所存在的隱患：
允許的請求隊列長度為 Integer.MAX_VALUE，可能會堆積大量的請求，從而導致 OOM。
允許的創建線程數量為 Integer.MAX_VALUE，可能會創建大量的線程，從而導致 OOM。（且CPU會變成100%）
4.PS：既然隱患這麼嚴重，使用原生的不太合適。正所謂，人無橫財不富，馬無夜草不肥，打不過就加入。ScheduledThreadPoolExecutor繼承自ThreadPoolExecutor，那就寫個類繼承它然後調用ThreadPoolExecutor的構造方法區解決掉創建線程數被寫死為最大值的情況，然後瞭解一下DelayedWorkQueue（這個本質上也是優先順序隊列），繼承一下也改寫吧。畢竟自己的最合適不是嗎。【畢竟我覺得這些都是大佬們留給菜雞的底版，如拒絕策略不也是四個預設都沒人用嗎，都是要你根據自己的場景改】（畢竟我這猜測的原因是因為有了無盡隊列，其實線程數設置為Integer.MAX_VALUE已經沒有意義了）
【3】ScheduledThreadPoolExecutor的使用
1）schedule(Runnable command, long delay, TimeUnit unit)
方法說明：無返回值的延遲任務，有個嚴重的問題，就是沒有辦法獲知task的執行結果
2）schedule(Callable callable, long delay, TimeUnit unit)
方法說明：有返回值的延遲任務 :接收的是Callable實例，會返回一個ScheduleFuture對象，通過ScheduleFuture可以取消一個未執行的task，也可以獲得這個task的執行結果
3）scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)
方法說明： 固定頻率周期任務：第一次執行的延遲根據initialDelay參數確定，以後每一次執行都間隔period時長
4）scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)
方法說明： 固定延遲周期任務 ：scheduleWithFixedDelay的參數和scheduleAtFixedRate參數完全一致，它們的不同之處在於對period調度周期的解釋。在scheduleAtFixedRate中，period指的兩個任務開始執行的時間間隔，也就是當前任務的開始執行時間和下個任務的開始執行時間之間的間隔。而在scheduleWithFixedDelay中，period指的當前任務的結束執行時間到下個任務的開始執行時間。
【4】任務ScheduledFutureTask類源碼分析
1.構造方法展示
代碼展示
private class ScheduledFutureTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V> { ... ScheduledFutureTask(Runnable r, V result, long triggerTime, long sequenceNumber) { super(r, result); this.time = triggerTime; //表示這個任務將要被執行的具體時間 this.period = 0; //表示任務執行的間隔周期 this.sequenceNumber = sequenceNumber; //表示這個任務被添加到ScheduledThreadPoolExecutor中的序號（採用AtomicLong原子類累加當做序號） } ScheduledFutureTask(Runnable r, V result, long triggerTime, long period, long sequenceNumber) { super(r, result); this.time = triggerTime; this.period = period; this.sequenceNumber = sequenceNumber; } ScheduledFutureTask(Callable<V> callable, long triggerTime, long sequenceNumber) { super(callable); this.time = triggerTime; this.period = 0; this.sequenceNumber = sequenceNumber; } ... }
代碼說明
1.三個標註的參數是任務中主要的成員變數。
2.其次，我們會發現callable的任務是沒有間隔周期的：因為callable本身就是阻塞等待，而且周期性的也不合適。
3.實現了RunnableScheduledFuture介面，其主要方法isPeriodic()用於判斷是不是周期任務，又繼承了RunnableFuture介面.
4.ScheduledFutureTask又繼承了FutureTask類，而FutureTask類實現了RunnableFuture介面。（故感覺RunnableFuture介面的那些方法挺重要的）
5.RunnableFuture介面主要是由Runnable和Future兩大介面組成（自己去看繼承關係），主要有run()方法。
2.ScheduledFutureTask類#run方法
代碼展示
// 重寫FutureTask，如果是周期性任務需要重新放入隊列 public void run() { // 檢查當前狀態 不能執行任務，則取消任務 if (!canRunInCurrentRunState(this)) cancel(false); //如果不是周期任務，調用FutureTask.run()執行任務(非周期任務直接執行) else if (!isPeriodic()) super.run(); // 周期性任務 else if (super.runAndReset()) { //與run方法的不同就是正常完成後任務的狀態不會變化，依舊是NEW，且返回值為成功或失敗，不會設置result屬性 setNextRunTime(); //設置任務下次執行時間 reExecutePeriodic(outerTask); } }
代碼說明
1.這裡面很明顯存在一個隱患，那就是沒有捕捉異常，所以如果我們自定義的run()方法中如果沒有捕捉異常的話，那麼出現異常的時候我們容易兩眼摸瞎。
2.故使用定時任務的時候，自定義的run方法需要自行捕捉異常進行處理。
3.ScheduledFutureTask類#setNextRunTime方法
代碼展示
//判斷指定的任務是否為定期任務 private void setNextRunTime() { long p = period; //取出周期時間 if (p > 0) time += p; //time是周期任務的下一次執行時間 else time = triggerTime(-p); } // ScheduledThreadPoolExecutor中的方法 long triggerTime(long delay) { //delay 的值是否小於 Long.MAX_VALUE 的一半，是的話，當前時間+延遲時間 return System.nanoTime() + ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay)); } // ScheduledThreadPoolExecutor中的方法 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; }
代碼說明
1.周期時間period有正有負，這是ScheduledThreadPoolExecutor的ScheduledAtFixedRate和ScheduledWithFixedDelay的方法區別，前者為正數，後者為負數。
2.正數時，下一次執行時間為原來的執行時間+周期，即以執行開始時間為基準。
3.負數時，不考慮溢出情況，下一次執行時間為當前時間+周期，即以執行結束時間為基準。如果溢出，下一次執行時間為Long.MAX_VALUE + headDelay。
疑問說明（這一步有興趣的需要自己去調試然後在核心方法處斷點查看就可以了）
其實只要當做作System.nanoTime() + delay就可以了，沒必要關註overflowFree這一步，原因：
1.如果執行了 Long.MAX_VALUE + headDelay ，triggerTime方法會獲得負數，示例代碼
executor.scheduleAtFixedRate(task, 20, 1244574199069500L, TimeUnit.NANOSECONDS);//任延遲取最大值 穩定定時器 executor.scheduleWithFixedDelay(task, 1, 9223272036854775807L, TimeUnit.NANOSECONDS); //任務+延遲
2.如果不執行 Long.MAX_VALUE + headDelay ，triggerTime方法也有可能獲得負數，示例代碼：
executor.scheduleAtFixedRate(task, 20, 4611686018427387900L, TimeUnit.NANOSECONDS);
executor.scheduleWithFixedDelay(task, 1, 9223272036854775807L, TimeUnit.NANOSECONDS);
3.而且獲得負數在compareTo這一步不影響排序。【可能是由於科技發展的緣故吧，現在Long.MAX_VALUE【9223372036854775807L】溢出了，就會變為-9223372036854775808L，對排序不影響】
【5】ScheduledThreadPoolExecutor類源碼分析
1.ScheduledThreadPoolExecutor的四種使用方法
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) { if (command == null || unit == null) throw new NullPointerException(); RunnableScheduledFuture<Void> t = decorateTask(command, new ScheduledFutureTask<Void>(command, null, triggerTime(delay, unit), sequencer.getAndIncrement())); delayedExecute(t); return t; } public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) { if (callable == null || unit == null) throw new NullPointerException(); RunnableScheduledFuture<V> t = decorateTask(callable, new ScheduledFutureTask<V>(callable, triggerTime(delay, unit), sequencer.getAndIncrement())); delayedExecute(t); return t; } public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) { if (command == null || unit == null) throw new NullPointerException(); if (delay <= 0L) throw new IllegalArgumentException(); //這裡設置的-unit.toNanos(delay)是負數 ScheduledFutureTask<Void> sft = new ScheduledFutureTask<Void>(command, null, triggerTime(initialDelay, unit), -unit.toNanos(delay), sequencer.getAndIncrement()); //這個方法是用於以後做擴展的 RunnableScheduledFuture<Void> t = decorateTask(command, sft); sft.outerTask = t; delayedExecute(t); return t; } public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) { if (command == null || unit == null) throw new NullPointerException(); if (period <= 0L) throw new IllegalArgumentException(); //這裡設置unit.toNanos(period)是正數 ScheduledFutureTask<Void> sft = new ScheduledFutureTask<Void>(command, null, triggerTime(initialDelay, unit), unit.toNanos(period), sequencer.getAndIncrement()); //這個方法是用於以後做擴展的 RunnableScheduledFuture<Void> t = decorateTask(command, sft); sft.outerTask = t; delayedExecute(t); return t; }
2.ScheduledThreadPoolExecutor類#triggerTime方法
//獲取初始的延遲執行時間（以納秒的形式，相當於我在哪個時間點要執行） private long triggerTime(long delay, TimeUnit unit) { return triggerTime(unit.toNanos((delay < 0) ? 0 : delay)); } long triggerTime(long delay) { return System.nanoTime() + ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay)); }
3.ScheduledThreadPoolExecutor類#delayedExecute方法
private void delayedExecute(RunnableScheduledFuture<?> task) { //如果處於非運行狀態則拒絕任務（這個方法裡面比較的是不是比關閉狀態大） if (isShutdown()) reject(task); else { //加入隊列 super.getQueue().add(task); //如果加入隊列後canRunInCurrentRunState檢測線程池，返回false則移除任務 if (!canRunInCurrentRunState(task) && remove(task)) task.cancel(false); //以不可中斷方式執行完成執行中的調度任務 else ensurePrestart(); } } boolean canRunInCurrentRunState(RunnableScheduledFuture<?> task) { //如果處於運行狀態返回true if (!isShutdown()) return true; //處於停止狀態，整理狀態，銷毀狀態，三者之一返回false if (isStopped()) return false; //處於關閉狀態，返回run-after-shutdown參數 return task.isPeriodic() ? continueExistingPeriodicTasksAfterShutdown //預設false : (executeExistingDelayedTasksAfterShutdown || task.getDelay(NANOSECONDS) <= 0); } void ensurePrestart() { int wc = workerCountOf(ctl.get()); if (wc < corePoolSize) //保持工作者與核心線程數持平 addWorker(null, true); else if (wc == 0) //即時核心線程是0，也至少會啟動一個 addWorker(null, false); }
【6】DelayedWorkQueue類源碼分析
0.DelayedWorkQueue類#核心屬性
private static final int INITIAL_CAPACITY = 16; // 初始容量 private RunnableScheduledFuture<?>[] queue = new RunnableScheduledFuture<?>[INITIAL_CAPACITY]; // 控制併發和阻塞等待 private final ReentrantLock lock = new ReentrantLock(); private final Condition available = lock.newCondition(); //這個可以參考take方法與offer方法，個人覺得是採用中斷方式喚醒持有鎖的線程 private int size; // 節點數量 private Thread leader;//記錄持有鎖的線程（當等待的時候）
1.DelayedWorkQueue類#add方法
public boolean add(Runnable e) { return offer(e); } public boolean offer(Runnable x) { //空值校驗 if (x == null) throw new NullPointerException(); RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x; final ReentrantLock lock = this.lock; //加鎖 lock.lock(); try { int i = size; // 超過容量，擴容 if (i >= queue.length) grow(); size = i + 1; //更新當前節點數 if (i == 0) { //插入的是第一個節點（阻塞隊列原本為空） queue[0] = e; setIndex(e, 0); //setIndex(e, 0)用於修改ScheduledFutureTask的heapIndex屬性，表示該對象在隊列里的下標 } else {//阻塞隊列非空 siftUp(i, e); //在插入新節點後對堆進行調整，進行節點上移，保持其特性（節點的值小於子節點的值）不變 } /** * 這裡最好結合take方法理解一下 * 隊列頭等於當前任務，說明瞭當前任務的等待時間是最小的。此時為什麼要去清空leader？ * leader代表的是某一個正在等待獲取元素的線程句柄， * 在take的時候因為之前的頭結點時間未到，不能拿，被休眠了一定時間（而這個時間就是距離之前那個隊列頭結點的可以出隊列的時間差）。 * 此時頭結點換了，理應清空句柄，喚醒它，讓它再次嘗試去獲取最新的頭結點（就算是再次休眠，時間也會比之前的少）。 */ if (queue[0] == e) { leader = null; available.signal(); } } finally { lock.unlock(); //解鎖 } return true; }
2.DelayedWorkQueue類#siftUp方法
//其實把這個隊列看作樹結構會更容易理解（要理解數組與完全二叉樹的關聯） private void siftUp(int k, RunnableScheduledFuture<?> key) { while (k > 0) { int parent = (k - 1) >>> 1; //父節點坐標 RunnableScheduledFuture<?> e = queue[parent]; //獲取父節點的值 // 如果 節點>= 父節點，確定最終位置 if (key.compareTo(e) >= 0) break; // 節點<父節點，將節點向上移動（就是將父節點放在k處） queue[k] = e; setIndex(e, k); k = parent; } //確定key的最後落腳處 queue[k] = key; setIndex(key, k); }
3.ScheduledFutureTask類#compareTo方法
/** * compareTo 作用是加入元素到延遲隊列後，內部建立或者調整堆時候會使用該元素的 compareTo 方法與隊列裡面其他元素進行比較， * 讓最快要過期的元素放到隊首。所以無論什麼時候向隊列裡面添加元素，隊首的的元素都是最即將過期的元素。 * 如果時間相同，序列號小的排前面。 */ public int compareTo(Delayed other) { if (other == this) // 如果2個指向的同一個對象，則返回0 return 0; // other必須是ScheduledFutureTask類型的 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; } // 到這裡，說明other不是ScheduledFutureTask類型的 long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS); return (diff < 0) ? -1 : (diff > 0) ? 1 : 0; }
4.DelayedWorkQueue類#take方法
public RunnableScheduledFuture<?> take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); //加鎖，響應中斷 try { // 死迴圈自旋 for (;;) { RunnableScheduledFuture<?> first = queue[0]; //頭節點 // 隊列為null，則等待在條件上 if (first == null) available.await(); //隊列非空 else { //判斷延時時間是否滿足條件 long delay = first.getDelay(NANOSECONDS); if (delay <= 0L) return finishPoll(first); // 頭節點時間沒到，還不能取出頭節點 first = null; // 等待的時候，不要持有頭節點 if (leader != null) //已經存在leader線程，當前線程await阻塞 available.await(); else { //如果不存在leader線程，當前線程作為leader線程，並制定頭結點的延遲時間作為阻塞時間 Thread thisThread = Thread.currentThread(); leader = thisThread; try { available.awaitNanos(delay); } finally { //leader線程阻塞結束 if (leader == thisThread) leader = null; } } } } } finally { //leader線程沒有阻塞，可以找到頭結點，喚醒阻塞線程 if (leader == null && queue[0] != null) available.signal(); lock.unlock(); } }
5.DelayedWorkQueue類#grow方法
private void grow() { int oldCapacity = queue.length; int newCapacity = oldCapacity + (oldCapacity >> 1); //新容量為原來的1.5倍 if (newCapacity < 0) // overflow newCapacity = Integer.MAX_VALUE; queue = Arrays.copyOf(queue, newCapacity); //從舊數組 複製到 新數組 }
6.DelayedWorkQueue類#remove方法
public boolean remove(Object x) { final ReentrantLock lock = this.lock; lock.lock(); //加鎖 try { int i = indexOf(x); //定位x if (i < 0) //節點元素不存在 return false; setIndex(queue[i], -1); int s = --size; //末節點作為替代節點 RunnableScheduledFuture<?> replacement = queue[s]; queue[s] = null; //原本末節點處置空，便於GC if (s != i) { //下移，保證該節點的子孫節點保持特性 siftDown(i, replacement); // queue[i] == replacement說明下移沒有發生 if (queue[i] == replacement) //上移，保證該節點的祖先節點保持特性 siftUp(i, replacement); } return true; } finally { lock.unlock(); //加鎖 } }
7.DelayedWorkQueue類#siftDown方法
//情況說明：一般發生在隊列頭結點任務被取出了；這時候頭結點空閑，會把隊列【可看做是數組的情況會更好理解】末尾的元素【看作是樹的話，上層數據要比下層的要小】放入頭結點，然後向下轉移，達到保持優先隊列的情況。 private void siftDown(int k, RunnableScheduledFuture<?> key) { int half = size >>> 1; while (k < half) { int child = (k << 1) + 1; //左子節點坐標 RunnableScheduledFuture<?> c = queue[child]; //c表示左右子節點中的較小者，暫時是左 int right = child + 1; //右子節點坐標 //兩者進行比較，且下標沒有超出數據個數 if (right < size && c.compareTo(queue[right]) > 0) c = queue[child = right]; //右節點更小的話要變更數據和記錄下標 //直至找到下層沒有比自身小的元素時就停下 if (key.compareTo(c) <= 0) break; queue[k] = c; setIndex(c, k); k = child; } queue[k] = key; setIndex(key, k); }
8.DelayedWorkQueue類#finishPoll方法
// f是隊列頭節點（！！！） private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) { int s = --size; RunnableScheduledFuture<?> x = queue[s]; //取出隊列尾節點的值（之後放到合適位置） queue[s] = null; //置空，便於GC // 尾節點從0開始向下遍歷調整順序 if (s != 0) siftDown(0, x); setIndex(f, -1); //設置f的heapIndex屬性 return f; }
1244574199069500L
