当前位置：首页 > news >正文

Quartz中禁止并发机制源码级解析

news 2025/7/13 18:11:55

文章目录

Quartz进行任务调度时通常会要求一个任务禁止并发执行，此时只需要在Job类上面添加一个注解@DisallowConcurrentExecution即可。在保存到数据库里面时，对应QRTZ_JOB_DETAILS表中的IS_NONCONCURRENT字段的值为1（true）。那么这里是怎么控制的呢？
在Quartz项目搭建与任务执行源码分析中详细介绍一个正常流程过程中涉及的表和状态的变化过程。总结的表格如下

表名	scheduleJob	acquireNextTriggers	triggersFired	triggeredJobComplete
QRTZ_TRIGGERS	WAITING	（CAS）ACQUIRED	WAITING	WAITING
QRTZ_FIRED_TRIGGERS		ACQUIRED	EXECUTING

而对于禁止并发的任务，其状态则如下所示

表名	scheduleJob	acquireNextTriggers	triggersFired	triggeredJobComplete
QRTZ_TRIGGERS	WAITING	（CAS）ACQUIRED	BLOCKED	WAITING
QRTZ_FIRED_TRIGGERS		ACQUIRED	EXECUTING

调度任务在查询待触发任务时，如果同时查出来一个任务对应的多个触发器，只有第一个有效（通过acquiredJobKeysForNoConcurrentExec集合保证唯一性）
org.quartz.impl.jdbcjobstore.JobStoreSupport#acquireNextTrigger

Set<JobKey> acquiredJobKeysForNoConcurrentExec = new HashSet<JobKey>();
// ... 其他代码省略// If trigger's job is set as @DisallowConcurrentExecution, and it has already been added to result, then// put it back into the timeTriggers set and continue to search for next trigger.JobKey jobKey = nextTrigger.getJobKey();JobDetail job = getDelegate().selectJobDetail(conn, jobKey, getClassLoadHelper());if (job.isConcurrentExectionDisallowed()) {if (acquiredJobKeysForNoConcurrentExec.contains(jobKey)) {continue; // next trigger} else {acquiredJobKeysForNoConcurrentExec.add(jobKey);}}

任务触发时，将不支持并发机制的触发器的状态由 WAITING -> BLOCKED，这样调度线程查询待触发任务时便不会满足条件（调度任务只会查询WAITING状态的触发器）

org.quartz.impl.jdbcjobstore.JobStoreSupport#triggersFired

if (job.isConcurrentExectionDisallowed()) {state = STATE_BLOCKED;force = false;try {getDelegate().updateTriggerStatesForJobFromOtherState(conn, job.getKey(),STATE_BLOCKED, STATE_WAITING);getDelegate().updateTriggerStatesForJobFromOtherState(conn, job.getKey(),STATE_BLOCKED, STATE_ACQUIRED);getDelegate().updateTriggerStatesForJobFromOtherState(conn, job.getKey(),STATE_PAUSED_BLOCKED, STATE_PAUSED);} catch (SQLException e) {throw new JobPersistenceException("Couldn't update states of blocked triggers: "+ e.getMessage(), e);}
}

任务执行完毕，会将触发器的状态修改回来 BLOCKED -> WAITING

org.quartz.core.QuartzScheduler#notifyJobStoreJobComplete
执行org.quartz.spi.JobStore#triggeredJobComplete方法

if (jobDetail.isConcurrentExectionDisallowed()) {getDelegate().updateTriggerStatesForJobFromOtherState(conn,jobDetail.getKey(), STATE_WAITING,STATE_BLOCKED);getDelegate().updateTriggerStatesForJobFromOtherState(conn,jobDetail.getKey(), STATE_PAUSED,STATE_PAUSED_BLOCKED);signalSchedulingChangeOnTxCompletion(0L);
}

除了以上地方，在Quartz启动时，org.quartz.impl.jdbcjobstore.JobStoreSupport#recoverJobs首先会将一些阻塞的任务都修改为WAITING，防止系统崩溃导致任务未执行完而来不及恢复禁止并发任务的状态。

// update inconsistent job states
int rows = getDelegate().updateTriggerStatesFromOtherStates(conn,STATE_WAITING, STATE_ACQUIRED, STATE_BLOCKED);rows += getDelegate().updateTriggerStatesFromOtherStates(conn,STATE_PAUSED, STATE_PAUSED_BLOCKED, STATE_PAUSED_BLOCKED);getLog().info("Freed " + rows+ " triggers from 'acquired' / 'blocked' state.");

另外在项目启动、新增触发器、定时任务补偿等所有涉及数据库操作Trigger中都会执行org.quartz.impl.jdbcjobstore.JobStoreSupport#storeTrigger操作，而其中都会检查状态，checkBlockedState.

if (job.isConcurrentExectionDisallowed() && !recovering) { state = checkBlockedState(conn, job.getKey(), state);
}if (existingTrigger) {getDelegate().updateTrigger(conn, newTrigger, state, job);
} else {getDelegate().insertTrigger(conn, newTrigger, state, job);
}

而checkBlockedState中，就会根据是否禁止并发并判断是否任务已经执行（通过selectFiredTriggerRecordsByJob查询QRTZ_FIRED_TRIGGERS表中是否有数据）返回BLOCKED状态，org.quartz.impl.jdbcjobstore.JobStoreSupport#checkBlockedState。

/*** Determines if a Trigger for the given job should be blocked.  * State can only transition to STATE_PAUSED_BLOCKED/BLOCKED from * PAUSED/STATE_WAITING respectively.* * @return STATE_PAUSED_BLOCKED, BLOCKED, or the currentState. */
protected String checkBlockedState(Connection conn, JobKey jobKey, String currentState)throws JobPersistenceException {// State can only transition to BLOCKED from PAUSED or WAITING.if ((!currentState.equals(STATE_WAITING)) &&(!currentState.equals(STATE_PAUSED))) {return currentState;}try {List<FiredTriggerRecord> lst = getDelegate().selectFiredTriggerRecordsByJob(conn,jobKey.getName(), jobKey.getGroup());if (lst.size() > 0) {FiredTriggerRecord rec = lst.get(0);if (rec.isJobDisallowsConcurrentExecution()) { // OLD_TODO: worry about failed/recovering/volatile job  states?return (STATE_PAUSED.equals(currentState)) ? STATE_PAUSED_BLOCKED : STATE_BLOCKED;}}return currentState;} catch (SQLException e) {throw new JobPersistenceException("Couldn't determine if trigger should be in a blocked state '"+ jobKey + "': "+ e.getMessage(), e);}}