【基于springboot分析Quartz（v2.3.2）的启动流程】

基于springboot分析Quartz（v2.3.2）的启动流程

最近公司的定时任务使用了Quartz框架，在开发中经常出现定任务不执行了的问题，但是我又找不到原因所在，可把我愁坏了。于是我决定看看Quartz框架是怎么调度任务的。（ps:适合用过Quart框架的同学阅读，如果从来没有用过Quartz框架的同学，可以看看我之前的文章【Quartz入门】）

如何定位到关键代码

1.通过控制台打印的关键日志入手

在程序启动时候，可以看到控制台会输出很多quartz相关的日志，从这些日志我们可以定位到quartz框架的初始化关键代码,下面是我本地启动时候打印的日志

2024-03-29T22:14:00.779+08:00  INFO 10044 --- [           main] org.quartz.core.QuartzScheduler          : Scheduler meta-data: Quartz Scheduler (v2.3.2) 'quartzScheduler' with instanceId 'NON_CLUSTERED'Scheduler class: 'org.quartz.core.QuartzScheduler' - running locally.NOT STARTED.Currently in standby mode.Number of jobs executed: 0Using thread pool 'org.quartz.simpl.SimpleThreadPool' - with 10 threads.Using job-store 'org.springframework.scheduling.quartz.LocalDataSourceJobStore' - which supports persistence. and is not clustered.2024-03-29T22:14:00.779+08:00  INFO 10044 --- [           main] org.quartz.impl.StdSchedulerFactory      : Quartz scheduler 'quartzScheduler' initialized from an externally provided properties instance.
2024-03-29T22:14:00.779+08:00  INFO 10044 --- [           main] org.quartz.impl.StdSchedulerFactory      : Quartz scheduler version: 2.3.2
2024-03-29T22:14:00.779+08:00  INFO 10044 --- [           main] org.quartz.core.QuartzScheduler          : JobFactory set to: org.springframework.scheduling.quartz.SpringBeanJobFactory@70a898b0
2024-03-29T22:14:01.496+08:00  INFO 10044 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port 8087 (http) with context path ''
2024-03-29T22:14:01.497+08:00  INFO 10044 --- [           main] o.s.s.quartz.SchedulerFactoryBean        : Starting Quartz Scheduler now

我这儿就通过最后一行的打印（o.s.s.quartz.SchedulerFactoryBean : Starting Quartz Scheduler now）定位到具体的代码中如下，并在此debug

image.png

  1. 可以看到scheduler.start()这行代码肯定是我们一个重要的突破口，从字面意思可以得知，Quartz框架在这个地方就启动了。
  2. 从左下角的堆栈信息可以看到quart启动流程是在context.refresh()阶段调用。

从日志定位到了关键方法，接下来我们就深入到start方法，深入到start方法，下面就看看start的核心逻辑到底在干嘛把

2.在job任务中debug分析上下文

image.png
可以看到第一个栈是SimpleThreadPool的WorkerThread内部类的一个线程，顺腾摸瓜最后定位到关键代码入口
QuartzSchedulerThread.run

分析代码

1.SchedulerFactoryBean.start

通过打印的日志定位到，代码入口SchedulerFactoryBean.start

public void start() throws SchedulerException {//首先，检查调度器的状态，如果已经在关闭中（shuttingDown）或已经关闭（closed），则抛出 SchedulerException 异常，表示调度器无法在关闭后重新启动if (shuttingDown|| closed) {throw new SchedulerException("The Scheduler cannot be restarted after shutdown() has been called.");}// QTZ-212 : calling new schedulerStarting() method on the listeners// right after entering start()//调用 notifySchedulerListenersStarting() 方法通知调度器监听器，表示调度器即将启动notifySchedulerListenersStarting();//如果 initialStart 为 null，说明调度器是第一次启动：//设置 initialStart 为当前日期和时间。//调用作业存储器的 schedulerStarted() 方法，通知作业存储器调度器已经启动。//调用 startPlugins() 方法，启动插件。if (initialStart == null) {initialStart = new Date();this.resources.getJobStore().schedulerStarted();            startPlugins();} else {//如果 initialStart 不为 null，说明调度器已经启动过：//调用作业存储器的 schedulerResumed() 方法，通知作业存储器调度器已经恢复运行。resources.getJobStore().schedulerResumed();}//将调度器线程的暂停状态设置为 false，以确保调度器不处于暂停状态。schedThread.togglePause(false);getLog().info("Scheduler " + resources.getUniqueIdentifier() + " started.");//通知调度器监听器调度器已经完全启动。notifySchedulerListenersStarted();
}

