歸檔

• GitHub: JDK-調度線程池

使用示例

• https://github.com/zengxf/small-frame-demo/blob/master/jdk-demo/simple-demo/src/main/java/test/jdkapi/juc/thread_pool/TestSchedule.java

JDK 版本

openjdk version "17" 2021-09-14
OpenJDK Runtime Environment (build 17+35-2724)
OpenJDK 64-Bit Server VM (build 17+35-2724, mixed mode, sharing)

原理

類結構

• java.util.concurrent.ScheduledThreadPoolExecutor

/*** 調度線程池 */
public class ScheduledThreadPoolExecutorextends ThreadPoolExecutorimplements ScheduledExecutorService 
{/*** 構造器 */public ScheduledThreadPoolExecutor(int corePoolSize) {super(corePoolSize, Integer.MAX_VALUE,DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,new DelayedWorkQueue()); // 使用內部隊列作任務隊列 sign_c_001}}

• java.util.concurrent.ScheduledThreadPoolExecutor.DelayedWorkQueue

    /*** sign_c_001 延遲任務隊列 */static class DelayedWorkQueue extends AbstractQueue<Runnable>implements BlockingQueue<Runnable> {private static final int INITIAL_CAPACITY = 16; // 初始隊列大小 16 個private RunnableScheduledFuture<?>[] queue =new RunnableScheduledFuture<?>[INITIAL_CAPACITY];private final ReentrantLock lock = new ReentrantLock();private final Condition available = lock.newCondition();}

• java.util.concurrent.ScheduledThreadPoolExecutor.ScheduledFutureTask

    /*** sign_c_010 調度任務 */private class ScheduledFutureTask<V>extends FutureTask<V> implements RunnableScheduledFuture<V> // 繼承 sign_c_011 實現 sign_i_011{private final long sequenceNumber;  // 序列號(添加時遞增設置)private volatile long time;         // 觸發時間(執行的時間點)，基于納米/*** 重復任務的周期，以納秒為單位。* 正值表示固定速率執行。* 負值表示固定延遲執行。* 值為 0 表示非重復（單次）任務。*/private final long period;// 構造器ScheduledFutureTask(Runnable r, V result, long triggerTime,long period, long sequenceNumber) {super(r, result); // 執行體交給父類(sign_c_011)保存this.time = triggerTime;this.period = period;this.sequenceNumber = sequenceNumber;}}

• java.util.concurrent.FutureTask

/*** sign_c_011 */
public class FutureTask<V> implements RunnableFuture<V> {private volatile int state;     // 狀態private Callable<V> callable;   // 封裝的任務public FutureTask(Runnable runnable, V result) {this.callable = Executors.callable(runnable, result);this.state = NEW;       // ensure visibility of callable}
}

• Future 相關接口定義

// sign_i_011
// java.util.concurrent.RunnableScheduledFuture
public interface RunnableScheduledFuture<V> extends RunnableFuture<V>, ScheduledFuture<V> {/*** 判斷是不是周期任務*/boolean isPeriodic();
}// java.util.concurrent.ScheduledFuture
public interface ScheduledFuture<V> extends Delayed, Future<V> {
}// java.util.concurrent.Delayed
public interface Delayed extends Comparable<Delayed> {/*** 獲取延時時間，* Comparable 用于排序，延時小的排前面。*/long getDelay(TimeUnit unit);
}

• java.util.concurrent.ThreadPoolExecutor.Worker

    /*** sign_c_020 工作者(線程) */private final class Workerextends AbstractQueuedSynchronizerimplements Runnable{final Thread thread;Runnable firstTask;volatile long completedTasks;Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;this.thread = getThreadFactory().newThread(this); // 創建一個線程，自己為執行體 sign_m_030}}

調用鏈

• java.util.concurrent.ScheduledThreadPoolExecutor

    /*** 創建調度任務 */public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit) {... // 省略校驗ScheduledFutureTask<Void> sft = // 創建調度任務 sign_c_010 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;}// 延遲執行private void delayedExecute(RunnableScheduledFuture<?> task) {if (isShutdown())reject(task);else {super.getQueue().add(task); // 添加到隊列if (!canRunInCurrentRunState(task) && remove(task))task.cancel(false);elseensurePrestart();       // 啟動任務執行者線程 sign_sm_001}}

• java.util.concurrent.ThreadPoolExecutor

    // sign_sm_001 啟動任務執行者線程 void ensurePrestart() {int wc = workerCountOf(ctl.get());if (wc < corePoolSize)addWorker(null, true);  // 添加工作線程else if (wc == 0)addWorker(null, false);}// 添加工作線程private boolean addWorker(Runnable firstTask, boolean core) {... // 省略其他處理boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {w = new Worker(firstTask);  // 創建工作者 sign_c_020final Thread t = w.thread;if (t != null) {......workers.add(w); // 添加到工作者集合...t.start();          // 啟動線程...}} ... // finallyreturn workerStarted;}// sign_m_022 執行隊列中的任務 final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;... // 省略其他try {while (task != null || (task = getTask()) != null) { // 獲取隊列中的任務w.lock();... // 線程中斷處理try {beforeExecute(wt, task);        // 調用執行前的鉤子函數try {task.run();                 // 執行任務體(相當于定時任務的執行體 sign_m_040)afterExecute(task, null);   // 調用執行后的鉤子函數} ... // catch} finally {task = null;w.completedTasks++; // 完成的任務數 +1w.unlock();}}...} ... // finally}private Runnable getTask() {...for (;;) {... // 省略其他判斷處理try {Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();   // 默認調用 take(): 獲取任務或限時等待 sign_m_050if (r != null)return r;...} ... // catch }}

• java.util.concurrent.ThreadPoolExecutor.Worker

        // sign_m_030 執行體 public void run() {runWorker(this);    // 執行隊列中的任務 sign_m_022}

• java.util.concurrent.ScheduledThreadPoolExecutor.ScheduledFutureTask

        /*** sign_m_040 定時任務執行體 */public void run() {if (!canRunInCurrentRunState(this))cancel(false);else if (!isPeriodic())super.run();else if (super.runAndReset()) {     // 執行任務具體邏輯setNextRunTime();               // 設置任務下次被執行的時間reExecutePeriodic(outerTask);   // 重新添加到(當前線程池的)隊列}}

• java.util.concurrent.ScheduledThreadPoolExecutor.DelayedWorkQueue

        /*** sign_m_050 從隊列中獲取任務(或限時等待) */public RunnableScheduledFuture<?> take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {for (;;) {RunnableScheduledFuture<?> first = queue[0];if (first == null)available.await();  // 沒有任務則無限等待else {long delay = first.getDelay(NANOSECONDS);   // 需要延時的時間，實例 ref: sign_c_010if (delay <= 0L)return finishPoll(first);   // 小于 0 表示不需要延時，直接返回first = null; // don't retain ref while waitingif (leader != null)available.await();  // 相當于有線程在等待延時，就無限等待else {Thread thisThread = Thread.currentThread();leader = thisThread;    // 相當于設置標識(給其他線程判斷是不是要無限等待)try {available.awaitNanos(delay);    // 等待需要延時的時間} finally {if (leader == thisThread)leader = null;  // 相當于清空標識}}}}} finally {if (leader == null && queue[0] != null)available.signal();lock.unlock();}}

總結

• 線程等待是在隊列 take() 方法中處理
  • 等待時延是由 ScheduledFutureTask #getDelay() 進行判斷
  • 而不是通過 DelayQueue 實現，但底層原理一樣
• 重復延時執行的任務，每次執行完，會重新添加到隊列中