计划任务线程池ScheduledThreadPoolExecutor原理

Preview

部分内容来源：《深入浅出Java多线程》 - 计划任务

JDK版本：OpenJDK16.0.2

使用样例

将消息（包含发送时间）存储在数据库中，用一个定时任务，每隔1秒检查数据库在当前时间有没有需要发送的消息：

private static final ScheduledExecutorService executor = 
    new ScheduledThreadPoolExecutor(1, Executors.defaultThreadFactory());

private static SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

public static void main(String[] args){
    // 新建一个固定延迟时间的计划任务
    // 新建任务1s以后，任务开始执行
    // 上一个任务执行完以后，等待2s，执行下一个任务
    System.err.printf("【%s】新建任务%n" , df.format(new Date()));
    executor.scheduleWithFixedDelay(new Runnable() {
        @Override
        public void run() {
            if (haveMsgAtCurrentTime()) {
                System.err.printf("【%s】大家注意了，我要发消息了%n" , df.format(new Date()));
            }
        }
    }, 1, 2, TimeUnit.SECONDS);
}

public static boolean haveMsgAtCurrentTime(){
    // 查询数据库，有没有当前时间需要发送的消息
    // 这里省略实现，直接返回true
    return true;
}

输出：

【2021-10-12 20:27:35】新建任务
【2021-10-12 20:27:36】大家注意了，我要发消息了
【2021-10-12 20:27:38】大家注意了，我要发消息了
【2021-10-12 20:27:40】大家注意了，我要发消息了

计划任务的特性

计划任务分为两种：

• 非周期性任务，这种任务只执行一次，需要在指定的时间运行
• 周期性任务，这种任务要执行多次，周期性任务又可以分为两种
  • 固定频率：每隔一段时间，任务就执行一次，比如每五分钟执行一次
  • 固定间隔：两次任务的执行之间需要间隔一定的时间，比如本次任务执行后，等待五分钟，然后执行下一次任务

假如让我们自己来实现一个计划任务线程池，我们需要实现两个特性：

1. 多次执行任务
2. 在指定时间执行任务

如果只执行非周期性任务，只需要满足第二点特性就可以，但对于周期性任务，必须两个特性都要满足，可以说，只要线程池可以实现这两个特性，这个线程池就是计划任务线程池

所以，ScheduledThreadPoolExecutor的关键就在于，它是如何实现这两个特性的

下面，带着这两个疑问，我们来分析ScheduledThreadPoolExecutor的源码

在分析过程中，我们顺着线程池的使用方式来阅读源码，首先看一下线程池在提交任务时会做些什么，然后再看看任务在执行时又会做些什么

ScheduledThreadPoolExecutor

类结构

public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {
    
    // 计划任务线程池的构造方法之一
    // 注意，这里使用的workQueue是DelayedWorkQueue，关于这个队列的具体内容，我们后面再聊
    public ScheduledThreadPoolExecutor(int corePoolSize,ThreadFactory threadFactory) {
        super(corePoolSize, Integer.MAX_VALUE,
              DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
              new DelayedWorkQueue(), threadFactory);
    }
}

ScheduledThreadPoolExecutor继承了ThreadPoolExecutor，这个类就是线程池，不多赘述

ScheduledThreadPoolExecutor还实现了ScheduledExecutorService接口，这个接口规定了一些方法签名，这些方法负责把周期性任务提交到线程池，源码如下

public interface ScheduledExecutorService extends ExecutorService {
    
    // 单次执行任务，无返回值
    public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
    
    // 单次执行任务，有返回值
    public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);

    // 多次执行任务，创建任务后，经过 initialDelay 时间，执行第一次任务
    // 此后，每隔 period 时间，执行一次任务，无论上一次任务是否完成，都会执行
    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit);
    
    // 多次执行任务，创建任务后，经过 initialDelay 时间，执行第一次任务
    // 每次任务执行完成之后，间隔 delay 时间，才执行下一次任务
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit);
}

提交任务的四个方法

ScheduledExecutorService中制定了四个提交周期性任务，在ScheduledThreadPoolExecutor中的实现如下：

schedule(无返回值)

public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {

    // 用于打破调度关系的序列号，保证绑定项之间的FIFO顺序
    private static final AtomicLong sequencer = new AtomicLong();
    
    // 单次执行任务，无返回值
    public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        // decorateTask：直接返回第二个参数
        // 在这里，会直接返回 new 出来的 ScheduledFutureTask 对象
        RunnableScheduledFuture<Void> t = decorateTask(command,
            // 创建任务，带有初始延时
            new ScheduledFutureTask<Void>(command, null,
                                          // triggerTime：根据delay、unit和当前系统时间，计算出第一次执行任务的时间
                                          triggerTime(delay, unit),
                                          // 序列号+1
                                          sequencer.getAndIncrement()));
        // 延期或周期性任务的主要方法
        delayedExecute(t);
        return t; 
    }
    
    // 直接返回第二个参数
    protected <V> RunnableScheduledFuture<V> decorateTask(
        Runnable runnable, RunnableScheduledFuture<V> task) {
        return task;
    }
}

