33-ScheduledThreadPoolExecutor源码分析

程序员文章站 2022-05-20 14:27:07

...

ScheduledThreadPoolExecutor源码分析

自JDK1.5开始，JDK提供了ScheduledThreadPoolExecutor类来支持周期性任务的调度。在这之前的实现需要依靠Timer和TimerTask或者其它第三方工具来完成。但Timer有不少的缺陷：

Timer是单线程模式，如果在执行任务期间某个TimerTask耗时较久，那么就会影响其它任务的调度；
Timer的任务调度是基于绝对时间的，对系统时间敏感；
Timer不会捕获执行TimerTask时所抛出的异常，由于Timer是单线程，所以一旦出现异常，则线程就会终止，其他任务也得不到执行。

ScheduledThreadPoolExecutor继承ThreadPoolExecutor来重用线程池的功能，它的实现方式如下：

将任务封装成ScheduledFutureTask对象，ScheduledFutureTask基于相对时间，不受系统时间的改变所影响；
ScheduledFutureTask实现了java.lang.Comparable接口和java.util.concurrent.Delayed接口，所以有两个重要的方法：compareTo和getDelay。compareTo方法用于比较任务之间的优先级关系，如果距离下次执行的时间间隔较短，则优先级高；getDelay方法用于返回距离下次任务执行时间的时间间隔；
ScheduledThreadPoolExecutor定义了一个DelayedWorkQueue，它是一个有序队列，会按照每个任务距离下次执行时间间隔的大小来排序；
ScheduledFutureTask继承自FutureTask，可以通过返回Future对象来获取执行的结果。

通过如上的介绍，可以对比一下Timer和ScheduledThreadPoolExecutor：

Timer	ScheduledThreadPoolExecutor
单线程	多线程
单个任务执行时间影响其他任务调度	多线程，不会影响
基于绝对时间	基于相对时间
一旦执行任务出现异常不会捕获，其他任务得不到执行	多线程，单个任务的执行不会影响其他线程

所以，在JDK1.5之后，没什么理由继续使用Timer进行任务调度了。

源码深入分析

先看一下ScheduledThreadPoolExecutor类结构图：

33-ScheduledThreadPoolExecutor源码分析

ScheduledThreadPoolExecutor继承自ThreadPoolExecutor，实现了ScheduledExecutorService接口，该接口定义了schedule等任务调度的方法。

同时ScheduledThreadPoolExecutor有两个重要的内部类：DelayedWorkQueue和ScheduledFutureTask。DelayeddWorkQueue是一个阻塞队列，而ScheduledFutureTask继承自FutureTask，并且实现了Delayed接口。

ScheduledThreadPoolExecutor的构造方法

ScheduledThreadPoolExecutor有4个构造方法：

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

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

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

public ScheduledThreadPoolExecutor(int corePoolSize,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
          new DelayedWorkQueue(), threadFactory, handler);
}

因为ScheduledThreadPoolExecutor继承自ThreadPoolExecutor，所以这里都是调用的ThreadPoolExecutor类的构造方法。这里有三点需要注意：

使用DelayedWorkQueue作为阻塞队列，并没有像ThreadPoolExecutor类一样开放给用户进行自定义设置。该队列是ScheduledThreadPoolExecutor类的核心组件，后面详细介绍。
这里没有向用户开放maximumPoolSize的设置，原因是DelayedWorkQueue底层使用DelayedQueue，而DelayedQueue底层使用PriorityQueue，PriorityQueue最大大小为Integer.MAX_VALUE，也就是说队列不会装满，maximumPoolSize参数即使设置了也不会生效。
worker线程没有回收时间，原因跟第2点一样，因为不会触发回收操作。所以这里的线程存活时间都设置为0。

当我们创建出一个调度线程池以后，就可以开始提交任务了。

schedule方法

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

public ScheduledFuture<?> schedule(Runnable command,
                                   long delay,
                                   TimeUnit unit) {
    if (command == null || unit == null)
        throw new NullPointerException();
    if (delay < 0) delay = 0;
    long triggerTime = now() + unit.toNanos(delay);
    RunnableScheduledFuture<?> t = decorateTask(command,
        new ScheduledFutureTask<Boolean>(command, null, triggerTime));
    delayedExecute(t);
    return t;
}

public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                       long delay,
                                       TimeUnit unit) {
    if (callable == null || unit == null)
        throw new NullPointerException();
    if (delay < 0) delay = 0;
    long triggerTime = now() + unit.toNanos(delay);
    RunnableScheduledFuture<V> t = decorateTask(callable,
        new ScheduledFutureTask<V>(callable, triggerTime));
    delayedExecute(t);
    return t;
}

