高并发之——深度解析ScheduledThreadPoolExecutor类的源代码

简介: 在【高并发专题】的专栏中,我们深度分析了ThreadPoolExecutor类的源代码,而ScheduledThreadPoolExecutor类是ThreadPoolExecutor类的子类。今天我们就来一起手撕ScheduledThreadPoolExecutor类的源代码。

在【高并发专题】的专栏中,我们深度分析了ThreadPoolExecutor类的源代码,而ScheduledThreadPoolExecutor类是ThreadPoolExecutor类的子类。今天我们就来一起手撕ScheduledThreadPoolExecutor类的源代码。

构造方法

我们先来看下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);
}

从代码结构上来看,ScheduledThreadPoolExecutor类是ThreadPoolExecutor类的子类,ScheduledThreadPoolExecutor类的构造方法实际上调用的是ThreadPoolExecutor类的构造方法。

schedule方法

接下来,我们看一下ScheduledThreadPoolExecutor类的schedule方法,源代码如下所示。

public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
    //如果传递的Runnable对象和TimeUnit时间单位为空
    //抛出空指针异常
    if (command == null || unit == null)
        throw new NullPointerException();
    //封装任务对象,在decorateTask方法中直接返回ScheduledFutureTask对象
    RunnableScheduledFuture<?> t = decorateTask(command, new ScheduledFutureTask<Void>(command, null, triggerTime(delay, unit)));
    //执行延时任务
    delayedExecute(t);
    //返回任务
    return t;
}
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) 
    //如果传递的Callable对象和TimeUnit时间单位为空
    //抛出空指针异常
    if (callable == null || unit == null)
        throw new NullPointerException();
    //封装任务对象,在decorateTask方法中直接返回ScheduledFutureTask对象
    RunnableScheduledFuture<V> t = decorateTask(callable,
        new ScheduledFutureTask<V>(callable, triggerTime(delay, unit)));
    //执行延时任务
    delayedExecute(t);
    //返回任务
    return t;
}

从源代码可以看出,ScheduledThreadPoolExecutor类提供了两个重载的schedule方法,两个schedule方法的第一个参数不同。可以传递Runnable接口对象,也可以传递Callable接口对象。在方法内部,会将Runnable接口对象和Callable接口对象封装成RunnableScheduledFuture对象,本质上就是封装成ScheduledFutureTask对象。并通过delayedExecute方法来执行延时任务。

在源代码中,我们看到两个schedule都调用了decorateTask方法,接下来,我们就看看decorateTask方法。

decorateTask方法

decorateTask方法源代码如下所示。

protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
    return task;
}
protected <V> RunnableScheduledFuture<V> decorateTask(Callable<V> callable, RunnableScheduledFuture<V> task) {
    return task;
}

通过源码可以看出decorateTask方法的实现比较简单,接收一个Runnable接口对象或者Callable接口对象和封装的RunnableScheduledFuture任务,两个方法都是将RunnableScheduledFuture任务直接返回。在ScheduledThreadPoolExecutor类的子类中可以重写这两个方法。

接下来,我们继续看下scheduleAtFixedRate方法。

scheduleAtFixedRate方法

scheduleAtFixedRate方法源代码如下所示。

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
    //传入的Runnable对象和TimeUnit为空,则抛出空指针异常
    if (command == null || unit == null)
        throw new NullPointerException();
    //如果执行周期period传入的数值小于或者等于0
    //抛出非法参数异常
    if (period <= 0)
        throw new IllegalArgumentException();
    //将Runnable对象封装成ScheduledFutureTask任务,
    //并设置执行周期
    ScheduledFutureTask<Void> sft =
        new ScheduledFutureTask<Void>(command, null, triggerTime(initialDelay, unit), unit.toNanos(period));
    //调用decorateTask方法,本质上还是直接返回ScheduledFutureTask对象
    RunnableScheduledFuture<Void> t = decorateTask(command, sft);
    //设置执行的任务
    sft.outerTask = t;
    //执行延时任务
    delayedExecute(t);
    //返回执行的任务
    return t;
}

通过源码可以看出,scheduleAtFixedRate方法将传递的Runnable对象封装成ScheduledFutureTask任务对象,并设置了执行周期,下一次的执行时间相对于上一次的执行时间来说,加上了period时长,时长的具体单位由TimeUnit决定。采用固定的频率来执行定时任务。

