JDK 源码分析：ScheduledThreadPoolExecutor

2013 年 2 月 13 日

概述

ScheduledThreadPoolExecutor 也是一个线程池类，是线程池类 ThreadPoolExecutor 的子类。
除了 ThreadPoolExecutor 相关的方法之外，它还增加了执行定时任务和周期性任务的方法。它的类
签名和继承结构如下：

public class ScheduledThreadPoolExecutor
extends ThreadPoolExecutor
implements ScheduledExecutorService {}

可以看到，它继承了
ThreadPoolExecutor 类（参考「
JDK源码分析-ThreadPoolExecutor
」），并且实现了 ScheduledExecutorService 接口（参考「
JDK源码分析-ScheduledExecutorService
」），因此具有二者的特性。
下面分析其代码实现。

代码分析

内部嵌套类 DelayedWorkQueue

先看它的一个内部嵌套类 DelayedWorkQueue，该类是一个延迟队列，它的类签名和继承结构如下：

static class DelayedWorkQueue extends AbstractQueue<Runnable>

    implements BlockingQueue<Runnable> {}

Delay
edWorkQue

ue 类与前文分析的 DelayQueue 「 JDK源码分析-DelayQueue
」实现原理类似
，这里就不再赘述。

构造器

ScheduledThreadPoolExecutor 有如下四个构造器：

public ScheduledThreadPoolExecutor(int corePoolSize) {

    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,

          new DelayedWorkQueue());

}




public ScheduledThreadPoolExecutor(int corePoolSize,

                                   ThreadFactory threadFactory) {

    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,

          new DelayedWorkQueue(), threadFactory);

}




public ScheduledThreadPoolExecutor(int corePoolSize,

                                   RejectedExecutionHandler handler) {

    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,

          new DelayedWorkQueue(), handler);

}




public ScheduledThreadPoolExecutor(int corePoolSize,

                                   ThreadFactory threadFactory,

                                   RejectedExecutionHandler handler) {

    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,

          new DelayedWorkQueue(), threadFactory, handler);

}

这几个构造器都是直接调用父类 ThreadPoolExecutor 的构造器，只是传入了不同的参数。而其中的参数 workQueue 都传入了上面的延迟队列 DelayedWorkQueue。

内部类 ScheduledFutureTask

ScheduledThreadPoolExecutor 还有一个内部类 ScheduledFutureTask，它的继承结构如下：

它继承了 FutureTask 类（可参考前文「 JDK源码分析-FutureTask
」的分析），且实现了 RunnableScheduledFuture 接口，该接口定义如下：

public interface RunnableScheduledFuture<V> extends RunnableFuture<V>, ScheduledFuture<V> {

    // 一个任务是否周期性执行的，若是则可以重复执行；否则只能运行一次

    boolean isPeriodic();

}

RunnableScheduledFuture 只定义了一个方法 isPeriodic，该方法用于判断一个任务是否是周期性执行的。它继承的
Runna
bleF
utur
e 接口在前文 FutureTask 类中已进行分析，而 ScheduledFuture 接口如
下：

public interface ScheduledFuture<V> extends Delayed, Future<V> {

}

它的内部并未定义方法，只是整合了 Delayed 接口和 Future 接口，Delayed 接口前文也已分析，下面分析该类的主要代码。

先看它的主要成员变量：

// 定时任务执行的时间（单位：纳秒）

private long time;




/**

 * 重复执行的任务的时间间隔（单位：纳秒） 

 * 正数表示固定频率（fixed-rate）执行

 * 负数表示固定延迟（fixed-delay）执行

 * 零表示非重复执行的任务

 */

private final long period;




// reExecutePeriodic 方法中重新排队的任务

RunnableScheduledFuture outerTask = this;




// 延迟队列中的索引位置，便于快速取消

int heapIndex;

构造器：

/**

 * 构造器一：用给定的触发时间（纳秒），创建一个一次性任务

 */

ScheduledFutureTask(Runnable r, V result, long ns) {

    super(r, result);

    this.time = ns;

    this.period = 0;

    this.sequenceNumber = sequencer.getAndIncrement();

}




/**

 * 构造器二：用给定的触发时间和间隔（纳秒），创建一个周期性任务

 */

ScheduledFutureTask(Runnable r, V result, long ns, long period) {

    super(r, result);

    this.time = ns;

    this.period = period;

    this.sequenceNumber = sequencer.getAndIncrement();

}




/**

 * 构造器三：用给定的触发时间（纳秒），创建一个一次性任务

 */

ScheduledFutureTask(Callable callable, long ns) {

    super(callable);

    this.time = ns;

    this.period = 0;

    this.sequenceNumber = sequencer.getAndIncrement();

}

ScheduledFutureTask 有三个构造器，可分为两类：分别是创建一次性任务（一和三）和周期性任务（二）。
其中一和
三还是 Runnable 和 Callable 的区别。

该类是一个任务类，即 Runnable 接口的实现类，因此它最核心的就是 run 方法，如下：

public void run() {

    // 是否为周期性任务

    boolean periodic = isPeriodic();

    // 若任务不能执行，则取消

    if (!canRunInCurrentRunState(periodic))

        cancel(false);

    // 若为非周期性任务

    else if (!periodic)

        // 若为周期性任务，调用 ScheduledFutureTask 的父类（即 FutureTask）的 run 方法执行

        ScheduledFutureTask.super.run();

    // 若为周期性任务，调用 ScheduledFutureTask 的父类（即 FutureTask）的 runAndReset 方法执行

    else if (ScheduledFutureTask.super.runAndReset()) {

        setNextRunTime(); // 设置下一次执行时间

        reExecutePeriodic(outerTask); // 周期性执行

    }

}

