一、QuartzSchedulerThread 的 run方法大致阐述

先说一下run方法的执行时机：

当 Quartzscheduler执行start方法 时，方法体中有一句

schedThread.togglePause(false); ，接着就会调用 QuartzSchedulerThread 下的 togglePause 方法，将 paused 置为 false ，在此之后， QuartzSchedulerThread 下的 run 方法开始真正运行

/**通知主处理循环在下一个可能的点暂停 */ void togglePause(boolean pause) { synchronized (sigLock) { paused = pause; if (paused) { signalSchedulingChange(0); } else { sigLock.notifyAll(); } } }复制代码 public void run() { int acquiresFailed = 0; // 这里就是判断调度器是否该停止，如果没有收到信号的话，这个调度器是一直处于循环之中的 while (!halted.get()) { try { // 这里是检查我们是否应该暂停 synchronized (sigLock) { // 我们在初始化的时候，paused 是置为 true的， // 因此在上文中，我们才说 // 当 Quartzscheduler 执行 start方法时调用togglePause, // 将 paused 置为false，run 方法才开始运行 // 也是因为此处的判断 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) {} } // 从线程池拿出空闲可利用的线程数量 // 这里多谈一嘴 blockForAvailableThreads()方法 // 它是一个阻塞式方法，直到至少有一个可用线程。 int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads(); if(availThreadCount > 0) { List triggers; long now = System.currentTimeMillis(); // 清除信号调度变更 clearSignaledSchedulingChange(); try { //如果可用线程数量足够那么就是30后再次扫描， //acquireNextTriggers方法的三个参数的意思分别是： //idleWaitTime ：为如果没有的再次扫描的时间，默认是 // private static long DEFAULT_IDLE_WAIT_TIME = 30L * 1000L; 30秒 //Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()) ：这里的意思就是一次最多能取几个出来 //batchTimeWindow ：默认是0，同样是一个时间范围， //如果有两个任务只差一两秒，而执行线程数量满足及batchTimeWindow时间也满足的情况下就会两个都取出来 // 具体的方法的执行，后文再看~ triggers = qsRsrcs.getJobStore().acquireNextTriggers( now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow()); acquiresFailed = 0; if (log.isDebugEnabled()){ //… } //在获取到 triggers 触发器不为空后， //trigger列表是以下次执行时间排序查出来的 if (triggers != null && !triggers.isEmpty()) { now = System.currentTimeMillis(); //取出集合中最早执行的触发器 //获取它的下一个触发时间 long triggerTime = triggers.get(0).getNextFireTime().getTime(); long timeUntilTrigger = triggerTime – now; // 判断距离执行时间是否大于2 毫秒 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; } // this happens if releaseIfScheduleChangedSignificantly decided to release triggers if(triggers.isEmpty()) continue; // set triggers to ‘executing’ List bndles = new ArrayList(); boolean goAhead = true; synchronized(sigLock) { goAhead = !halted.get(); } if(goAhead) { try { //开始根据需要执行的trigger从数据库中获取相应的JobDetail 同时这一步也更新了 triggers 的状态，稍后会讲到~ List 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; } } //将查询到的结果封装成为 TriggerFiredResult for (int i = 0; i 0) // should never happen, if threadPool.blockForAvailableThreads() follows contract continue; // while (!halted) } long now = System.currentTimeMillis(); long waitTime = now + getRandomizedIdleWaitTime(); long timeUntilContinue = waitTime – now; synchronized(sigLock) { // …. } } // …. } // while (!halted) // …. }复制代码

二、一些细节

2.1、先获取线程池中的可用线程数量

（若没有可用的会阻塞，直到有可用的）；

int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads(); 复制代码

2.2、获取 30m 内要执行的 trigger

(即 acquireNextTriggers )

我们来看一看 acquireNextTriggers 方法

首先说 acquireNextTriggers 具体实现是在 JobStoreSupport 中，同时 quartz 与数据库关联的实现大都在 JobStoreSupport 中，当然更具体的SQL执行还是在 DriverDelegate 接口下的。

acquireNextTriggers 做了哪些事情呢？

我们看看这两个方法：

首先看第一个 acquireNextTrigger(conn, noLaterThan, maxCount, timeWindow);

主要就是获取下一个 30m内可执行的triggers的触发器，在里面 JobStoreSupport 从数据库取出 triggers 时是按照 nextFireTime 排序的

更具体的就需要大家点进方法去看啦~另外里面还包含triggers状态的变更，属于是更加细节化的东西。

第二个就是获取到触发的触发记录~

然后在执行 executeInNonManagedTXLock 时，是需要先获得锁，之后再在提交时释放锁的。

待直到获取的trigger中最先执行的trigger在2ms内；

if (triggers != null && !triggers.isEmpty()) { now = System.currentTimeMillis(); long triggerTime = triggers.get(0).getNextFireTime().getTime(); long timeUntilTrigger = triggerTime – now; while(timeUntilTrigger > 2) { //… } }复制代码

2.3、triggersFired(triggers)

List res = qsRsrcs.getJobStore().triggersFired(triggers);

这一步看着只是获取了 List 对象，实际上在 triggersFired(triggers) 方法中隐藏了很多东西~

首先查询，确保触发器没有被删除、暂停或完成…，就更新 firedTrigger 的 status=STATE_EXECUTING ;代码的注释上还说，如果没有这些就会将状态该为 deleted

另外就是更新触发触发器：

如果下一次的执行时间为空，状态则改为 STATE_COMPLETE

在执行 executeInNonManagedTXLock 方法时，提交前先获得锁， transOwner = getLockHandler().obtainLock(conn, lockName);

最后是释放锁： commitConnection(conn);

2.4、创建JobRunShell，放进线程池执行

针对每个要执行的trigger，创建JobRunShell，并放入线程池执行：

然后由execute:执行job

更详细的看不下去啦~

来源：https://juejin.cn/post/7131671438901116935