热门标签 | HotTags
当前位置:  开发笔记 > 编程语言 > 正文

深入解析ScheduledThreadPoolExecutor并发执行机制源代码

本文深入剖析了ScheduledThreadPoolExecutor的并发执行机制及其源代码,详细解读了该线程池如何在指定延时或定期执行任务,探讨了其内部的工作原理和优化策略,为开发者提供了宝贵的参考和实践指导。

ScheduledThreadPoolExecutor是一个可以在指定一定延时时间后或者定时进行任务调度的线程池,ScheduledThreadPoolExecutor继承了ThreadPoolExecutor并实现了ScheduledExecutorService接口。线程池的队列是DelayedWorkQueue(他是ScheduledThreadPoolExecutor的一个内部类)。
在这里插入图片描述

还要在看一下ScheduledFutureTask(同样是ScheduledThreadPoolExecutor的一个内部类),继承FutureTask,FutureTask的内部有一个变量state用来表示任务的状态,一开始状态为NEW。下面是所有状态定义。

private volatile int state;private static final int NEW = 0;//初始化状态private static final int COMPLETING = 1;//执行中状态private static final int NORMAL = 2;//正常运行结束状态private static final int EXCEPTIONAL = 3;//运行中异常private static final int CANCELLED = 4;//任务被取消private static final int INTERRUPTING = 5;//任务正在被中段private static final int INTERRUPTED = 6;//任务已经被中断

ScheduledFutureTask内部还有一个变量period用来表示任务的类型,如果==0,则表示任务是一次性的,任务执行完毕后退出,如果为负数,说明当前任务是以固定延时的定时可重复执行任务,如果为正数,说明任务是以固定频率的定时定时可重复执行任务。

一、chedule(Runnable command, long delay, TimeUnit unit)方法解析

他的作用是提交一个延时执行的任务,任务从提交时间算起延时单位为unit的delay时间后开始执行,任务只会执行一次。

