Quartz触发Trigger及状态转换(quartz 多个trigger)
ccwgpt 2024-09-18 12:26 28 浏览 0 评论
前言
分布式定时任务框架Quartz的定时任务依赖触发器Trigger来触发执行,那么Trigger如何被触发,在整个触发流程中,Trigger的状态如何变化,本文将对这部分内容进行详细分析。
Quartz框架的基本概念和基本使用可以参考翻翻Quartz框架的旧账。
本文基于Quartz的2.3.2版本展开分析。
正文
先回顾一下QuartzSchedulerThread的作用,其是由QuartzScheduler持有的 调度线程,在QuartzScheduler创建出来并被调用start() 方法后,QuartzSchedulerThread就会开始运行,会不断的去判断哪些Trigger到点需要触发了,需要触发的Trigger就会被从ThreadPool中分配一个线程,然后执行Trigger关联的JobDetail。
具体的整套逻辑,全部在QuartzSchedulerThread的run() 方法中,下面一起来看一下。
(下面方法比较长,分支也比较多,所以重点看有注释的部分,再结合后面的补充说明进行理解)
@Override
public void run() {
int acquiresFailed = 0;
while (!halted.get()) {
try {
synchronized (sigLock) {
while (paused && !halted.get()) {
try {
sigLock.wait(1000L);
} catch (InterruptedException ignore) {
}
acquiresFailed = 0;
}
if (halted.get()) {
break;
}
}
if (acquiresFailed > 1) {
try {
long delay = computeDelayForRepeatedErrors(qsRsrcs.getJobStore(), acquiresFailed);
Thread.sleep(delay);
} catch (Exception ignore) {
}
}
// 从ThreadPool中获取当前可用线程数量
// 若没有可用线程则阻塞直到有可用线程
int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();
if(availThreadCount > 0) {
List<OperableTrigger> triggers;
long now = System.currentTimeMillis();
clearSignaledSchedulingChange();
try {
// 获取下一次触发时间在30分钟内的Trigger
// 这里的步骤可以分解如下
// 1. 从qrtz_locks表获取TRIGGER_ACCESS锁
// 2. 从qrtz_triggers表获取触发时间在30分钟内且状态是WAITING的Trigger
// 3. 遍历每一个步骤2拿到的Trigger
// 4. 从qrtz_job_details表查询出Trigger对应的JobDetail
// 5. 如果JobDetail不允许并发执行则判断一下当前JobDetail是否已经由另外一个Tragger执行
// 若已经由另外一个Trigger执行则当前Trigger本次不执行
// 6. 将确定要执行的Trigger在qrtz_triggers表中的状态设置为ACQUIRED
// 7. 将确定要执行的Trigger插入qrtz_fired_triggers表且状态为ACQUIRED
// 8. 继续遍历步骤2拿到的Trigger直至全部遍历完
// 9. 释放TRIGGER_ACCESS锁
// 10. 返回所有符合条件的Trigger
triggers = qsRsrcs.getJobStore().acquireNextTriggers(
now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());
acquiresFailed = 0;
if (log.isDebugEnabled())
log.debug("batch acquisition of " + (triggers == null ? 0 : triggers.size()) + " triggers");
} catch (JobPersistenceException jpe) {
if (acquiresFailed == 0) {
qs.notifySchedulerListenersError(
"An error occurred while scanning for the next triggers to fire.",
jpe);
}
if (acquiresFailed < Integer.MAX_VALUE)
acquiresFailed++;
continue;
} catch (RuntimeException e) {
if (acquiresFailed == 0) {
getLog().error("quartzSchedulerThreadLoop: RuntimeException "
+e.getMessage(), e);
}
if (acquiresFailed < Integer.MAX_VALUE)
acquiresFailed++;
continue;
}
if (triggers != null && !triggers.isEmpty()) {
now = System.currentTimeMillis();
long triggerTime = triggers.get(0).getNextFireTime().getTime();
long timeUntilTrigger = triggerTime - now;
// 所有Trigger中最先会触发的Trigger的触发时间如果距离当前大于2ms则等待
// 直到最先会触发的Trigger的触发时间距离当前小于2ms
while(timeUntilTrigger > 2) {
synchronized (sigLock) {
if (halted.get()) {
break;
}
if (!isCandidateNewTimeEarlierWithinReason(triggerTime, false)) {
try {
now = System.currentTimeMillis();
timeUntilTrigger = triggerTime - now;
if(timeUntilTrigger >= 1)
sigLock.wait(timeUntilTrigger);
} catch (InterruptedException ignore) {
}
}
}
if(releaseIfScheduleChangedSignificantly(triggers, triggerTime)) {
break;
}
now = System.currentTimeMillis();
timeUntilTrigger = triggerTime - now;
}
if(triggers.isEmpty())
continue;
List<TriggerFiredResult> bndles = new ArrayList<TriggerFiredResult>();
boolean goAhead = true;
synchronized(sigLock) {
goAhead = !halted.get();
}
if(goAhead) {
try {
// 将Trigger进行fire
// 但是这里并不会执行对应的任务逻辑
// 对应的步骤可以分解如下
// 1. 从qrtz_locks表获取TRIGGER_ACCESS锁
// 2. 遍历每一个需要执行的Trigger
// 3. 将qrtz_fired_triggers表中需要执行的Trigger的状态设置为EXECUTING
// 4. 将Trigger对象的下一次执行时间nextFireTime更新
// 5. 如果Trigger执行的JobDetail不允许并发执行
// 将Trigger对象的状态更新为BLOCKED
// 否则将Trigger对象的状态更新为WAITING
// 6. 如果Trigger执行的JobDetail不允许并发执行
// 将JobDetail关联的其它Trigger在qrtz_triggers表里的状态更新
// 如果是WAITING或ACQUIRED则更新为BLOCKED
// 7. 如果Trigger是最后一次执行则将Trigger对象的状态更新为COMPLETE
// 8. 将Trigger对象更新回qrtz_triggers表
// 9. 基于这个Trigger创建一个TriggerFiredResult并添加到集合
// 10. 继续遍历下一个需要执行的Trigger直至遍历完毕
// 11. 释放TRIGGER_ACCESS锁
// 12. 此时得到了所有执行的Trigger对应的TriggerFiredResult的集合
// 上面步骤执行完后所有fire的Trigger在qrtz_fired_triggers表中的状态是EXECUTING
// 在qrtz_triggers表中的状态可能是WAITING,BLOCKED或COMPLETE
// 但是此时Trigger对应的JobDetail实际是还没有被执行的
List<TriggerFiredResult> res = qsRsrcs.getJobStore().triggersFired(triggers);
if(res != null)
bndles = res;
} catch (SchedulerException se) {
qs.notifySchedulerListenersError(
"An error occurred while firing triggers '"
+ triggers + "'", se);
for (int i = 0; i < triggers.size(); i++) {
qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));
}
continue;
}
}
// 遍历每一个执行的Trigger对应的TriggerFiredResult
for (int i = 0; i < bndles.size(); i++) {
TriggerFiredResult result = bndles.get(i);
TriggerFiredBundle bndle = result.getTriggerFiredBundle();
Exception exception = result.getException();
if (exception instanceof RuntimeException) {
getLog().error("RuntimeException while firing trigger " + triggers.get(i), exception);
qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));
continue;
}
if (bndle == null) {
qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));
continue;
}
JobRunShell shell = null;
try {
// 基于TriggerFiredResult创建JobRunShell
shell = qsRsrcs.getJobRunShellFactory().createJobRunShell(bndle);
shell.initialize(qs);
} catch (SchedulerException se) {
qsRsrcs.getJobStore().triggeredJobComplete(triggers.get(i), bndle.getJobDetail(), CompletedExecutionInstruction.SET_ALL_JOB_TRIGGERS_ERROR);
continue;
}
// 在ThreadPool中分配一个线程来执行JobRunShell
// 随后就会在JobRunShell的run()方法中执行JobDetail
// 执行完毕后会再执行Trigger的完成逻辑
// 对应的步骤可以拆分如下
// 1. 从qrtz_locks表获取TRIGGER_ACCESS锁
// 2. 如果Trigger后续不再执行则在qrtz_triggers表里删除这个Trigger
// 3. 如果Trigger执行的任务是不允许并发执行则将所有关联的Trigger状态做如下更新
// 将Trigger状态由BLOCKED恢复成WAITING
// 4. 如果任务类由@PersistJobDataAfterExecution注解修饰则
// 将qrtz_job_details表里的JobDetail的JOB_DATA字段更新
// 5. 删除Trigger在qrtz_fired_triggers表中对应的记录
if (qsRsrcs.getThreadPool().runInThread(shell) == false) {
getLog().error("ThreadPool.runInThread() return false!");
qsRsrcs.getJobStore().triggeredJobComplete(triggers.get(i), bndle.getJobDetail(), CompletedExecutionInstruction.SET_ALL_JOB_TRIGGERS_ERROR);
}
}
continue;
}
} else {
continue;
}
long now = System.currentTimeMillis();
long waitTime = now + getRandomizedIdleWaitTime();
long timeUntilContinue = waitTime - now;
synchronized(sigLock) {
try {
if(!halted.get()) {
if (!isScheduleChanged()) {
sigLock.wait(timeUntilContinue);
}
}
} catch (InterruptedException ignore) {
}
}
} catch(RuntimeException re) {
getLog().error("Runtime error occurred in main trigger firing loop.", re);
}
}
qs = null;
qsRsrcs = null;
}
(下面分小节进行讲解)
1.拿到即将触发的所有Trigger
这里的即将触发,就是触发时间在30分钟内且状态是WAITING的Trigger。
针对每一个即将触发的Trigger,其在qrtz_triggers表里的状态此时会被置为ACQUIRED,针对这个Trigger同时也会插入一条记录到qrtz_fired_triggers表中,状态也是ACQUIRED,表示这个Trigger已经在fire处理了。
假如我们有一个trigger-1,对应任务允许并发执行,还有一个trigger-2,对应任务不允许并发执行(由@DisallowConcurrentExecution注解修饰),并且这两个Trigger的触发时间均在30分钟内。
那么此时在qrtz_triggers表中,它们的状态是这样的。
TRIGGER_NAME | ... | TRIGGER_STATE |
trigger-1 | ... | ACQUIRED |
trigger-2 | ... | ACQUIRED |
在qrtz_fired_triggers表中,它们的状态是这样的。
TRIGGER_NAME | ... | STATE |
trigger-1 | ... | ACQUIRED |
trigger-2 | ... | ACQUIRED |
2.等待最先触发的Trigger的触发时间在2ms内
如果最先触发的Trigger的触发时间距离当前大于2ms,则进行等待,直到小于等于2ms。
3.将Trigger进行fire
fire一个Trigger其实就是将这个Trigger在qrtz_fired_triggers表中记录的状态设置为EXECUTING,后面会为这个Trigger分配线程来执行任务,注意此时Trigger对应的任务实际上是还没有执行的。
Trigger被fire之后,这个Trigger在qrtz_triggers里面的状态及下一次fire的时间会被更新,这里需要关注一下 状态 的更新。
如果Trigger对应的任务没有被@DisallowConcurrentExecution注解修饰,那么这个Trigger的状态更新为WAITING;如果Trigger对应的任务被@DisallowConcurrentExecution注解修饰,那么这个Trigger的状态会更新为BLOCKED,并且还会将这个被@DisallowConcurrentExecution注解修饰的任务所有关联的Trigger的状态更新为BLOCKED。
注意到一个Trigger会被fire,首先就是需要满足触发时间在30分钟内且状态是WAITING,所以如果一个被@DisallowConcurrentExecution注解修饰的任务正在被执行,那么这个任务关联的所有Trigger的状态都应该被置为BLOCKED,以防止这些Trigger再次被fire。
回到第1小节中的例子,此时在qrtz_triggers表中,trigger-1和trigger-2它们的状态是这样的。
TRIGGER_NAME | ... | TRIGGER_STATE |
trigger-1 | ... | WAITING |
trigger-2 | ... | BLOCKED |
在qrtz_fired_triggers表中,它们的状态是这样的。
TRIGGER_NAME | ... | STATE |
trigger-1 | ... | EXECUTING |
trigger-2 | ... | EXECUTING |
4.为fire的Trigger分配线程并执行任务
被fire的Trigger会在qrtz_fired_triggers表中插入一条记录,随后就会被分配一个线程来执行这个Trigger关联的JobDetail。
执行JobDetail没什么好说的,就是调用到这个任务的execute() 方法,我们这里需要关注的是任务执行完毕后的对于Trigger的complete逻辑。
首先会判断当前这个Trigger是不是不会再执行了,如果不会再执行了,那么就会在qrtz_triggers表里删除这个Trigger。
然后就是如果这个Trigger执行的任务是不允许并发执行的,那么此时这个任务关联的所有Trigger的状态肯定都是BLOCKED,所以还需要将这些Trigger的状态由BLOCKED还原为WAITING。
最后就是删除fire的Trigger在qrtz_fired_triggers表中的记录。
回到第1小节中的例子,此时在qrtz_triggers表中,trigger-1和trigger-2它们的状态是这样的。
TRIGGER_NAME | ... | TRIGGER_STATE |
trigger-1 | ... | WAITING |
trigger-2 | ... | WAITING |
在qrtz_fired_triggers表中,它们都没有记录了。
5.暂停一个Trigger
在上面的所有讨论中,都没有提及Trigger的暂停状态,也就是PAUSED状态,因为这个状态相对独立,没必要和上面的其余状态转换混在一起讨论。
我们可以通过如下手段将Trigger的状态置为PAUSED。
- 通过Trigger的group和name找到Trigger,然后将其状态置为PAUSED。这种可以理解为暂停一个Trigger;
- 通过JobDetail找到所有关联的Trigger,然后将这些Trigger的状态置为PAUSED。这种可以理解为暂停一个JobDetail。
当Trigger在qrtz_triggers表中的状态是PAUSED之后,就不再满足触发时间在30分钟内且状态是WAITING,从而Trigger就不会被fire,对应的任务也不会被执行。
总结
阅读完本文后,应该能够回答下面的问题。
1. Trigger的触发流程是怎么样的
首先触发时间在30分钟内且状态是WAITING的Trigger会被获取出来;
其次最先触发的Trigger的触发时间在2s内时就会开始fire这些Trigger;
fire一个Trigger就是将这个Trigger插入一条数据到qrtz_fired_triggers表,然后会为这个Trigger对应的任务分配一个线程来执行,执行完毕后删除Trigger在qrtz_fired_triggers表里的记录。
2. Trigger的状态是怎么变化的
Trigger不触发时状态是WAITING,表示等待着被触发并且允许被触发;
当Trigger触发时间在30分钟内时会被获取出来等待被fire,此时Trigger状态是ACQUIRED,表示已经被获取;
Trigger被fire后,如果Trigger关联的任务允许并发执行,此时Trigger状态还原为WAITING,表示等待着下一次触发,如果Trigger关联的任务不允许并发执行,此时这个任务关联的所有Trigger的状态会被设置为BLOCKED,表示这些Trigger都阻塞住了;
当Trigger对应的任务被执行完毕后,如果Trigger后续不会再触发了,则删除Trigger,如果执行的任务是不允许并发执行的,则需要将这个任务关联的所有Trigger的状态从BLOCKED还原为WAITING。
3. Trigger如何暂停
我们可以通过暂停JobDetail来暂停其关联的所有Trigger,也可以单独暂停某一个Trigger。
所谓暂停Trigger,其实就是将这个Trigger的状态设置为PAUSED,一旦设置为PAUSED,这些Trigger就不满足触发时间在30分钟内且状态是WAITING,从而就不会被触发了。
4. Quartz如何保证同时只有一个实例执行定时任务
Quartz基于数据库实现了一套分布式锁,可以理解为抢占到锁的实例才有资格来触发Trigger从而执行定时任务。
相关推荐
- 详解DNFSB2毒王的各种改动以及大概的加点框架
-
首先附上改动部分,然后逐项分析第一个,毒攻掌握技能意思是力量智力差距超过15%的话差距会被强行缩小到15%,差距不到15%则无效。举例:2000力量,1650智力,2000*0.85=1700,则智力...
- 通篇干货!纵观 PolarDB-X 并行计算框架
-
作者:玄弟七锋PolarDB-X面向HTAP的混合执行器一文详细说明了PolarDB-X执行器设计的初衷,其初衷一直是致力于为PolarDB-X注入并行计算的能力,兼顾TP和AP场景,逐渐...
- 字节新推理模型逆袭DeepSeek,200B参数战胜671B,豆包史诗级加强
-
梦晨发自凹非寺量子位|公众号QbitAI字节最新深度思考模型,在数学、代码等多项推理任务中超过DeepSeek-R1了?而且参数规模更小。同样是MoE架构,字节新模型Seed-Thinkin...
- 阿里智能化研发起飞!RTP-LLM 实现 Cursor AI 1000 token/s 推理技术揭秘
-
作者|赵骁勇阿里巴巴智能引擎事业部审校|刘侃,KittyRTP-LLM是阿里巴巴大模型预测团队开发的高性能LLM推理加速引擎。它在阿里巴巴集团内广泛应用,支撑着淘宝、天猫、高德、饿...
- 多功能高校校园小程序/校园生活娱乐社交管理小程序/校园系统源码
-
校园系统通常是为学校、学生和教职工提供便捷的数字化管理工具。综合性社交大学校园小程序源码:同城校园小程序-大学校园圈子创业分享,校园趣事,同校跑腿交友综合性论坛。小程序系统基于TP6+Uni-app...
- 婚恋交友系统nuiAPP前端解决上传视频模糊的问题
-
婚恋交友系统-打造您的专属婚恋交友平台系统基于TP6+Uni-app框架开发;客户移动端采用uni-app开发,管理后台TH6开发支持微信公众号端、微信小程序端、H5端、PC端多端账号同步,可快速打包...
- 已节省数百万GPU小时!字节再砍MoE训练成本,核心代码全开源
-
COMET团队投稿量子位|公众号QbitAI字节对MoE模型训练成本再砍一刀,成本可节省40%!刚刚,豆包大模型团队在GitHub上开源了叫做COMET的MoE优化技术。COMET已应用于字节...
- 通用电气完成XA102发动机详细设计审查 将为第六代战斗机提供动力
-
2025年2月19日,美国通用电气航空航天公司(隶属于通用电气公司)宣布,已经完成了“下一代自适应推进系统”(NGAP)计划下提供的XA102自适应变循环发动机的详细设计审查阶段。XA102是通用电气...
- tpxm-19双相钢材质(双相钢f60材质)
-
TPXM-19双相钢是一种特殊的钢材,其独特的化学成分、机械性能以及广泛的应用场景使其在各行业中占有独特的地位。以下是对TPXM-19双相钢的详细介绍。**化学成分**TPXM-19双相钢的主要化学成...
- thinkphp6里怎么给layui数据表格输送数据接口
-
layui官网已经下架了,但是产品还是可以使用。今天一个朋友问我怎么给layui数据表格发送数据接口,当然他是学前端的,后端不怎么懂,自学了tp框架问我怎么调用。其实官方文档上就有相应的数据格式,js...
- 完美可用的全媒体广告精准营销服务平台PHP源码
-
今天测试了一套php开发的企业网站展示平台,还是非常不错的,下面来给大家说一下这套系统。1、系统架构这是一套基于ThinkPHP框架开发的HTML5响应式全媒体广告精准营销服务平台PHP源码。现在基于...
- 一对一源码开发,九大方面完善基础架构
-
以往的直播大多数都是一对多进行直播社交,弊端在于不能满足到每个用户的需求,会降低软件的体验感。伴随着用户需求量的增加,一对一直播源码开始出现。一个完整的一对一直播流程即主播发起直播→观看进入房间观看→...
- Int J Biol Macromol .|交联酶聚集体在分级共价有机骨架上的固定化:用于卤代醇不对称合成的高稳定酶纳米反应器
-
大家好,今天推送的文章发表在InternationalJournalofBiologicalMacromolecules上的“Immobilizationofcross-linkeden...
- 【推荐】一款开源免费的 ChatGPT 聊天管理系统,支持PC、H5等多端
-
如果您对源码&技术感兴趣,请点赞+收藏+转发+关注,大家的支持是我分享最大的动力!!!项目介绍GPTCMS是一款开源且免费(基于GPL-3.0协议开源)的ChatGPT聊天管理系统,它基于先进的GPT...
- 高性能计算(HPC)分布式训练:训练框架、混合精度、计算图优化
-
在深度学习模型愈发庞大的今天,分布式训练、高效计算和资源优化已成为AI开发者的必修课。本文将从数据并行vs模型并行、主流训练框架(如PyTorchDDP、DeepSpeed)、混合精度训练(...
你 发表评论:
欢迎- 一周热门
- 最近发表
-
- 详解DNFSB2毒王的各种改动以及大概的加点框架
- 通篇干货!纵观 PolarDB-X 并行计算框架
- 字节新推理模型逆袭DeepSeek,200B参数战胜671B,豆包史诗级加强
- 阿里智能化研发起飞!RTP-LLM 实现 Cursor AI 1000 token/s 推理技术揭秘
- 多功能高校校园小程序/校园生活娱乐社交管理小程序/校园系统源码
- 婚恋交友系统nuiAPP前端解决上传视频模糊的问题
- 已节省数百万GPU小时!字节再砍MoE训练成本,核心代码全开源
- 通用电气完成XA102发动机详细设计审查 将为第六代战斗机提供动力
- tpxm-19双相钢材质(双相钢f60材质)
- thinkphp6里怎么给layui数据表格输送数据接口
- 标签列表
-
- MVC框架 (46)
- spring框架 (46)
- 框架图 (58)
- bootstrap框架 (43)
- flask框架 (53)
- quartz框架 (51)
- abp框架 (47)
- jpa框架 (47)
- laravel框架 (46)
- express框架 (43)
- springmvc框架 (49)
- 分布式事务框架 (65)
- scrapy框架 (52)
- java框架spring (43)
- grpc框架 (55)
- orm框架有哪些 (43)
- ppt框架 (48)
- 内联框架 (52)
- winform框架 (46)
- gui框架 (44)
- cad怎么画框架 (58)
- ps怎么画框架 (47)
- ssm框架实现登录注册 (49)
- oracle字符串长度 (48)
- oracle提交事务 (47)