看到这儿，嘿嘿关键代码又来咯，核心代码this.resources.getJobStore().schedulerStarted();那我们接着分析吧

public void schedulerStarted() throws SchedulerException {
//首先，检查是否为集群模式（调用 isClustered() 方法）。
//如果是集群模式，创建并初始化集群管理线程（ClusterManager）。//如果指定了 initializersLoader，将其设置为集群管理线程的上下文类加载器。
//调用集群管理线程的 initialize() 方法进行初始化。if (isClustered()) {clusterManagementThread = new ClusterManager();if(initializersLoader != null)clusterManagementThread.setContextClassLoader(initializersLoader);clusterManagementThread.initialize();} else {try {recoverJobs();} catch (SchedulerException se) {throw new SchedulerConfigException("Failure occured during job recovery.", se);}}//初始化触发器misfireHandler = new MisfireHandler();if(initializersLoader != null)misfireHandler.setContextClassLoader(initializersLoader);misfireHandler.initialize();schedulerRunning = true;getLog().debug("JobStore background threads started (as scheduler was started).");
}
  1. clusterManagementThread.initialize 判断当前节点是否是集群中目前执行任务节点，是则发送任务调度通知signalSchedulingChangeImmediately

public void run() {while (!shutdown) {if (!shutdown) {long timeToSleep = getClusterCheckinInterval();long transpiredTime = (System.currentTimeMillis() - lastCheckin);timeToSleep = timeToSleep - transpiredTime;if (timeToSleep <= 0) {timeToSleep = 100L;}if(numFails > 0) {timeToSleep = Math.max(getDbRetryInterval(), timeToSleep);}try {Thread.sleep(timeToSleep);} catch (Exception ignore) {}}if (!shutdown && this.manage()) {signalSchedulingChangeImmediately(0L);}}//while !shutdown
}
  1. misfireHandler.initialize主要就是启动一个线程，去查询错过执行的任务，立即发出调度变更的信号signalSchedulingChangeImmediately，并传递最早的新时间（earliestNewTime）。