ScheduledThreadPoolExecutor类中另一个定时调度任务的方法是scheduleWithFixedDelay方法,接下来,我们就一起看看scheduleWithFixedDelay方法。

scheduleWithFixedDelay方法

scheduleWithFixedDelay方法的源代码如下所示。

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
    //传入的Runnable对象和TimeUnit为空,则抛出空指针异常
    if (command == null || unit == null)
        throw new NullPointerException();
    //任务延时时长小于或者等于0,则抛出非法参数异常
    if (delay <= 0)
        throw new IllegalArgumentException();
    //将Runnable对象封装成ScheduledFutureTask任务
    //并设置固定的执行周期来执行任务
    ScheduledFutureTask<Void> sft =
        new ScheduledFutureTask<Void>(command, null,triggerTime(initialDelay, unit), unit.toNanos(-delay));
    //调用decorateTask方法,本质上直接返回ScheduledFutureTask任务
    RunnableScheduledFuture<Void> t = decorateTask(command, sft);
    //设置执行的任务
    sft.outerTask = t;
    //执行延时任务
    delayedExecute(t);
    //返回任务
    return t;
}

从scheduleWithFixedDelay方法的源代码,我们可以看出在将Runnable对象封装成ScheduledFutureTask时,设置了执行周期,但是此时设置的执行周期与scheduleAtFixedRate方法设置的执行周期不同。此时设置的执行周期规则为:下一次任务执行的时间是上一次任务完成的时间加上delay时长,时长单位由TimeUnit决定。也就是说,具体的执行时间不是固定的,但是执行的周期是固定的,整体采用的是相对固定的延迟来执行定时任务。

如果大家细心的话,会发现在scheduleWithFixedDelay方法中设置执行周期时,传递的delay值为负数,如下所示。

ScheduledFutureTask<Void> sft =
        new ScheduledFutureTask<Void>(command, null, triggerTime(initialDelay, unit), unit.toNanos(-delay));

这里的负数表示的是相对固定的延迟。

在ScheduledFutureTask类中,存在一个setNextRunTime方法,这个方法会在run方法执行完任务后调用,这个方法更能体现scheduleAtFixedRate方法和scheduleWithFixedDelay方法的不同,setNextRunTime方法的源码如下所示

private void setNextRunTime() {
    //距离下次执行任务的时长
    long p = period;
    //固定频率执行,
    //上次执行任务的时间
    //加上任务的执行周期
    if (p > 0)
        time += p;
    //相对固定的延迟
    //使用的是系统当前时间
    //加上任务的执行周期
    else
        time = triggerTime(-p);
}

在setNextRunTime方法中通过对下次执行任务的时长进行判断来确定是固定频率执行还是相对固定的延迟。

triggerTime方法

在ScheduledThreadPoolExecutor类中提供了两个triggerTime方法,用于获取下一次执行任务的具体时间。triggerTime方法的源码如下所示。

private long triggerTime(long delay, TimeUnit unit) {
    return triggerTime(unit.toNanos((delay < 0) ? 0 : delay));
}
long triggerTime(long delay) {
    return now() +
        ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
}