schedule(有返回值)

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;
}

scheduledAtFixedRate

// 多次执行任务，创建任务后，经过 initialDelay 时间，执行第一次任务
// 此后，每隔 period 时间，执行一次任务，无论上一次任务是否完成，都会执行
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();
    // 创建任务
    ScheduledFutureTask<Void> sft =
        new ScheduledFutureTask<Void>(command,
                                      null,
                                      triggerTime(initialDelay, unit),
                                      unit.toNanos(period),
                                      sequencer.getAndIncrement());
    // decorateTask直接返回第二个参数，也就是创建的任务对象
    RunnableScheduledFuture<Void> t = decorateTask(command, sft);
    sft.outerTask = t;
    // 延迟或周期性任务的主要执行方法，拒绝任务或者把任务放入workQueue中
    delayedExecute(t);
    return t;
}

scheduledAtFixedDelay

// 多次执行任务，创建任务后，经过 initialDelay 时间，执行第一次任务
// 每次任务执行完成之后，间隔 delay 时间，才执行下一次任务
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();
    // 创建任务，带有初始延时和固定间隔（一个负数）
    ScheduledFutureTask<Void> sft =
        new ScheduledFutureTask<Void>(command,
                                      null,
                                      triggerTime(initialDelay, unit),
                                      -unit.toNanos(delay),
                                      sequencer.getAndIncrement());
    // decorateTask直接返回第二个参数，也就是创建的任务对象
    RunnableScheduledFuture<Void> t = decorateTask(command, sft);
    sft.outerTask = t;
    // 延迟或周期性任务的主要执行方法，拒绝任务或者把任务放入workQueue中
    delayedExecute(t);
    return t;
}

四个提交方法的执行流程

可以看到，四个提交任务的内容大体相同，都做了两件事：

1. 创建RunnableScheduledFuture对象
2. 调用delayedExecute(t)，这是延期或周期性任务的主要方法

delayedExecute - 计划任务的主要执行方法

下面来看看delayedExecute(t)具体都做了什么

public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {
    
    // 延迟或周期性任务的主要执行方法
    // 如果池关闭，则拒绝任务
    // 否则，将任务添加到队列并在必要时启动一个线程来运行它
    // 如果在添加任务时池被关闭，而且state和run-after-shutdown需要的话，取消并删除这个任务
    private void delayedExecute(RunnableScheduledFuture<?> task) {
        // 如果线程池关闭
        // 根据抛弃策略 RejectedExecutionHandler handler，拒绝任务 handler.rejectedExecution(command, this);
        if (isShutdown())
            reject(task);
        else {
            // 如果线程池正常运行，放入workQueue中
            super.getQueue().add(task);
            // 如果当前的线程池状态不能运行任务，就从workQueue里移除任务
            // 如果任务移除成功，取消任务
            if (!canRunInCurrentRunState(task) && remove(task))
                task.cancel(false);
            else
                // 如果线程池状态可以运行任务，或者从workQueue里移除失败，确保线程可以运行
                ensurePrestart();
        }
    }
    
    void ensurePrestart() {
        // 获取线程池中的线程池数量
        int wc = workerCountOf(ctl.get());
        // 如果线程数少于核心线程数，创建一个核心线程
        if (wc < corePoolSize)
            addWorker(null, true);
        // 如果线程数为0，创建一个非核心线程
        else if (wc == 0)
            addWorker(null, false);
    }
}

delayedExecute()方法中，最核心的内容就是super.getQueue().add(task);，也就是把创建的RunnableScheduledFuture对象放入线程池的workQueue中

放入队列后，按线程池的实际情况决定是否创建新的工作线程

总结

提交任务时，主要做了两件事：

1. 根据Runnable/Callable对象、执行时间等入参，创建RunnableScheduledFuture对象，将一个普通的Runnable/Callable对象包装计划任务
2. 调用delayedExecute(t)方法，把这个包装好的任务放入队列中，如果有需要的话，为线程池创建新的工作线程

在提交任务中，线程池做的事情十分简单，无非是创建任务、放入队列

提交任务以后，线程池中存活的工作线程worker就可以从工作队列workQueue中提取计划任务并执行：

// 计划线程池ScheduledThreadPoolExecutor 是 线程池ThreadPoolExecutor 的子类
public class ThreadPoolExecutor extends AbstractExecutorService {
    
    final void runWorker(Worker w) {
        // ...
        while (task != null || (task = getTask()) != null) {
            // ...
            task.run();
            // ...
        }
        // ...
    }

