首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >JDK源码分析-ScheduledThreadPoolExecutor

JDK源码分析-ScheduledThreadPoolExecutor

作者头像
WriteOnRead
发布2019-10-28 17:48:19
3550
发布2019-10-28 17:48:19
举报
文章被收录于专栏:WriteOnReadWriteOnRead

概述

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

DelayedWorkQueue 类与前文分析的 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,该方法用于判断一个任务是否是周期性执行的。它继承的 RunnableFuture 接口在前文 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<V> 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<V> 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<Void>(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 <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)));
    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 对象,其他的执行流程几乎一样。

scheduleWithFixedDelay 方法:

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 方法(使用延迟队列实现)。

相关阅读:

JDK源码分析-ThreadPoolExecutor

JDK源码分析-ScheduledExecutorService

JDK源码分析-DelayQueue

JDK源码分析-FutureTask

本文参与 腾讯云自媒体分享计划,分享自微信公众号。
原始发表:2019-10-28,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 WriteOnRead 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
相关产品与服务
腾讯云代码分析
腾讯云代码分析(内部代号CodeDog)是集众多代码分析工具的云原生、分布式、高性能的代码综合分析跟踪管理平台,其主要功能是持续跟踪分析代码,观测项目代码质量,支撑团队传承代码文化。
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档