reExecutePeriodic 方法如下：

/**

 * 该方法主要是将周期性任务重新排队

 * 它的实现与 delayedExecute 方法（后面分析）逻辑有些类似

 */

void reExecutePeriodic(RunnableScheduledFuture task) {

    if (canRunInCurrentRunState(true)) {

        super.getQueue().add(task);

        if (!canRunInCurrentRunState(true) && remove(task))

            task.cancel(false);

        else

            ensurePrestart();

    }

}

schedule & scheduleAtFixedRate & scheduleWithFixedDelay

这几个就是执行定时任务和周期性任务的方法，它们是对前文「 JDK源码分析-ScheduledExecutorService
」接口所定义的方法实现，可参考前文的分析。

schedule 方法 1：其作用是延迟指定的时间后执行任务（即执行定时任务），只会执行一次。

public ScheduledFuture<?> schedule(Runnable command,

                                   long delay,

                                   TimeUnit unit) {

    if (command == null || unit == null)

        throw new NullPointerException();

    // 把用户提交的 Runnable 对象包装为 RunnableScheduledFuture 对象

    // decorateTask 方法默认返回第二个参数

    // decorateTask 方法的修饰符是 protected，可根据需求自行扩展

    RunnableScheduledFuture<?> t = decorateTask(command,

        new ScheduledFutureTask(command, null,

                                      triggerTime(delay, unit)));

    // 执行给定的任务

    delayedExecute(t);

    return t;

}

delayExecute 方法：

/*

 * 延迟或周期性任务的主要执行方法。

 * 若线程池已关闭，则拒绝该任务（执行拒绝策略）；

 * 否则将任务添加到工作队列，若有需要启动一个线程去执行。

 * 若在添加任务时关闭了线程池，则将其从队列移除并取消该任务

 */

private void delayedExecute(RunnableScheduledFuture task) {

    // 若线程池已关闭，则执行拒绝策略

    if (isShutdown())

        reject(task);

    else {

        // 将该任务添加到任务队列（即前面的延迟队列）

        super.getQueue().add(task);

        // 若当前任务无法执行，则将其从队列移除并且取消执行（类似事务的回滚操作）

        if (isShutdown() &&

            !canRunInCurrentRunState(task.isPeriodic()) &&

            remove(task))

            task.cancel(false);

        // 任务可以执行，若有需要新增线程以执行该任务

        else

            ensurePrestart();

    }

}

schedule 方法 2：

public  ScheduledFuture schedule(Callable callable,

                                       long delay,

                                       TimeUnit unit) {

    if (callable == null || unit == null)

        throw new NullPointerException();

    RunnableScheduledFuture t = decorateTask(callable,

        new ScheduledFutureTask(callable,

                                   triggerTime(delay, unit)));

    delayedExecute(t);

    return t;

}

该方法与前者类似，差别在于这里的参数类型是 Callable，前者是 Runnable 类型，其他操作一样。

scheduleAtFixedRate 方法：

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();
// 将 Runnable 对象包装为 ScheduledFutureTask 对象
ScheduledFutureTask<Void> sft =
new ScheduledFutureTask<Void>(command,
null,
triggerTime(initialDelay, unit),
unit.toNanos(period));
RunnableScheduledFuture<Void> t = decorateTask(command, sft);
sft.outerTask = t;
delayedExecute(t);
return t;
}

该方法与前面的 schedule 方法类似，区别仅在于使用了不同的 ScheduledFutureTask 对象，其他的执行流程几乎一样。

scheduleWithFix
edDelay 方法：

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();
ScheduledFutureTask<Void> sft =
new ScheduledFutureTask<Void>(command,
null,
triggerTime(initialDelay, unit),
unit.toNanos(-delay));
RunnableScheduledFuture<Void> t = decorateTask(command, sft);
sft.outerTask = t;
delayedExecute(t);
return t;
}

该方法与
scheduleAtFixedRate 方法基本一样，区别仅在于构建 ScheduledFutureTask 对象时参数 period 不同（一正一负，用以区分类型）。

execute & submit 方法

这两个方法是 Executor 接口和 ExecutorService 接口所定义的方法，代码实现如下：

public void execute(Runnable command) {
schedule(command, 0, NANOSECONDS);
}

public Future<?> submit(Runnable task) {
return schedule(task, 0, NANOSECONDS);
}

它们内部直接调用了 schedule(Runnable) 方法。
另外两个 submit 方法
：

public <T> Future<T> submit(Runnable task, T result) {
return schedule(Executors.callable(task, result), 0, NANOSECONDS);
}

public <T> Future<T> submit(Callable<T> task) {
return schedule(task, 0, NANOSECONDS);
}

它们内
部直接调用了
schedule(Callable)
方法。

小结

1. ScheduledThreadPoolExecutor 是线程池的实现类之一；

2. 它继承自 ThreadPoolExecutor，并实现了 ScheduledExecutorService 接口；

3. 提供了异步提交任务的 execute 方法和 submit 方法；

4.
提
供了执行定时任务的 schedule 方法和周期性任务的 scheduleAtFixedRate/scheduleWithFixedDelay 方法（使用延迟队列实现）。

M	T	W	T	F	S	S
« Jan
						1
2	3	4	5	6	7	8
9	10	11	12	13	14	15
16	17	18	19	20	21	22
23	24	25	26	27	28	29
30

演道网

JDK 源码分析：ScheduledThreadPoolExecutor

About The Author

hellowo

Related Posts

About The Author

hellowo