    // 从工作队列中提取任务
    private Runnable getTask() {
        // ...
        for (;;) {
            // ...
            // 通过poll/take方法提取任务
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            // ...
        }
        
    }
}

可以看到，工作线程worker从工作队列workQueue中提取任务以后，直接调用task.run()执行任务

因此，计划任务执行时的具体步骤就在RunnableScheduledFuture 类的run()方法中，下面就来看看RunnableScheduledFuture这个任务是如何执行的

ScheduledFutureTask - 计划任务

计划任务ScheduledFutureTask是计划任务线程池ScheduledThreadPoolExecutor的一个内部类，先看一下这个类的继承关系

继承关系

ScheduledFuture、 RunnableScheduledFuture、 ScheduledFutureTask的关系（实线为继承，虚线为实现）：

Delayed、ScheduledFuture、RunnableScheduledFuture的源码：

// 继承Comparable接口，表示该类对象支持排序
// 子类需要实现Comparable中的compareTo方法
public interface Delayed extends Comparable<Delayed> {
    // 返回对象的剩余延迟，零或负值表示延迟已经过去
    long getDelay(TimeUnit unit);
}

// 仅仅继承了Delayed和Future接口，自己没有任何代码
public interface ScheduledFuture<V> extends Delayed, Future<V> {}

public interface RunnableScheduledFuture<V> extends RunnableFuture<V>, ScheduledFuture<V> {
    // 如果此任务是周期性的，返回true
    // 如果此任务只执行一次，返回false
    // 周期性任务可能会根据某个计划重新运行，一个非周期性任务只能运行一次
    boolean isPeriodic();
}

接口的实现

对于上面展示的三个接口，ScheduledThreadPoolExecutor的实现源码如下：

public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService {

    private class ScheduledFutureTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V> {
        
        // 任务开始执行的时间，单位为纳秒ns
        private volatile long time;
        
        // 重复任务的周期，以纳秒为单位
        // 正数表示固定速率执行（上一次任务开始执行的period时间以后，执行下一次任务）
        // 负数表示固定延迟执行（上一次任务执行完成的period时间以后，执行下一次任务）
        // 0表示非周期性任务（只执行一次）
        private final long period;
        
        // 实现Delay接口的方法，返回任务开始执行的剩余时间
        public long getDelay(TimeUnit unit) {
            return unit.convert(time - System.nanoTime(), NANOSECONDS);
        }
        
        // 实现Comparable接口的方法，用于比较两个ScheduledFutureTask任务的大小
        // 因为计划任务线程池的workQueue是有序的，把任务放入队列中的时候，就会使用compareTo方法进行比较两个任务执行时间的先后
        public int compareTo(Delayed other) {
            if (other == this) // compare zero if same object
                // 同一个任务，返回0
                return 0;
            // 如果是ScheduledFutureTask类型的任务
            if (other instanceof ScheduledFutureTask) {
                ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
                long diff = time - x.time;
                if (diff < 0)
                    // 当前任务的执行时间早于other，要排在队列的前面
                    return -1;
                else if (diff > 0)
                    // 当前任务的执行时间晚于other，要排在队列的后面
                    return 1;
                else if (sequenceNumber < x.sequenceNumber)
                    // 如果执行时间相等，比较序号大小，序号小的排前面
                    return -1;
                else
                    return 1;
            }
            // 如果任务类型不是ScheduledFutureTask，通过getDelay()方法获取两个任务距离执行的剩余时间，然后比较
            long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
            return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
        }
        
        // 是否周期性任务
        public boolean isPeriodic() {
            return period != 0;
        }
        
    }
}

构造方法

public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {

    private class ScheduledFutureTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V> {
        
        // 任务开始执行的时间，单位为纳秒ns
        private volatile long time;
        
        // 重复任务的周期，以纳秒为单位
        // 正数表示固定速率执行（上一次任务开始执行的period时间以后，执行下一次任务）
        // 负数表示固定延迟执行（上一次任务执行完成的period时间以后，执行下一次任务）
        // 0表示非周期性任务（只执行一次）
        private final long period;
        
        ScheduledFutureTask(Runnable r, V result, long triggerTime, long sequenceNumber) {
            // 调用父类 FutureTask 的构造方法
            super(r, result);
            // 任务下次执行的时间
            this.time = triggerTime; 
            // 周期任务的间隔，正数表示按照固定速率，负数表示按照固定时延,0表示不是周期任务
            this.period = 0; 
            // 任务的序列号
            this.sequenceNumber = sequenceNumber; 
        }
        
        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;
        }
    }
}

run()方法

看完上面关于ScheduledFutureTask的源码，对ScheduledFutureTask的基础属性有了一些了解，接下来看一下它最核心的方法：源自RunnableFuture类的run()方法