这两个triggerTime方法的代码比较简单,就是获取下一次执行任务的具体时间。有一点需要注意的是:delay < (Long.MAX_VALUE >> 1判断delay的值是否小于Long.MAX_VALUE的一半,如果小于Long.MAX_VALUE值的一半,则直接返回delay,否则需要处理溢出的情况。

我们看到在triggerTime方法中处理防止溢出的逻辑使用了overflowFree方法,接下来,我们就看看overflowFree方法的实现。

overflowFree方法

overflowFree方法的源代码如下所示。

private long overflowFree(long delay) {
    //获取队列中的节点
    Delayed head = (Delayed) super.getQueue().peek();
    //获取的节点不为空,则进行后续处理
    if (head != null) {
        //从队列节点中获取延迟时间
        long headDelay = head.getDelay(NANOSECONDS);
        //如果从队列中获取的延迟时间小于0,并且传递的delay
        //值减去从队列节点中获取延迟时间小于0
        if (headDelay < 0 && (delay - headDelay < 0))
            //将delay的值设置为Long.MAX_VALUE + headDelay
            delay = Long.MAX_VALUE + headDelay;
    }
    //返回延迟时间
    return delay;
}

通过对overflowFree方法的源码分析,可以看出overflowFree方法本质上就是为了限制队列中的所有节点的延迟时间在Long.MAX_VALUE值之内,防止在ScheduledFutureTask类中的compareTo方法中溢出。

ScheduledFutureTask类中的compareTo方法的源码如下所示。

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

compareTo方法的主要作用就是对各延迟任务进行排序,距离下次执行时间靠前的任务就排在前面。

delayedExecute方法

delayedExecute方法是ScheduledThreadPoolExecutor类中延迟执行任务的方法,源代码如下所示。

private void delayedExecute(RunnableScheduledFuture<?> task) {
    //如果当前线程池已经关闭
    //则执行线程池的拒绝策略
    if (isShutdown())
        reject(task);
    //线程池没有关闭
    else {
        //将任务添加到阻塞队列中
        super.getQueue().add(task);
        //如果当前线程池是SHUTDOWN状态
        //并且当前线程池状态下不能执行任务
        //并且成功从阻塞队列中移除任务
        if (isShutdown() &&
            !canRunInCurrentRunState(task.isPeriodic()) &&
            remove(task))
            //取消任务的执行,但不会中断执行中的任务
            task.cancel(false);
        else
            //调用ThreadPoolExecutor类中的ensurePrestart()方法
            ensurePrestart();
    }
}

可以看到在delayedExecute方法内部调用了canRunInCurrentRunState方法,canRunInCurrentRunState方法的源码实现如下所示。

boolean canRunInCurrentRunState(boolean periodic) {
    return isRunningOrShutdown(periodic ? continueExistingPeriodicTasksAfterShutdown : executeExistingDelayedTasksAfterShutdown);
}

可以看到canRunInCurrentRunState方法的逻辑比较简单,就是判断线程池当前状态下能够执行任务。

另外,在delayedExecute方法内部还调用了ThreadPoolExecutor类中的ensurePrestart()方法,接下来,我们看下ThreadPoolExecutor类中的ensurePrestart()方法的实现,如下所示。

void ensurePrestart() {
    int wc = workerCountOf(ctl.get());
    if (wc < corePoolSize)
        addWorker(null, true);
    else if (wc == 0)
        addWorker(null, false);
}

在ThreadPoolExecutor类中的ensurePrestart()方法中,首先获取当前线程池中线程的数量,如果线程数量小于corePoolSize则调用addWorker方法传递null和true,如果线程数量为0,则调用addWorker方法传递null和false。

关于addWork()方法的源码解析,大家可以参考【高并发专题】中的《高并发之——通过ThreadPoolExecutor类的源码深度解析线程池执行任务的核心流程》一文,这里,不再赘述。

reExecutePeriodic方法

reExecutePeriodic方法的源代码如下所示。

void reExecutePeriodic(RunnableScheduledFuture<?> task) {
    //线程池当前状态下能够执行任务
    if (canRunInCurrentRunState(true)) {
        //将任务放入队列
        super.getQueue().add(task);
        //线程池当前状态下不能执行任务,并且成功移除任务
        if (!canRunInCurrentRunState(true) && remove(task))
            //取消任务
            task.cancel(false);
        else
            //调用ThreadPoolExecutor类的ensurePrestart()方法
            ensurePrestart();
    }
}

总体来说reExecutePeriodic方法的逻辑比较简单,但是,这里需要注意和delayedExecute方法的不同点:调用reExecutePeriodic方法的时候已经执行过一次任务,所以,并不会触发线程池的拒绝策略;传入reExecutePeriodic方法的任务一定是周期性的任务。

onShutdown方法

onShutdown方法是ThreadPoolExecutor类中的钩子函数,它是在ThreadPoolExecutor类中的shutdown方法中调用的,而在ThreadPoolExecutor类中的onShutdown方法是一个空方法,如下所示。

void onShutdown() {
}

ThreadPoolExecutor类中的onShutdown方法交由子类实现,所以ScheduledThreadPoolExecutor类覆写了onShutdown方法,实现了具体的逻辑,ScheduledThreadPoolExecutor类中的onShutdown方法的源码实现如下所示。

@Override
void onShutdown() {
    //获取队列
    BlockingQueue<Runnable> q = super.getQueue();
    //在线程池已经调用shutdown方法后,是否继续执行现有延迟任务
    boolean keepDelayed = getExecuteExistingDelayedTasksAfterShutdownPolicy();
    //在线程池已经调用shutdown方法后,是否继续执行现有定时任务
    boolean keepPeriodic = getContinueExistingPeriodicTasksAfterShutdownPolicy();
    //在线程池已经调用shutdown方法后,不继续执行现有延迟任务和定时任务
    if (!keepDelayed && !keepPeriodic) {
        //遍历队列中的所有任务
        for (Object e : q.toArray())
            //取消任务的执行
            if (e instanceof RunnableScheduledFuture<?>)
                ((RunnableScheduledFuture<?>) e).cancel(false);
        //清空队列
        q.clear();
    }
    //在线程池已经调用shutdown方法后,继续执行现有延迟任务和定时任务
    else {
        //遍历队列中的所有任务
        for (Object e : q.toArray()) {
            //当前任务是RunnableScheduledFuture类型
            if (e instanceof RunnableScheduledFuture) {
                //将任务强转为RunnableScheduledFuture类型
                RunnableScheduledFuture<?> t = (RunnableScheduledFuture<?>)e;
                //在线程池调用shutdown方法后不继续的延迟任务或周期任务
                //则从队列中删除并取消任务
                if ((t.isPeriodic() ? !keepPeriodic : !keepDelayed) ||
                    t.isCancelled()) {
                    if (q.remove(t))
                        t.cancel(false);
                }
            }
        }
    }
    //最终调用tryTerminate()方法
    tryTerminate();
}

ScheduledThreadPoolExecutor类中的onShutdown方法的主要逻辑就是先判断线程池调用shutdown方法后,是否继续执行现有的延迟任务和定时任务,如果不再执行,则取消任务并清空队列;如果继续执行,将队列中的任务强转为RunnableScheduledFuture对象之后,从队列中删除并取消任务。大家需要好好理解这两种处理方式。最后调用ThreadPoolExecutor类的tryTerminate方法。有关ThreadPoolExecutor类的tryTerminate方法的源码解析,大家可以参考【高并发专题】中的《高并发之——通过源码深度分析线程池中Worker线程的执行流程》一文,这里不再赘述。

至此,ScheduledThreadPoolExecutor类中的核心方法的源代码,我们就分析完了。

相关文章
|
7月前
|
Java 开发者
重学Java基础篇—Java类加载顺序深度解析
本文全面解析Java类的生命周期与加载顺序,涵盖从加载到卸载的七个阶段,并深入探讨初始化阶段的执行规则。通过单类、继承体系的实例分析,明确静态与实例初始化的顺序。同时,列举六种触发初始化的场景及特殊场景处理(如接口初始化)。提供类加载完整流程图与记忆口诀,助于理解复杂初始化逻辑。此外,针对空指针异常等问题提出排查方案,并给出最佳实践建议,帮助开发者优化程序设计、定位BUG及理解框架机制。最后扩展讲解类加载器层次与双亲委派机制,为深入研究奠定基础。
240 0
|
8月前
|
编译器 C++ 开发者
【C++篇】深度解析类与对象(下)
在上一篇博客中,我们学习了C++的基础类与对象概念,包括类的定义、对象的使用和构造函数的作用。在这一篇,我们将深入探讨C++类的一些重要特性,如构造函数的高级用法、类型转换、static成员、友元、内部类、匿名对象,以及对象拷贝优化等。这些内容可以帮助你更好地理解和应用面向对象编程的核心理念,提升代码的健壮性、灵活性和可维护性。
|
10月前
|
数据可视化 数据挖掘 BI
团队管理者必读:高效看板类协同软件的功能解析
在现代职场中,团队协作的效率直接影响项目成败。看板类协同软件通过可视化界面,帮助团队清晰规划任务、追踪进度,提高协作效率。本文介绍看板类软件的优势,并推荐五款优质工具:板栗看板、Trello、Monday.com、ClickUp 和 Asana,助力团队实现高效管理。
195 2
|
7月前
|
存储 监控 安全
重学Java基础篇—类的生命周期深度解析
本文全面解析了Java类的生命周期,涵盖加载、验证、准备、解析、初始化、使用及卸载七个关键阶段。通过分阶段执行机制详解(如加载阶段的触发条件与技术实现),结合方法调用机制、内存回收保护等使用阶段特性,以及卸载条件和特殊场景处理,帮助开发者深入理解JVM运作原理。同时,文章探讨了性能优化建议、典型异常处理及新一代JVM特性(如元空间与模块化系统)。总结中强调安全优先、延迟加载与动态扩展的设计思想,并提供开发建议与进阶方向,助力解决性能调优、内存泄漏排查及框架设计等问题。
270 5
|
7月前
|
安全 IDE Java
重学Java基础篇—Java Object类常用方法深度解析
Java中,Object类作为所有类的超类,提供了多个核心方法以支持对象的基本行为。其中,`toString()`用于对象的字符串表示,重写时应包含关键信息;`equals()`与`hashCode()`需成对重写,确保对象等价判断的一致性;`getClass()`用于运行时类型识别;`clone()`实现对象复制,需区分浅拷贝与深拷贝;`wait()/notify()`支持线程协作。此外,`finalize()`已过时,建议使用更安全的资源管理方式。合理运用这些方法,并遵循最佳实践,可提升代码质量与健壮性。
194 1
|
6月前
|
人工智能 缓存 NoSQL
高并发秒杀系统设计:关键技术解析与典型陷阱规避
在电商、在线票务等场景中,高并发秒杀活动对系统性能和稳定性提出极大挑战。海量请求可能导致服务器资源耗尽、数据库锁争用及库存超卖等问题。通过飞算JavaAI生成的Redis + Lua分布式锁代码,可有效解决高并发下的锁问题,提升系统QPS达70%,同时避免缓存击穿与库存超卖。相较传统写法,AI优化代码显著提高性能与响应速度,为高并发系统开发提供高效解决方案。
|
7月前
|
传感器 监控 Java
Java代码结构解析:类、方法、主函数(1分钟解剖室)
### Java代码结构简介 掌握Java代码结构如同拥有程序世界的建筑蓝图,类、方法和主函数构成“黄金三角”。类是独立的容器,承载成员变量和方法;方法实现特定功能,参数控制输入环境;主函数是程序入口。常见错误包括类名与文件名不匹配、忘记static修饰符和花括号未闭合。通过实战案例学习电商系统、游戏角色控制和物联网设备监控,理解类的作用、方法类型和主函数任务,避免典型错误,逐步提升编程能力。 **脑图速记法**:类如太空站,方法即舱段;main是发射台,static不能换;文件名对仗,括号要成双;参数是坐标,void不返航。
256 5
|
8月前
|
Java 数据库 开发者
详细介绍SpringBoot启动流程及配置类解析原理
通过对 Spring Boot 启动流程及配置类解析原理的深入分析,我们可以看到 Spring Boot 在启动时的灵活性和可扩展性。理解这些机制不仅有助于开发者更好地使用 Spring Boot 进行应用开发,还能够在面对问题时,迅速定位和解决问题。希望本文能为您在 Spring Boot 开发过程中提供有效的指导和帮助。
817 12
|
7月前
|
机器学习/深度学习 人工智能 监控
鸿蒙赋能智慧物流:AI类目标签技术深度解析与实践
在数字化浪潮下,物流行业面临变革,传统模式的局限性凸显。AI技术为物流转型升级注入动力。本文聚焦HarmonyOS NEXT API 12及以上版本,探讨如何利用AI类目标签技术提升智慧物流效率、准确性和成本控制。通过高效数据处理、实时监控和动态调整,AI技术显著优于传统方式。鸿蒙系统的分布式软总线技术和隐私保护机制为智慧物流提供了坚实基础。从仓储管理到运输监控再到配送优化,AI类目标签技术助力物流全流程智能化,提高客户满意度并降低成本。开发者可借助深度学习框架和鸿蒙系统特性,开发创新应用,推动物流行业智能化升级。
204 1
|
8月前
|
安全 编译器 C语言
【C++篇】深度解析类与对象(中)
在上一篇博客中,我们学习了C++类与对象的基础内容。这一次,我们将深入探讨C++类的关键特性,包括构造函数、析构函数、拷贝构造函数、赋值运算符重载、以及取地址运算符的重载。这些内容是理解面向对象编程的关键,也帮助我们更好地掌握C++内存管理的细节和编码的高级技巧。

热门文章

最新文章

推荐镜像

更多
  • DNS