定时线程池

定时线程池类的类结构图

它用来处理延时任务或定时任务。

提交任务方式

它接收SchduledFutureTask类型的任务，是线程池调度任务的最小单位，有三种提交任务的方式：

• schedule 延迟执行

• scheduledAtFixedRate 延迟执行，如果任务执行时间超过周期时间，则任务执行完成后，马上执行下一个任务

• scheduledWithFixedDelay 延迟执行，如果任务执行时间超过周期时间，则任务执行完成后，再过一个周期，执行下一个任务

任务成员变量

• private long time：任务开始的时间

• private final long sequenceNumber;：任务的序号

• private final long period：任务执行的时间间隔

任务存储方式

采用DelayQueue存储等待的任务

1. DelayQueue内部封装了一个PriorityQueue，它会根据time的先后时间排序，若time相同则根据sequenceNumber排序；
2. DelayQueue也是一个无界队列

 public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }

工作线程的执行过程

• 工作线程会从DelayQueue取已经到期的任务去执行；
• 执行结束后重新设置任务的到期时间，再次放回DelayQueue

示例

@Slf4j
public class ScheduledThreadPool {

    static ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);

    public static void main(String[] args) {
        log.info("开始执行");
        scheduledThreadPoolExecutor.schedule(()->{
            log.info("延迟一秒执行");
        },1,TimeUnit.SECONDS);

        //每2秒执行一次,上一个任务完成后马上执行
        scheduledThreadPoolExecutor.scheduleAtFixedRate(()->{
            log.info("延迟一秒执行,每2秒执行一次");
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },1,2,TimeUnit.SECONDS);

        //每2秒执行一次，在上一个任务执行结束后2秒，再执行
        scheduledThreadPoolExecutor.scheduleWithFixedDelay(()->{
            log.info("延迟一秒执行,每2秒执行一次");
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },1,2,TimeUnit.SECONDS);
    }
}

源码分析

ScheduledThreadPoolExecutor会把待执行的任务放到工作队列DelayQueue中，DelayQueue封装了一个PriorityQueue，PriorityQueue会对队列中的ScheduledFutureTask进行排序，具体的排序算法实现如下：

public int compareTo(Delayed other) {
    if (other == this) // compare zero if same object
        return 0;
    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;
    }
    long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
    return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
}

1. 首先按照time排序，time小的排在前面，time大的排在后面；
2. 如果time相同，按照sequenceNumber排序，sequenceNumber小的排在前面，sequenceNumber大的排在后面，换句话说，如果两个task的执行时间相同，优先执行先提交的task。

SchduledFutureTask之run方法实现

run方法是调度task的核心，task的执行实际上是run方法的执行。

public void run() {
    boolean periodic = isPeriodic();
//如果当前线程池已经不支持执行任务，则取消
    if (!canRunInCurrentRunState(periodic))
        cancel(false);
//如果不需要周期性执行，则直接执行run方法然后结束
    else if (!periodic)
        ScheduledFutureTask.super.run();
//如果需要周期执行，则在执行完任务以后，设置下一次执行时间
    else if (ScheduledFutureTask.super.runAndReset()) {
        // 计算下次执行该任务的时间
 setNextRunTime();
 //重复执行任务
        reExecutePeriodic(outerTask);
    }
}

1. 如果当前线程池运行状态不可以执行任务，取消该任务，然后直接返回，否则执行步骤2；
2. 如果不是周期性任务，调用FutureTask中的run方法执行，会设置执行结果，然后直接返回，否则执行步骤3；
3. 如果是周期性任务，调用FutureTask中的runAndReset方法执行，不会设置执行结果，然后直接返回，否则执行步骤4和步骤5；
4. 计算下次执行该任务的具体时间；
5. 重复执行任务。

reExecutePeriodic方法

void reExecutePeriodic(RunnableScheduledFuture<?> task) {
    if (canRunInCurrentRunState(true)) {
        super.getQueue().add(task);
        if (!canRunInCurrentRunState(true) && remove(task))
            task.cancel(false);
        else
            ensurePrestart();
    }
}

该方法和delayedExecute方法类似，不同的是：

1. 由于调用reExecutePeriodic方法时已经执行过一次周期性任务了，所以不会reject当前任务；
2. 传入的任务一定是周期性任务。

线程池任务的提交

首先是schedule方法，该方法是指任务在指定延迟时间到达后触发，只会执行一次。

public ScheduledFuture<?> schedule(Runnable command,
                                   long delay,
                                   TimeUnit unit) {
//参数校验
if (command == null || unit == null)
        throw new NullPointerException();
//这里是一个嵌套结构，首先把用户提交的任务包装成ScheduledFutureTask
//然后在调用decorateTask进行包装，该方法是留给用户去扩展的，默认是个空方法
    RunnableScheduledFuture<?> t = decorateTask(command,
        new ScheduledFutureTask<Void>(command, null,
                                      triggerTime(delay, unit)));
   //包装好任务以后，就进行提交了
delayedExecute(t);
    return t;
}

任务提交方法：

private void delayedExecute(RunnableScheduledFuture<?> task) {
    //如果线程池已经关闭，则使用拒绝策略把提交任务拒绝掉
if (isShutdown())
        reject(task);
    else {
//与ThreadPoolExecutor不同，这里直接把任务加入延迟队列
        super.getQueue().add(task);//使用用的DelayedWorkQueue
//如果当前状态无法执行任务，则取消
        if (isShutdown() &&
            !canRunInCurrentRunState(task.isPeriodic()) &&
            remove(task))
            task.cancel(false);
        else
//这里是增加一个worker线程，避免提交的任务没有worker去执行
//原因就是该类没有像ThreadPoolExecutor一样，woker满了才放入队列
          ensurePrestart();
    }
}

总结

主要总结为以下几个方面：

• 与Timer执行定时任务的比较，相比Timer，ScheduedThreadPoolExecutor有什么优点；
• ScheduledThreadPoolExecutor继承自ThreadPoolExecutor，所以它也是一个线程池，也有coorPoolSize和workQueue，ScheduledThreadPoolExecutor特殊的地方在于，自己实现了优先工作队列DelayedWorkQueue；
• ScheduedThreadPoolExecutor实现了ScheduledExecutorService，所以就有了任务调度的方法，如schedule，scheduleAtFixedRate和scheduleWithFixedDelay，同时注意他们之间的区别；
• 内部类ScheduledFutureTask继承自FutureTask，实现了任务的异步执行并且可以获取返回结果。同时也实现了Delayed接口，可以通过getDelay方法获取将要执行的时间间隔；
• 周期任务的执行其实是调用了FutureTask类中的runAndReset方法，每次执行完不设置结果和状态。
• 详细分析了DelayedWorkQueue的数据结构，它是一个基于最小堆结构的优先队列，并且每次出队时能够保证取出的任务是当前队列中下次执行时间最小的任务。同时注意一下优先队列中堆的顺序，堆中的顺序并不是绝对的，但要保证子节点的值要比父节点的值要大，这样就不会影响出队的顺序。

总体来说，ScheduedThreadPoolExecutor的重点是要理解下次执行时间的计算，以及优先队列的出队、入队和删除的过程，这两个是理解ScheduedThreadPoolExecutor的关键。