根据 RunnableScheduledFuture - 继承关系 中的类图，ScheduledFutureTask继承了FutureTask类，FutureTask类实现了RunnableFuture接口（run()是这个接口唯一的方法）

不过，RunnableFuture接口也继承了Runnable接口（run()也是这个接口唯一的方法），因此也可以说ScheduledFutureTask的run()源自Runnable接口

虽然两个接口里面run()方法的的方法签名都是void run()，但是接口上面的注释不一样，也就是说，虽然方法签名一样，但是JDK希望这两个run()方法在实现时完成的功能，是有所区别的，感兴趣的同学可以去了解一下

run()方法的源码如下：

public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {

    private class ScheduledFutureTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V> {
        
        // 计划任务具体执行时的执行内容
        public void run() {
            // 判断任务是否可以运行，如果当前的线程池状态不能执行任务，则取消任务
            if (!canRunInCurrentRunState(this))
                cancel(false);
            // 如果不是周期性任务，直接调用父类FutureTask的run方法，执行一次任务，会设置结果（private Object outcome）
            else if (!isPeriodic())
                super.run();
            
            // 是周期性任务，需要多次执行，调用FutureTask的runAndReset方法
            // runAndReset：直接执行计算，执行完以后不会设置任务的执行结果(FutureTask中的private Object outcome)
            //              执行完以后还会把这个Future重置为初始状态NEW(run方法就不会重置状态)
            //              如果任务成功运行并重置，返回true
            // 如果任务成功运行而且重置Future，设置任务下一次执行的时间，并将该任务重新入队，等待再次被调度
            else if (super.runAndReset()) {
                // 设置下次执行的时间
                setNextRunTime();
                // 重新排队周期任务
                reExecutePeriodic(outerTask);
            }
        }// run()
    }// ScheduledFutureTask
    
    // 为周期性任务设置下次执行的时间
    private void setNextRunTime() {
        long p = period;
        // 固定速率，不在乎上一次任务是否完成，下次任务执行时间 = 上一次任务执行时间 + 指定周期
        if (p > 0)
            time += p;
        // 固定延迟，上一次任务完成以后才开始计算时间，下次任务执行时间 = 上次任务执行完成的时间 + 指定周期
        else
            // 固定延迟，p是负数，需要变回正数
            time = triggerTime(-p);
    }
    
    // 除非当前线程池状态不能运行该任务，不然就重新排队定期任务
    void reExecutePeriodic(RunnableScheduledFuture<?> task) {
        if (canRunInCurrentRunState(task)) {
            // 放入线程池的workQueue中
            super.getQueue().add(task);
            // 利用短路原则
            // 首先判断线程池状态是否可以运行该任务
            // 如果任务可以运行，调用ensurePrestart确保任务可以运行
            // 如果任务不能运行，尝试从workQueue中移除任务，如果移除task失败，也要调用ensurePrestart确保任务可以运行
            if (canRunInCurrentRunState(task) || !remove(task)) {
                // 确保任务可以运行
                ensurePrestart();
                return;
            }
        }
        // 如果当前线程池状态不能运行该任务，而且从workQueue中移除任务成功，取消该任务
        task.cancel(false); 
    }

    // 确保线程池可以运行任务
    void ensurePrestart() {
        // 获取线程池中的线程池数量
        int wc = workerCountOf(ctl.get());
        // 如果线程数少于核心线程数，创建一个核心线程
        if (wc < corePoolSize)
            addWorker(null, true);
        // 如果线程数为0，创建一个非核心线程
        else if (wc == 0)
            addWorker(null, false);
    }
    
}

在run()方法中，简单地说，执行流程如下：

• 对于非周期性任务，只需要运行一次，直接让工作线程执行这个任务就完事了

• 对于周期性任务，需要运行多次，处理步骤如下：

  1. 执行任务
  2. 设置任务下一次执行的时间
  3. 把任务放入队列

  至此，一次周期性任务就执行完毕

总结

到这里，我们知道计划任务在提交之后，会被放入线程池的workQueue中，在任务执行时

• 如果是非周期性任务，会直接执行
• 如果是非周期性，执行完成后，会把任务再放入workQueue中，线程池中的存活的工作线程会一直从workQueue中提取任务

还记得在文章开头提到的两个特性吗？

1. 多次执行任务
2. 在指定时间执行任务

现在，对于第一个特性：多次执行任务，我们已经可以给出答案：

对于需要多次执行的周期性任务，任务执行完以后会再次放入线程池的workQueue中，工作线程可以从workQueue中提取任务并执行，

这就可以实现任务的多次执行

接下来，尝试解决第二个疑问，ScheduledThreadPoolExecutor如何在指定时间执行任务