public ScheduledFuture<?> schedule(Runnable command,long delay,TimeUnit unit) {//判断参数是否为空&#xff0c;空则跑出异常if (command &#61;&#61; null || unit &#61;&#61; null)throw new NullPointerException();//任务转换RunnableScheduledFuture<?> t &#61; decorateTask(command,new ScheduledFutureTask<Void>(command, null,triggerTime(delay, unit)));//添加任务到延时队列delayedExecute(t);return t;}

首先是参数判断&#xff0c;为空则抛出异常&#xff0c;接着是装饰任务&#xff0c;把提交的command(Runnable对象)转换为ScheduledFutureTask&#xff0c;ScheduledFutureTask是具体放入延时队列里面的东西&#xff0c;由于是延时任务&#xff0c;所以ScheduledFutureTask实现了getDelay和compareTo方法&#xff0c;triggerTime方法将延时时间转换为绝对时间&#xff0c;也就是把当前时间的纳秒加上延迟的纳秒后的值。

ScheduledFutureTask的构造如下&#xff0c;设period为0&#xff0c;表示该任务是一次性任务。

ScheduledFutureTask(Callable<V> callable, long ns) {super(callable);this.time &#61; ns;this.period &#61; 0;this.sequenceNumber &#61; sequencer.getAndIncrement();}

然后通过delayedExecute将任务添加到延时队列。

private void delayedExecute(RunnableScheduledFuture<?> task) {//如果线程池管理&#xff0c;则执行线程拒绝策略if (isShutdown())reject(task);else {//添加任务到延时队列super.getQueue().add(task);//再次检查线程池状态if (isShutdown() &&!canRunInCurrentRunState(task.isPeriodic()) &&remove(task))task.cancel(false);else//确保至少一个线程在处理任务ensurePrestart();}}

上述代码首先确保线程池没关闭&#xff0c;关闭则执行拒绝策略&#xff0c;没关闭将任务添加到延时队列&#xff0c;添加后再重新检查线程池是否关闭&#xff0c;如果关闭则从延时队列里面删除刚才添加的任务。

再看ensurePrestart方法。

void ensurePrestart() {int wc &#61; workerCountOf(ctl.get());//增加核心线程数if (wc < corePoolSize)addWorker(null, true);else if (wc &#61;&#61; 0)addWorker(null, false);}

首先获取到了线程池中的线程数&#xff0c;如果个数小于核心线程池则新增一个线程&#xff0c;否则如果当前线程数为0个&#xff0c;则同样新增一个线程。

我们知道ThreadPoolExecutor在具体执行任务的线程是Worker线程&#xff0c;Worker线程调用具体任务的run方法来执行&#xff0c;在这里的任务是ScheduledFutureTask&#xff0c;所以来看看ScheduledFutureTask的run方法。

public void run() {//是否执行一次boolean periodic &#61; isPeriodic();//取消任务if (!canRunInCurrentRunState(periodic))cancel(false);else if (!periodic)ScheduledFutureTask.super.run();//定时任务else if (ScheduledFutureTask.super.runAndReset()) {setNextRunTime();reExecutePeriodic(outerTask);}
}

首先判断任务是一次性的还是可重复执行的任务&#xff0c;ScheduledFutureTask在构造方法中已经设置了period为0&#xff0c;所以&#xff0c;这里会返回false。

public boolean isPeriodic() {return period !&#61; 0;}

然后判断当前任务是否应该被取消。为true则取消任务。

由于periodic为false&#xff0c;则会执行代码ScheduledFutureTask.super.run();&#xff0c;调用父类FutureTask的run方法。

public void run() {if (state !&#61; NEW ||!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread()))return;try {Callable<V> c &#61; callable;if (c !&#61; null && state &#61;&#61; NEW) {V result;boolean ran;try {result &#61; c.call();ran &#61; true;} catch (Throwable ex) {result &#61; null;ran &#61; false;setException(ex);}if (ran)set(result);}} finally {runner &#61; null;int s &#61; state;if (s >&#61; INTERRUPTING)handlePossibleCancellationInterrupt(s);}}

FutureTask.run()首先判断任务状态&#xff0c;如果不是NEW则直接返回&#xff0c;或者如果任务状态为NEW&#xff0c;但是使用CAS设置当前任务的持有者为当前线程失败则直接返回。

然后具体调用callable的call方法执行任务&#xff0c;如果任务执行成功则修改任务状态&#xff0c;也就是set方法。

protected void set(V v) {if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {outcome &#61; v;UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final statefinishCompletion();}
}

使用CAS将当前任务的状态从NEW转换的COMPLETING。这里当有多个线程调用时只有一个线程会成功&#xff0c;成功的线程在通过 UNSAFE.putOrderedInt设置任务的状态为正常结束状态。

还有在任务执行失败后&#xff0c;执行setException方法&#xff0c;和set方法类似了。

protected void setException(Throwable t) {if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {outcome &#61; t;UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final statefinishCompletion();}}

二、scheduleWithFixedDelay(Runnable command,long initialDelay, long delay,TimeUnit unit)方法解析

他的作用是&#xff0c;当任务执行完毕后&#xff0c;让其延迟固定时间后再次运行&#xff0c;initialDelay表示提交任务后延迟多少时间开始执行任务command&#xff0c;delay表示当任务执行完毕后延长多少时间后再次运行command&#xff0c;unit是时间单位。

这个任务会一直重复运行下去&#xff0c;直到任务中抛出异常、被取消、线程池关闭。

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit) {//参数判断if (command &#61;&#61; null || unit &#61;&#61; null)throw new NullPointerException();if (delay <&#61; 0)throw new IllegalArgumentException();//任务转换ScheduledFutureTask<Void> sft &#61;new ScheduledFutureTask<Void>(command,null,triggerTime(initialDelay, unit),unit.toNanos(-delay));RunnableScheduledFuture<Void> t &#61; decorateTask(command, sft);sft.outerTask &#61; t;//添加任务到队列delayedExecute(t);return t;}

首先也是参数判断&#xff0c;为空则抛出异常&#xff0c;然后将command任务转换为ScheduledFutureTask&#xff0c;然后添加延迟到队列。

将任务添加到队列后线程池线程会从队列中获取任务&#xff0c;然后调用ScheduledFutureTask的run方法&#xff0c;由于这里period<0&#xff0c;所以isPeriodic返回true&#xff0c;则会执行方法runAndReset()。

protected boolean runAndReset() {if (state !&#61; NEW ||!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread()))return false;boolean ran &#61; false;int s &#61; state;try {Callable<V> c &#61; callable;if (c !&#61; null && s &#61;&#61; NEW) {try {c.call(); ran &#61; true;} catch (Throwable ex) {setException(ex);}}} finally {runner &#61; null;s &#61; state;if (s >&#61; INTERRUPTING)handlePossibleCancellationInterrupt(s);}return ran && s &#61;&#61; NEW;
}

他在任务执行完毕后不会设置任务的状态&#xff0c;是为了让任务可重复执行&#xff0c;看最后一句&#xff0c;判断如果当前任务正常执行完毕并且任务状态为NEW则返回true&#xff0c;如果返回true则执行方法setNextRunTime()&#xff0c;用于设置任务下一次的执行时间。

这里p是<0的&#xff0c;然后设置timer为当前时间加上-p&#xff0c;也就是延迟-p时间后再次执行。

private void setNextRunTime() {long p &#61; period;if (p > 0)time &#43;&#61; p;elsetime &#61; triggerTime(-p);}

三、scheduleAtFixedRate(Runnable command, long initialDelay, long period,TimeUnit unit)方法解析

该方法相对起始时间点以固定频率调用指定任务&#xff0c;当把任务提交到线程池并延迟initialDelay时间后开始执行任务command&#xff0c;然后从initialDelay&#43;period时间点再次执行&#xff0c;而后在initialDelay&#43;2*period时间点再次执行&#xff0c;直到抛出异常或者取消、关闭线程池。

原理和scheduleWithFixedDelay类似&#xff0c;我们就看几个不同点&#xff0c;

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit) {if (command &#61;&#61; null || unit &#61;&#61; null)throw new NullPointerException();if (period <&#61; 0)throw new IllegalArgumentException();ScheduledFutureTask<Void> sft &#61;new ScheduledFutureTask<Void>(command,null,triggerTime(initialDelay, unit),unit.toNanos(period));RunnableScheduledFuture<Void> t &#61; decorateTask(command, sft);sft.outerTask &#61; t;delayedExecute(t);return t;}

首先是period&#61;period&#xff0c;不再是-period。所以当前任务执行完毕后调用setNextRunTime设置任务下次执行的时间是 time &#43;&#61; p。

private void setNextRunTime() {long p &#61; period;if (p > 0)time &#43;&#61; p;elsetime &#61; triggerTime(-p);}

如果当前任务还没有执行完&#xff0c;下一次执行任务的时间到了&#xff0c;则不会并发执行&#xff0c;下次要执行的任务会延迟&#xff0c;要等到当前任务执行完毕后再次执行


推荐阅读
  • 2019年后蚂蚁集团与拼多多面试经验详述与深度剖析
    2019年后蚂蚁集团与拼多多面试经验详述与深度剖析 ... [详细]
  • Java集合框架特性详解与开发实践笔记
    Java集合框架特性详解与开发实践笔记 ... [详细]
  • 本文深入探讨了CGLIB BeanCopier在Bean对象复制中的应用及其优化技巧。相较于Spring的BeanUtils和Apache的BeanUtils,CGLIB BeanCopier在性能上具有显著优势。通过详细分析其内部机制和使用场景,本文提供了多种优化方法,帮助开发者在实际项目中更高效地利用这一工具。此外,文章还讨论了CGLIB BeanCopier在复杂对象结构和大规模数据处理中的表现,为读者提供了实用的参考和建议。 ... [详细]
  • 优化后的标题:深入解析09版Jedis客户端
    深入解析09版Jedis客户端,本文将详细介绍如何在Java项目中正确配置Jedis以操作Redis。首先,确保项目的JDK版本和编译器设置正确。接着,通过Maven或Gradle导入必要的依赖项,如 `redis.clients:jedis`。此外,文章还将探讨Jedis连接池的配置与优化,以及常见问题的解决方案,帮助开发者高效使用Jedis进行Redis操作。 ... [详细]
  • JDK 1.8引入了多项并发新特性,显著提升了编程效率。本文重点探讨了LongAdder和StampedLock的特性和应用场景。此外,还介绍了在多线程环境中发生死锁时,如何通过jps命令进行诊断和排查,提供了详细的步骤和示例。这些改进不仅增强了系统的性能,还简化了开发者的调试工作。 ... [详细]
  • 本书详细介绍了在最新Linux 4.0内核环境下进行Java与Linux设备驱动开发的全面指南。内容涵盖设备驱动的基本概念、开发环境的搭建、操作系统对设备驱动的影响以及具体开发步骤和技巧。通过丰富的实例和深入的技术解析,帮助读者掌握设备驱动开发的核心技术和最佳实践。 ... [详细]
  • Node.js 教程第五讲:深入解析 EventEmitter(事件监听与发射机制)
    本文将深入探讨 Node.js 中的 EventEmitter 模块,详细介绍其在事件监听与发射机制中的应用。内容涵盖事件驱动的基本概念、如何在 Node.js 中注册和触发自定义事件,以及 EventEmitter 的核心 API 和使用方法。通过本教程,读者将能够全面理解并熟练运用 EventEmitter 进行高效的事件处理。 ... [详细]
  • 如何使用 net.sf.extjwnl.data.Word 类及其代码示例详解 ... [详细]
  • 基于Node.js的高性能实时消息推送系统通过集成Socket.IO和Express框架,实现了高效的高并发消息转发功能。该系统能够支持大量用户同时在线,并确保消息的实时性和可靠性,适用于需要即时通信的应用场景。 ... [详细]
  • 本文探讨了如何在 Google Sheets 中通过自定义函数实现 AJAX 调用。具体介绍了编写脚本的方法,以便在电子表格中发起 AJAX 请求,从而实现数据的动态获取与更新。这种方法不仅简化了数据处理流程,还提高了工作效率。 ... [详细]
  • 在处理高并发场景时,确保业务逻辑的正确性是关键。本文深入探讨了Java原生锁机制的多种细粒度实现方法,旨在通过使用数据的时间戳、ID等关键字段进行锁定,以最小化对系统性能的影响。文章详细分析了不同锁策略的优缺点,并提供了实际应用中的最佳实践,帮助开发者在高并发环境下高效地实现锁机制。 ... [详细]
  • 深入解析 Android TextView 中 getImeActionLabel() 方法的使用与代码示例 ... [详细]
  • 我正在使用 Ruby on Rails 构建个人网站。总体而言,RoR 是一个非常出色的工具,它提供了丰富的功能和灵活性,使得创建自定义页面变得既高效又便捷。通过利用其强大的框架和模块化设计,我可以轻松实现复杂的功能,同时保持代码的整洁和可维护性。此外,Rails 的社区支持也非常强大,为开发过程中遇到的问题提供了丰富的资源和解决方案。 ... [详细]
  • 本文作为“实现简易版Spring系列”的第五篇,继前文深入探讨了Spring框架的核心技术之一——控制反转(IoC)之后,将重点转向另一个关键技术——面向切面编程(AOP)。对于使用Spring框架进行开发的开发者来说,AOP是一个不可或缺的概念。了解AOP的背景及其基本原理,对于掌握这一技术至关重要。本文将通过具体示例,详细解析AOP的实现机制,帮助读者更好地理解和应用这一技术。 ... [详细]
  • 本文深入探讨了Python线程池的内部实现机制,作为对Apshceduler调度器研究的延伸。在先前关于Apshceduler源码分析的文章中,我们提到调度器通过`def_do_submit_`函数将任务提交到线程池。本文将进一步解析线程池的工作原理,包括任务分配、线程管理及性能优化等方面,为读者提供更全面的技术理解。 ... [详细]
author-avatar
航头党员之家
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有