归档

• 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 实现，但底层原理一样
• 重复延时执行的任务，每次执行完，会重新添加到队列中