目前为止，关于线程池比较重要的部分：提交任务、执行任务的run()方法、工作队列，除了工作队列以外，我们都了解得差不多了，下面就来看看工作队列

DelayedWorkQueue

介绍

还记得 ScheduledThreadPoolExecutor - 类结构 中提到的构造方法吗？

public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {
    
    // 计划任务线程池的构造方法之一
    // 注意，这里使用的workQueue是DelayedWorkQueue
    public ScheduledThreadPoolExecutor(int corePoolSize,ThreadFactory threadFactory) {
        super(corePoolSize, Integer.MAX_VALUE,
              DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
              new DelayedWorkQueue(), threadFactory);
    }
}

当时我们发现，这里的workQueue使用的是DelayedWorkQueue，这是一个特殊的阻塞工作队列，它是ScheduledThreadPoolExecutor的一个内部类

ScheduledThreadPoolExecutor使用DelayedWorkQueue来存放任务，也就是存放ScheduledFutureTask对象，当线程池的工作线程调用take/poll方法尝试从DelayedWorkQueue中提取队首任务（将队首任务出队并返回）时，如果任务的执行时间还没到，就会阻塞这个工作线程，直到任务的执行时间来临，take/poll方法返回队首任务

数据结构

DelayedWorkQueue是一个无界优先队列，使用数组存储，底层使用最小堆来实现优先队列的功能

最小堆，是一种经过排序的完全二叉树，其中任一非终端节点的数据值均不大于其左子节点和右子节点的值

这里，我们不关心DelayedWorkQueue是如何使用最小堆来实现优先队列的，我们只要知道它是一个有序队列即可

DelayedWorkQueue里面的ScheduledFutureTask对象按照任务执行时间的先后排序，最早执行的任务放在队首，因此，线程池的工作线程worker只需要关心队首任务即可，如果队首任务的执行时间还未到，工作线程worker应该继续等待

DelayedWorkQueue中存放的ScheduledFutureTask对象是可比较的

在 RunnableScheduledFuture - 接口的实现 里面我们提到，ScheduledFutureTask间接实现了Comparable接口，因此ScheduledFutureTask可以通过compareTo方法进行比较

成员变量

源码

public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {

    static class DelayedWorkQueue extends AbstractQueue<Runnable>
    implements BlockingQueue<Runnable> {
        
        // 队列初始容量
        private static final int INITIAL_CAPACITY = 16;
        
        // 数组用来存储定时任务，通过数组实现堆排序
        private RunnableScheduledFuture[] queue = new RunnableScheduledFuture[INITIAL_CAPACITY];
        
        // leader线程负责等待队首任务的执行时间点到达，然后把队首任务出队，并把任务作为take方法的返回值返回
        // 线程池中会有很多线程调用take，这些线程中最早拿到锁的那个线程就可能成为leader线程
        private Thread leader;
        
        // 锁和监视器，线程池中的线程调用take方法竞争成为leader线程时使用，第一个拿到锁的线程就可以成为leader
        private final ReentrantLock lock = new ReentrantLock();
        // 工作线程都会调用take/poll方法获取队首元素
        // 如果队首元素还没有到执行时间，工作线程会调用 available.await() 或 available.awaitNanos(delay) 进入阻塞
        private final Condition available = lock.newCondition();
        ...
    }
}

leader线程

这里有一个很重要的概念，leader线程，这是ScheduledThreadPoolExecutor针对自身情况的一个优化措施

我们知道，一个线程池中可能会有很多个工作线程worker，这些工作线程会不断调用workQueue的take/poll方法提取任务，然后执行任务，任务执行完以后再继续从workQueue里面提取任务，线程池的线程复用就是这么实现的

在计划任务线程池中，会出现这么一个问题：由于workQueue中的任务是按时间顺序排列的，只要队首的任务没有到达执行时间，那么后面的任务也一定没有到达执行时间

假设现在workQueue中有三个任务，A、B、C，它们的执行时间顺序为A -> B -> C

当很多个工作线程worker一起调用take/poll方法时，这些工作线程都尝试从workQueue中获取队首任务A，如果A的执行时间都还没有到来，那么他们全都会阻塞，直到A的执行时间到来，这些线程一起被唤醒，然后纷纷尝试获取A

但是，只有一个工作线程的take/poll方法可以成功获取任务A，当任务A被取走之后，其他线程会发现队首任务变成了B，一般来说，因为刚刚取走任务A，任务B的执行时间离现在还有一段距离，因此这些线程又会进入等待，直到任务B的执行时间到来，又纷纷尝试从队列中提取B

鉴于workQueue的有序性，完全可以让大部分工作线程都进入等待状态，只留下一个工作线程来尝试获取workQueue中的队首任务，这个工作线程就是leader线程

当leader线程取走workQueue中的队首任务以后，就需要去执行这个队首任务，于是它会唤醒一个处于等待状态的工作线程，这个工作线程就会成为新的leader线程，让新的leader线程来尝试获取队首任务