两个重载的schedule方法只是传入的第一个参数不同，可以是Runnable对象或者Callable对象。会把传入的任务封装成一个RunnableScheduledFuture对象，其实也就是ScheduledFutureTask对象，decorateTask默认什么功能都没有做，子类可以重写该方法：

/**
 * 修改或替换用于执行 runnable 的任务。此方法可重写用于管理内部任务的具体类。默认实现只返回给定任务。
 */
protected <V> RunnableScheduledFuture<V> decorateTask(
    Runnable runnable, RunnableScheduledFuture<V> task) {
    return task;
}

/**
 * 修改或替换用于执行 callable 的任务。此方法可重写用于管理内部任务的具体类。默认实现只返回给定任务。
 */
protected <V> RunnableScheduledFuture<V> decorateTask(
    Callable<V> callable, RunnableScheduledFuture<V> task) {
    return task;
}

然后，通过调用delayedExecute方法来延时执行任务。最后，返回一个ScheduledFuture对象。

scheduleAtFixedRate方法

该方法设置了执行周期，下一次执行时间相当于是上一次的执行时间加上period，它是采用固定的频率来执行任务：

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                              long initialDelay,
                                              long period,
                                              TimeUnit unit) {
    if (command == null || unit == null)
        throw new NullPointerException();
    if (period <= 0)
        throw new IllegalArgumentException();
    if (initialDelay < 0) initialDelay = 0;
    long triggerTime = now() + unit.toNanos(initialDelay);
    RunnableScheduledFuture<?> t = decorateTask(command,
        new ScheduledFutureTask<Object>(command,
                                        null,
                                        triggerTime,
                                        unit.toNanos(period)));
    delayedExecute(t);
    return t;
}

scheduleWithFixedDelay方法

该方法设置了执行周期，与scheduleAtFixedRate方法不同的是，下一次执行时间是上一次任务执行完的系统时间加上period，因而具体执行时间不是固定的，但周期是固定的，是采用相对固定的延迟来执行任务：

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                 long initialDelay,
                                                 long delay,
                                                 TimeUnit unit) {
    if (command == null || unit == null)
        throw new NullPointerException();
    if (delay <= 0)
        throw new IllegalArgumentException();
    if (initialDelay < 0) initialDelay = 0;
    long triggerTime = now() + unit.toNanos(initialDelay);
    RunnableScheduledFuture<?> t = decorateTask(command,
        new ScheduledFutureTask<Boolean>(command,
                                         null,
                                         triggerTime,
                                         unit.toNanos(-delay)));
    delayedExecute(t);
    return t;
}

注意这里的unit.toNanos(-delay)，这里把周期设置为负数来表示是相对固定的延迟来执行。

scheduleAtFixedRate 和 scheduleWithFixedDelay的区别

scheduleAtFixedRate把周期延迟时间传入ScheduledFutureTask中，而scheduleWithFixedDelay却设置成负数传入，区别在哪里呢？看一下ScheduledFutureTask中，运行周期任务的方法runPeriodic的源码：

private void runPeriodic() {
    boolean ok = ScheduledFutureTask.super.runAndReset();
    boolean down = isShutdown();
    // Reschedule if not cancelled and not shutdown or policy allows
    if (ok && (!down ||
               (getContinueExistingPeriodicTasksAfterShutdownPolicy() &&
                !isTerminating()))) {
        long p = period;
        if (p > 0)
            // 大于0是scheduleAtFixedRate方法，表示执行时间是根据初始化参数计算的
            time += p;
        else
             // 小于0是scheduleWithFixedDelay方法，表示执行时间是根据当前时间重新计算的
            time = now() - p;
        ScheduledThreadPoolExecutor.super.getQueue().add(this);
    }
    // This might have been the final executed delayed
    // task.  Wake up threads to check.
    else if (down)
        interruptIdleWorkers();
}

也就是说当使用scheduleAtFixedRate方法提交任务时，任务后续执行的延迟时间都已经确定好了，分别是initialDelay，initialDelay + period，initialDelay + 2 * period以此类推。

而调用scheduleWithFixedDelay方法提交任务时，第一次执行的延迟时间为initialDelay，后面的每次执行时间都是在前一次任务执行完成以后的时间点上面加上period延迟执行。

ScheduledFutureTask

从上面几个调度方法的源码可以得知，任务都被封装成ScheduledFutureTask对象，下面看一下ScheduledFutureTask的源码。

ScheduledFutureTask的构造方法

ScheduledFutureTask继承自FutureTask并实现了RunnableScheduledFuture接口，构造方法如下：

/**
 * Creates a one-shot action with given nanoTime-based trigger time.
 */
