【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
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/509675.html
標籤：其他