通过这样的方式，可以避免大量工作线程反复地在 等待 - 唤醒 两种状态中切换

leader线程与其他线程的异同

leader线程 与 线程池中其他调用take方法的工作线程 之间存在一些异同：

• 共同点：available.signal()的时候，无论是leader线程还是其他线程，都有可能被唤醒
• 差异点：leader线程会调用awaitNanos(delay)，队首任务的执行时间点到达时会自动唤醒，而其他线程则调用await()无限期地等待

如果不能理解leader线程的作用，没有关系，我们先来看提取任务的take方法是如何实现的

take - 将任务出队并返回

在前面的 DelayedWorkQueue - 介绍 部分有提到，当线程池的工作线程调用take/poll方法尝试从DelayedWorkQueue中提取队首任务时，如果任务的执行时间还没到，就会阻塞这个工作线程，直到任务的执行时间来临，take/poll方法提取队首任务并返回

下面我们就来看看，take方法具体是怎么做到的

源码

public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService {

    static class DelayedWorkQueue extends AbstractQueue<Runnable>
    implements BlockingQueue<Runnable> {
        
        // leader线程负责等待队首任务的执行时间点到达，然后把队首任务出队，并把任务作为take方法的返回值返回
        private Thread leader;
        
        public RunnableScheduledFuture<?> take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            // 加锁
            lock.lockInterruptibly();
            try {
                // 自旋
                for (;;) {
                    // 获取队首任务（最小堆的堆顶）
                    RunnableScheduledFuture<?> first = queue[0];
                    // 如果队首是null，证明队列没有任务，当前线程阻塞
                    // 阻塞以后，有两种唤醒可能：
                    // 1. 有其他工作线程调用offer方法，往队列放入任务，并使用available.signal()时，当前线程有可能被唤醒
                    // 2. 线程因为队列没有任务而阻塞以后，有offer方法放入任务，但是没有被唤醒
                    //    直到leader线程准备执行任务，放弃自己的leader地位，使用available.signal()唤醒一个线程
                    //    这时，当前线程也有可能被唤醒
                    if (first == null)
                        available.await();
                    // 如果队列里面有任务
                    else {
                        // 计算队首任务在多久以后执行
                        long delay = first.getDelay(NANOSECONDS);
                        // 如果小于等于0，证明任务现在要执行，或者应该在过去执行
                        if (delay <= 0L)
                            // 从队列（队列实际上用堆实现）里面移除任务（然后重新调整为最小堆），并返回任务
                            // 注意，调用finishPoll(first)可以得到一个任务，return语句会把这个任务作为take方法的返回值
                            // 在take方法返回之前，会执行finally语句的内容，这部分内容在最下面的finally
                            return finishPoll(first);
                        
                        // 如果还没有到执行时间
                        first = null; // don't retain ref while waiting
                        // 如果leader线程不为空，说明队首任务已经有线程在等待
                        if (leader != null)
                            // 当前线程阻塞，直到其他线程调用available.signal()，当前工作线程恰好被唤醒
                            // 有以下两种被唤醒的可能：
                            // 1. 有线程调用offer方法，使队首任务变更，调用signal唤醒一个线程，恰好当前线程被唤醒
                            // 2. 旧leader线程从队列中提取任务返回，调用signal唤醒一个线程作为新leader，恰好当前线程被唤醒
                            available.await();
                        else {
                            // 如果leader线程为空，当前线程成为leader线程
                            Thread thisThread = Thread.currentThread();
                            leader = thisThread;
                            try {
                                // awaitNanos(delay)，等待delay时间，有两种被唤醒的可能：
                                // 1. 等待时间到达，自动唤醒，此时，醒来以后就可以从队列中提取任务并返回
                                // 2. 等待时间没到达，但是有线程调用offer方法放入新任务，新任务的执行时间更早，成为新队首
                                //    此时，offer方法会调用signal唤醒一个正在等待的线程
                                //    被唤醒的线程恰好是当前线程，那就继续当leader
                                //    2.1 如果新的队首任务刚好是现在执行，那就执行
                                //    2.2 如果新的队首任务在未来执行，继续awaitNanos(delay)
                                //        但是此时的delay变了，变成新的队首任务的delay
                                available.awaitNanos(delay);
                            } finally {
                                // 在上面线程调用awaitNanos等待一段时间，当线程被唤醒以后，会执行finally的内容
                                
                                // 在上面的两种唤醒可能中
                                // 如果是2，那么醒来以后判断一定不成立，因为offer会清除leader (leader = null)

                                // 如果是1，醒来以后当前线程仍然是leader线程
                                // 由于唤醒原因是队首任务执行的时间到了，当前线程需要从队列中提取队首任务
                                // 所以清除leader，为leader的竞争作准备（但还没有开始，signal以后才开始）
                                // 然后自己在下一轮for循环中if (delay <= 0L)判断成立
                                // 当前线程从队列中提取任务并返回，在return前会调用下面的finally，进行条件判断
                                // 如果条件合适，会调用signal唤醒一个线程，
                                if (leader == thisThread)
                                    leader = null;
                            }
                        }
                    }
                }//for(;;)
            } finally {
                // 在 return finishPoll(first) 返回任务之前，会执行finally的代码
                // 如果leader为null，证明队首任务没有线程在等待
                // 如果队首不为空，证明还有任务需要执行
                // 有任务，又没有leader线程，那就唤醒一个线程来成为leader
                // 正在等待的工作线程会竞争锁，竞争成功的工作线程就可以解除阻塞
                if (leader == null && queue[0] != null)
                    available.signal();
                lock.unlock();
            }
        }// take
    }
}

