Quartz触发Trigger及状态转换

大家好，欢迎来到IT知识分享网。

前言

分布式定时任务框架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表中，它们的状态是这样的。

在qrtz_fired_triggers表中，它们的状态是这样的。

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它们的状态是这样的。

在qrtz_fired_triggers表中，它们的状态是这样的。

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它们的状态是这样的。

在qrtz_fired_triggers表中，它们都没有记录了。

5.暂停一个Trigger

在上面的所有讨论中，都没有提及Trigger的暂停状态，也就是PAUSED状态，因为这个状态相对独立，没必要和上面的其余状态转换混在一起讨论。

我们可以通过如下手段将Trigger的状态置为PAUSED。

1. 通过Trigger的group和name找到Trigger，然后将其状态置为PAUSED。这种可以理解为暂停一个Trigger；
2. 通过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从而执行定时任务。