ScheduledFutureTask(Runnable r, V result, long ns) {
    super(r, result);
    this.time = ns;
    this.period = 0;
    this.sequenceNumber = sequencer.getAndIncrement();
}

/**
 * Creates a periodic action with given nano time and period.
 */
ScheduledFutureTask(Runnable r, V result, long ns, long period) {
    super(r, result);
    this.time = ns;
    this.period = period;
    this.sequenceNumber = sequencer.getAndIncrement();
}

/**
 * Creates a one-shot action with given nanoTime-based trigger.
 */
ScheduledFutureTask(Callable<V> callable, long ns) {
    super(callable);
    this.time = ns;
    this.period = 0;
    this.sequenceNumber = sequencer.getAndIncrement();
}

这里面有几个重要的属性，下面来解释一下：

time：下次任务执行的时间；
period：执行周期；
sequenceNumber：保存任务被添加到ScheduledThreadPoolExecutor中的序号。

在schedule方法中，创建完ScheduledFutureTask对象之后，会执行delayedExecute方法来执行任务。

delayedExecute方法

private void delayedExecute(Runnable command) {
    if (isShutdown()) {
        reject(command);
        return;
    }
    // Prestart a thread if necessary. We cannot prestart it
    // running the task because the task (probably) shouldn't be
    // run yet, so thread will just idle until delay elapses.
    if (getPoolSize() < getCorePoolSize())
        prestartCoreThread();

    super.getQueue().add(command);
}

这里的关键点其实就是super.getQueue().add(command)行代码，该行代码把任务添加到DelayedWorkQueue延时队列中。

ScheduledFutureTask的run方法

通过上面的逻辑，我们把提交的任务成功加入到了延迟队列中。回顾一下线程池的执行过程：当线程池中的工作线程启动时，不断地从阻塞队列中取出任务并执行，当然，取出的任务实现了Runnable接口，所以是通过调用任务的run方法来执行任务的。

这里的任务类型是ScheduledFutureTask，所以下面看一下ScheduledFutureTask的run方法：

public void run() {
    if (isPeriodic())
        runPeriodic();
    else
        ScheduledFutureTask.super.run();
}

private void runPeriodic() {
    boolean ok = ScheduledFutureTask.super.runAndReset();
    boolean down = isShutdown();
    // Reschedule if not cancelled and not shutdown or policy allows
    if (ok && (!down ||
               (getContinueExistingPeriodicTasksAfterShutdownPolicy() &&
                !isTerminating()))) {
        long p = period;
        if (p > 0)
            time += p;
        else
            time = now() - p;
        ScheduledThreadPoolExecutor.super.getQueue().add(this);
    }
    // This might have been the final executed delayed
    // task.  Wake up threads to check.
    else if (down)
        interruptIdleWorkers();
}

如果是周期性任务，调用FutureTask中的runAndReset方法执行任务，runAndReset方法不会设置执行结果，所以可以重复执行任务，任务执行完成后设置任务的下次执行时间，最后把该任务添加到延迟队列中；如果不是周期性任务，则直接调用FutureTask中的run方法执行。

至此，已经把任务从创建、提交、执行的流程说完了，下面看一下延迟队列DelayedWorkQueue。

DelayedWorkQueue

DelayedWorkQueue底层依赖DelayQueue，而DelayQueue底层又依赖PriorityQueue，所以DelayedWorkQueue是一个基于堆的数据结构。在执行定时任务的时候，每个任务的执行时间都不同，所以DelayedWorkQueue的工作就是按照执行时间的升序来排列，执行时间距离当前时间越近的任务在队列的前面，并且每次出队时能够保证取出的任务是当前队列中下次执行时间最小的任务。

由于DelayedWorkQueue底层完全依赖DelayQueue，所以可以参考上篇文章关于DelayQueue的分析。

33-ScheduledThreadPoolExecutor源码分析

ScheduledThreadPoolExecutor源码分析

源码深入分析

ScheduledThreadPoolExecutor的构造方法

schedule方法

scheduleAtFixedRate方法

scheduleWithFixedDelay方法

scheduleAtFixedRate 和 scheduleWithFixedDelay的区别

ScheduledFutureTask

ScheduledFutureTask的构造方法

delayedExecute方法

ScheduledFutureTask的run方法

DelayedWorkQueue

JDK源码深入学习之ArrayList

C#动态生成DropDownList执行失败原因分析

C#分析URL参数并获取参数和值对应列表的方法

北京网站建设公司分析2018年流行的营销型网站趋势

详细分析Java并发集合ArrayBlockingQueue的用法

详细分析Java并发集合LinkedBlockingQueue的用法

Java中子类调用父类构造方法的问题分析

深入分析Cookie的安全性问题

Java的正则表达式深入分析

浅谈vue.use()方法从源码到使用