执行流程

简单来说，take()方法的流程如下：

• 如果队首任务需要被执行，把任务出队，如果队列里还有任务需要执行，而且没有leader线程，就唤醒正在等待available的线程
• 如果队列为空，或者还没到执行时间，有两种等待模式
  • 如果没有leader线程，当前线程成为leader线程，awaitNanos(delay)等待任务执行时间到达后自动唤醒
  • 如果已有leader线程，无限期等待available.signal

执行案例

总结

分析到这里，我们可以得出第二个问题的答案

线程池如何实现在指定时间执行任务？

是通过特殊的工作队列，也就是DelayedWorkQueue实现的，工作线程会调用take方法从工作队列里面提取任务，如果任务的执行时间还没有到来，那么工作线程会阻塞一段时间，当任务的执行时间到来时，工作线程醒来，成功从工作队列中提取任务，并执行这个任务

poll - 在限期内，将任务出队并返回

poll方法与take方法在大体上相同，都是从队列中提取队首任务，但是有一点不同：

• take方法有可能会无限阻塞工作线程
• poll方法不会无限阻塞工作线程，如果阻塞的时间超过指定时间timeout，poll方法就会直接返回null

public RunnableScheduledFuture<?> poll(long timeout, TimeUnit unit) throws InterruptedException {
    // 可等待的最长时间
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    // 加锁
    lock.lockInterruptibly();
    try {
        // 自旋
        for (;;) {
            // 获取队首任务
            RunnableScheduledFuture<?> first = queue[0];
            // 如果队列为空
            if (first == null) {
                if (nanos <= 0L)
                    // 可等待的时间已到，直接返回null
                    return null;
                else
                    // 可等待的时间已到，工作线程阻塞一段时间
                    nanos = available.awaitNanos(nanos);
            } else {
                // 任务多久以后执行
                long delay = first.getDelay(NANOSECONDS);
                if (delay <= 0L)
                    // 任务的执行时间到了，执行
                    return finishPoll(first);
                if (nanos <= 0L)
                    // 任务的执行时间没到，但是可等待的时间到了，返回null
                    return null;
                first = null; // don't retain ref while waiting
                if (nanos < delay || leader != null)
                    // nanos：剩余可等待时间
                    // delay：距离任务执行的时间
                    
                    // 如果nanos < delay，即使把时间全部等完，任务也还没有执行，但是，有可能会有新的任务放进来
                    //                   这个新放入的任务的执行时间可能会比较早，成为新的队首任务，所以还是阻塞一段时间
                    //                   如果恰好有新的任务放入，成为新队首，就会唤醒一个线程，让它成为leader
                    //                   此时，如果当前线程恰好被唤醒，当前线程就可以成为leader
                    //                   因为新的队首任务的执行时间比旧队首的早，当前线程还是有机会在nanos时间内拿到任务的
                    
                    // 如果nanos >= delay，在等待时间内，任务的执行时间会到来，而且已经有leader线程，队首任务已经线程在等待
                    //                    如果leader线程把队首任务执行完以后，把当前线程唤醒，当前线程成为leader线程
                    //                    此时，当前线程就有机会可以获取新的队首任务
                    //                    因此，当前工作线程阻塞一段时间，等待被唤醒成为leader
                    nanos = available.awaitNanos(nanos);
                else {
                    // 如果nanos >= delay，而且没有leader线程
                    // 证明当前工作线程有机会等到任务执行，并且leader = null，没有leader线程
                    // 那么当前线程就可以成为leader线程
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        // 阻塞一段时间，任务执行时间到达时会被自动唤醒
                        long timeLeft = available.awaitNanos(delay);
                        // 被唤醒以后，看一下线程的可等待时间还剩多少
                        nanos -= delay - timeLeft;
                    } finally {
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        if (leader == null && queue[0] != null)
            available.signal();
        lock.unlock();
    }
}

poll方法的具体内容与take方法差不多，只是多了一个可等待时间timeout，因此不多赘述

offer - 将任务入队

虽然两个疑问都已经解决，但是我们还是要了解一下DelayedWorkQueue取出任务的方法

源码

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);
        } else {
            // 如果队列（实际上用最小堆实现）不空，放入元素，并重新调整堆
            siftUp(i, e);
        }
        // 如果放入的任务处于队首（是队列中最早的任务）
        if (queue[0] == e) {
            // 清除leader线程
            leader = null;
            // 通知一个等待的线程：
            // 队首任务被更换，旧leader线程的awaitNanos(delay)中的delay太久了
            // 等它自动唤醒的时候，新的队首任务的执行时间已经过去
            // 所以，唤醒一个等待中的线程，醒来的线程会成为新的leader线程，然后调用awaitNanos(delay)
            // 这个delay是新的队首任务的delay，等新任务执行时间到来的时候自动唤醒
            available.signal();
        }
    } finally {
        lock.unlock();
    }
    return true;
}