@Override
public void run() {while (!shutdown) {long sTime = System.currentTimeMillis();RecoverMisfiredJobsResult recoverMisfiredJobsResult = manage();if (recoverMisfiredJobsResult.getProcessedMisfiredTriggerCount() > 0) {signalSchedulingChangeImmediately(recoverMisfiredJobsResult.getEarliestNewTime());}if (!shutdown) {long timeToSleep = 50l;  // At least a short pause to help balance threadsif (!recoverMisfiredJobsResult.hasMoreMisfiredTriggers()) {timeToSleep = getMisfireThreshold() - (System.currentTimeMillis() - sTime);if (timeToSleep <= 0) {timeToSleep = 50l;}if(numFails > 0) {timeToSleep = Math.max(getDbRetryInterval(), timeToSleep);}}try {Thread.sleep(timeToSleep);} catch (Exception ignore) {}}//while !shutdown}
}

signalSchedulingChangeImmediately具体实现：QuartzSchedulerThread.signalSchedulingChange
到这儿，start方法执行已经到底了，维护了QuartzSchedulerThread类变量

public void signalSchedulingChange(long candidateNewNextFireTime) {synchronized(sigLock) {signaled = true;signaledNextFireTime = candidateNewNextFireTime;sigLock.notifyAll();}
}
总结一下scheduler.start()方法底层核心逻辑
  1. 器群模式实现启动集群线程，检查目前节点状态，如果目前节点可执行任务则标记立即执行任务调度（JobStoreSupport.signalSchedulingChangeImmediately）
  2. 启动查询错过的任务线程MisFireHandler，去监听是否有错过的执行任务，有则发送任务调度通知(JobStoreSupport.signalSchedulingChangeImmediately)

上面两个线程都没真正的去调度我们的任务，主要就是维护集群，发送是否要执行任务调度的信号，执行signalSchedulingChangeImmediately方法，此方法修改的就是QuartzSchedulerThread类变量，以及唤醒sigLock锁，说明有其他线程在获取sigLock，做一些事儿，估计就是真正的在做任务调度的事儿了。
接下来就可以分析QuartzSchedulerThread谁在使用sigLock，但是我没有继续分析哈哈，我是转头去job任务debug一下，看一下上下文方法栈找到调度任务的线程

2.QuartzSchedulerThread.run

通过在job任务中debug，定位到核心的run方法，接下来就是分析它在干嘛了

（SchedulerFactoryBean.afterPropertiesSet()中会进行QuartzScheduler的初始化，初始化过程有个重要的成员变量QuartzSchedulerThread这个线程的run方法就是核心所在）

@Override
public void run() {int acquiresFailed = 0;while (!halted.get()) {try {// check if we're supposed to pause...synchronized (sigLock) {while (paused && !halted.get()) {try {// wait until togglePause(false) is called...sigLock.wait(1000L);} catch (InterruptedException ignore) {}// reset failure counter when paused, so that we don't// wait again after unpausingacquiresFailed = 0;}if (halted.get()) {break;}}// wait a bit, if reading from job store is consistently// failing (e.g. DB is down or restarting)..if (acquiresFailed > 1) {try {long delay = computeDelayForRepeatedErrors(qsRsrcs.getJobStore(), acquiresFailed);Thread.sleep(delay);} catch (Exception ignore) {}}int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();if(availThreadCount > 0) { // will always be true, due to semantics of blockForAvailableThreads...List<OperableTrigger> triggers;long now = System.currentTimeMillis();clearSignaledSchedulingChange();try {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;while(timeUntilTrigger > 2) {synchronized (sigLock) {if (halted.get()) {break;}if (!isCandidateNewTimeEarlierWithinReason(triggerTime, false)) {try {// we could have blocked a long while// on 'synchronize', so we must recomputenow = 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;}// this happens if releaseIfScheduleChangedSignificantly decided to release triggersif(triggers.isEmpty())continue;// set triggers to 'executing'List<TriggerFiredResult> bndles = new ArrayList<TriggerFiredResult>();boolean goAhead = true;synchronized(sigLock) {goAhead = !halted.get();}if(goAhead) {try {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);//QTZ-179 : a problem occurred interacting with the triggers from the db//we release them and loop againfor (int i = 0; i < triggers.size(); i++) {qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));}continue;}}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;}// it's possible to get 'null' if the triggers was paused,// blocked, or other similar occurrences that prevent it being// fired at this time...  or if the scheduler was shutdown (halted)if (bndle == null) {qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));continue;}JobRunShell shell = null;try {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;}if (qsRsrcs.getThreadPool().runInThread(shell) == false) {// this case should never happen, as it is indicative of the// scheduler being shutdown or a bug in the thread pool or// a thread pool being used concurrently - which the docs// say not to do...getLog().error("ThreadPool.runInThread() return false!");qsRsrcs.getJobStore().triggeredJobComplete(triggers.get(i), bndle.getJobDetail(), CompletedExecutionInstruction.SET_ALL_JOB_TRIGGERS_ERROR);}}continue; // while (!halted)}} else { // if(availThreadCount > 0)// should never happen, if threadPool.blockForAvailableThreads() follows contractcontinue; // while (!halted)}long now = System.currentTimeMillis();long waitTime = now + getRandomizedIdleWaitTime();long timeUntilContinue = waitTime - now;synchronized(sigLock) {try {if(!halted.get()) {// QTZ-336 A job might have been completed in the mean time and we might have// missed the scheduled changed signal by not waiting for the notify() yet// Check that before waiting for too long in case this very job needs to be// scheduled very soonif (!isScheduleChanged()) {sigLock.wait(timeUntilContinue);}}} catch (InterruptedException ignore) {}}} catch(RuntimeException re) {getLog().error("Runtime error occurred in main trigger firing loop.", re);}} // while (!halted)// drop references to scheduler stuff to aid garbage collection...qs = null;qsRsrcs = null;
}

上面是 Quartz 中 QuartzSchedulerThread 类的 run() 方法的具体代码。该方法是线程运行的主要逻辑，负责获取触发器并执行作业。

以下是 run() 方法的大致流程：

  1. 定义一个变量 acquiresFailed，用于记录连续获取触发器失败的次数。
  2. 进入一个循环，只要 halted 标志为 false，就会一直执行。
  3. 检查是否需要暂停调度器。
    • 如果需要暂停，进入等待状态，直到调用 togglePause(false) 方法来恢复调度器。
    • 如果 halted 标志为 true，跳出循环。
  4. 如果获取触发器的连续失败次数大于 1，等待一段时间。
    • 等待时间由 computeDelayForRepeatedErrors() 方法计算。
  5. 获取可用的线程数。
  6. 如果有可用线程，则获取下一批触发器并执行作业。
    • 获取触发器时，指定了最大批处理大小和时间窗口。
    • 如果获取触发器过程中发生异常，根据失败次数进行错误处理。
  7. 如果获取到触发器且触发器列表不为空，等待触发器的执行时间到来。
    • 如果期间发生调度器关闭、时间变化等情况，跳出循环。
    • 如果触发器执行时间到达或发生了显著的调度变化，跳出循环。
  8. 如果触发器列表为空，跳过本次循环。
  9. 设置触发器为 “executing” 状态。
  10. 创建 JobRunShell 对象，并初始化。
  • 如果发生异常，标记作业触发指令为 “SET_ALL_JOB_TRIGGERS_ERROR”。
  1. 在线程池中运行 JobRunShell 。
  • 如果返回值为 false，表示调度器已关闭或存在线程池的问题，进行相应的错误处理。
  1. 继续下一次循环，获取并执行下一批触发器。
  2. 如果没有可用线程，继续下一次循环。
  3. 计算随机的空闲等待时间，并等待一段时间。
  • 如果调度计划发生变化，提前结束等待。
  1. 在循环中捕获并处理 RuntimeException 异常。
  2. 当 halted 标志为 true，跳出循环。
  3. 清除对调度器资源的引用，以便垃圾回收。

总结

通过启动日志、以及在任务中debug，反向推理出Quartz在springboot中的启动流程，以及Quartz框架调度任务的核心逻辑。授人以鱼不如授人以渔，希望本篇文章不仅仅能帮助大家理解Quartz，还能帮助大家学会去阅读框架源码。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/578806.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

课时79：流程控制_循环控制_控制解析

课时79：流程控制_循环控制_控制解析

1.4.1 控制解析 学习目标 这一节&#xff0c;我们从 基础知识、简单实践、小结 三个方面来学习。 基础知识 简介 所谓的流程控制&#xff0c;主要针对的是&#xff0c;当我们处于流程步骤执行的过程中&#xff0c;因为某些特殊的原因&#xff0c;不得不停止既定的操作进行步…
阅读更多...
去除图像背景的工具（轻量级，批量式）

去除图像背景的工具（轻量级，批量式）

GitHub地址:danielgatis/rembg&#xff1a;Rembg 是一种去除图像背景的工具 (github.com) 博主在最近做一个比赛项目,然后数据用的是网上爬取的,所以数据有点脏,想清洗一下图像数据,恰好这次看到了个GitHub里面有开源去除背景的库,所以拿来用了,用了之后发现效果还是很不错的,…
阅读更多...
更改chatglm认知

更改chatglm认知

ChatGLM-Efficient-Tuning 下载源代码 下载ChatGLM-Efficient-Tuning 解压 创建虚拟环境 conda create --prefixD:\CondaEnvs\chatglm6btrain python3.10 cd D:\ChatGLM-Efficient-Tuning-main conda activate D:\CondaEnvs\chatglm6btrain安装所需要的包 pip install -r…
阅读更多...
Arduino IDE导出esp8266工程编译后的bin文件

Arduino IDE导出esp8266工程编译后的bin文件

一、导出bin文件的方法一 1.通过IDE直接导出&#xff0c;选择 项目 --> 导出已编译的二进制文件&#xff0c;会在工程下生成 build 文件夹&#xff0c;里面有导出的bin文件。 一、导出bin文件的方法二 通过临时文件&#xff0c;找到生成的bin文件。 临时文件的位置&#…
阅读更多...
VsCode正确解决vue3+Eslint+prettier+Vetur的配置冲突

VsCode正确解决vue3+Eslint+prettier+Vetur的配置冲突

手把手教你VsCode正确解决vue3EslintprettierVetur的配置冲突 VsCode正确解决vue3EslintprettierVetur的配置冲突Eslint文档查看和修改规则&#xff1a;step1&#xff1a;首先快速浏览下规则简要setp2: ctrlF 搜索你要配置规则的英文名&#xff0c;例如attributesetp3: 修改配置…
阅读更多...
小程序UI设计规范，界面设计尺寸详解

小程序UI设计规范，界面设计尺寸详解

作为互联网技术的重要组成部分&#xff0c;小程序在日常生活中发挥着越来越重要的作用。因此&#xff0c;了解和严格遵守小程序的 UI 设计标准非常重要&#xff0c;它不仅可以帮助我们在保证良好用户体验的同时优化小程序&#xff0c;还可以使我们的产品在竞争激烈的市场中占据…
阅读更多...
Kubernetes-running app on kube

Kubernetes-running app on kube

Docker 安装Docker 首先&#xff0c;您需要在Linux机器上安装Docker。如果您不使用Linux&#xff0c;则需要启动一个Linux虚拟机(VM)并在该虚拟机中运行Docker。如果你使用的是Mac或Windows系统&#xff0c;并按照指令安装Docker, Docker将为你建立一个虚拟机&#xff0c;并在…
阅读更多...
速通汇编（二）汇编mov、addsub指令

速通汇编（二）汇编mov、addsub指令

一&#xff0c;mov指令 mov指令的全称是move&#xff0c;从字面上去理解&#xff0c;作用是移动&#xff08;比较确切的说是复制&#xff09;数据&#xff0c;mov指令可以有以下几种形式 无论哪种形式&#xff0c;都是把右边的值移动到左边 mov 寄存器&#xff0c;数据&#…
阅读更多...
数据库系统概论（超详解！！！） 第四节 关系数据库标准语言SQL（Ⅱ）

数据库系统概论（超详解！！！） 第四节 关系数据库标准语言SQL（Ⅱ）

1.数据查询 SELECT [ ALL | DISTINCT] <目标列表达式>[&#xff0c;<目标列表达式>] … FROM <表名或视图名>[&#xff0c; <表名或视图名> ] … [ WHERE <条件表达式> ] [ GROUP BY <列名1> [ HAVING <条件表达式> ] ] [ ORDER BY…
阅读更多...
Elment ui 动态表格与表单校验 列表数据 组件

Elment ui 动态表格与表单校验 列表数据 组件

组件做个记录&#xff0c;方便以后会用到。 效果&#xff1a; 代码 &#xff1a; <template><el-dialog title"商品详情" :visible.sync"dialogVisible" width"80%"><el-tabs v-model"activeTab"><el-tab-pane…
阅读更多...
esp单片机下arduino_gfx不相干显示驱动优化对flash空间的占用对比

esp单片机下arduino_gfx不相干显示驱动优化对flash空间的占用对比

一般情况下&#xff0c;很多esp32或者esp8266下的tft模块驱动都会包含很多种&#xff0c;而我们只需要其中一种&#xff0c;那就有个疑问这些被编译进的显示驱动到底占用了多少空间&#xff0c;是否需要把他优化掉&#xff1f; 这是默认的驱动列表&#xff1a; 84个文件&…
阅读更多...
C++:list类

C++:list类

list的介绍 1. list 是可以在常数范围内在任意位置进行插入和删除的序列式容器,并且该容器可以前后双向迭代 2. list 的底层是双向链表结构&#xff0c;双向链表中每个元素存储在互不相关的独立节点中&#xff0c;在节点中通过指针指向其前一个元素和后一个元素。 3. list 与 …
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023