当一个新的任务成为队首，或者需要有新的线程成为leader时，available监视器上的线程将会被通知，然后竞争成为leader线程，有些类似于生产者-消费者模式

为什么signal之前要清除leader线程

在offer方法里面，有这么一段代码

if (queue[0] == e) {
    leader = null;
    available.signal();
}

这里面，如果放入的新任务出于队首，代表处于队首的任务发生了变更，程序不仅会调用available.signal()唤醒线程，还会清除leader线程，这是为什么呢？

假设新加入的任务是A，原来的队首任务是B，这两个任务的时间顺序为A -> B

在take方法里面，旧的leader线程会调用available.awaitNanos(delayB)进入阻塞，直到被available.signal()唤醒，或者B任务的执行时间delayB到达，线程自动唤醒，然后清除自己的leader标记(leader = null)，然后拿出队首任务返回

但是，现在放入A任务后，应该先执行A任务，再执行B任务

如果只是signal，那么leader线程和其他线程都可能被唤醒

1. 被唤醒的是旧的leader线程，它会发现任务可以执行，然后清空自己的leader身份(leader = null)，从队列中提取任务并返回
2. 其他线程拿到锁，它会发现，已经存在leader线程，于是再次进入睡眠

除非被signal()唤醒的线程是旧的leader线程，否则被唤醒的线程都会重新进入睡眠，直至leader线程被唤醒，其他线程完全在浪费自己竞争到的CPU时间片

所以，这里除了signal信号唤醒正在等待的线程以外，还要把leader清空

清除leader标记(leader = null)以后，无论是旧的leader线程被唤醒，还是其他线程被唤醒，都能成为新的leader

计划任务线程池原理总结

在文章开头，我们针对计划任务线程池的实现提出了两点特性

1. 多次执行任务
2. 在指定时间执行任务

在分析的过程中，我们针对ScheduledThreadPoolExecutor的三部分内容进行源码分析：

• 任务提交：ScheduledThreadPoolExecutor.schedule()方法
• 任务执行：ScheduledThreadPoolExecutor.ScheduledFutureTask.run()方法
• 工作队列：ScheduledThreadPoolExecutor.DelayedWorkQueue类中的take()方法和offer()方法

在分析过程中，我们逐渐了解到ScheduledThreadPoolExecutor是如何实现这两点特性，现在，我们再来回顾一下

1. 多次执行任务：工作线程worker在工作时，会从工作队列workQueue中提取任务，然后执行任务，本次任务执行完以后，设定任务下一次执行的时候，然后将任务再次放入工作队列workQueue，工作线程worker就可以再次从工作队列workQueue中提取这个任务，然后执行，周而复始，就可以做到多次执行任务

2. 在指定时间执行任务：ScheduledThreadPoolExecutor使用特定的工作队列DelayedWorkQueue实现，工作线程worker在工作时，会从工作队列workQueue中提取任务，在提取任务时，如果任务还没有到执行的时间，那么工作线程worker就会阻塞一段时间，直到任务的执行时间到来，工作线程worker自动唤醒，成功从工作队列workQueue中提取任务，然后执行

   通过阻塞的方式，让工作线程worker进入阻塞，直到任务执行时间到来，工作线程才能成功拿到任务并执行，这就可以做到：任务只有在指定时间到来以后，才能执行

但是，由于使用队列来实现定时器，有出入队调整堆等操作，所以定时并不是非常非常精确

另外，有些内容，比如FutureTask的run方法会设置执行结果outcome，但是runAndReset方法就不会设置执行结果，以及DelayedWorkQueue中最小堆的具体实现，这些内容与计划任务线程池的主要原理关系不是很大，因此只是简单提及，不多赘述，感兴趣同学可以自行了解