QuartzSchedulerThread詳解
QuartzSchedulerThread是一個線程類，負(fù)責(zé)查詢并觸發(fā)Triggers。
public class QuartzSchedulerThread extends Thread {
QuartzSchedulerThread(QuartzScheduler qs, QuartzSchedulerResources qsRsrcs, boolean setDaemon, int threadPrio) {
super(qs.getSchedulerThreadGroup(), qsRsrcs.getThreadName());
........
paused = true;
halted = new AtomicBoolean(false);
}
}
該線程類的主要工作分為以下幾個步驟：
- 等待QuartzScheduler啟動
- 查詢待觸發(fā)的Trigger
- 等待Trigger觸發(fā)時(shí)間到來
- 觸發(fā)Trigger
- 循環(huán)上述步驟
/*-----------------run()方法有刪減----------------------*/
public void run() {
while (!halted.get()) {
// -------------------------------
// 1 等待QuartzScheduler啟動
// -------------------------------
synchronized (sigLock) {
while (paused && !halted.get()) {
// wait until togglePause(false) is called...
sigLock.wait(1000L);
}
}
// -------------------------------
// 2 查詢待觸發(fā)的Trigger
// -------------------------------
int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();
if(availThreadCount > 0) { // will always be true, due to semantics of blockForAvailableThreads...
// 查詢未來（now + idletime）時(shí)間內(nèi)待觸發(fā)的Triggers
// triggers是按觸發(fā)時(shí)間由近及遠(yuǎn)排序的集合
List<OperableTrigger> triggers = qsRsrcs.getJobStore().acquireNextTriggers(
now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());
if (triggers != null && !triggers.isEmpty()) {
now = System.currentTimeMillis();
long triggerTime = triggers.get(0).getNextFireTime().getTime();
long timeUntilTrigger = triggerTime - now;
// 通過循環(huán)阻塞，等待第一個Trigger觸發(fā)時(shí)間
while(timeUntilTrigger > 2) {
synchronized (sigLock) {
if (halted.get()) {
break;
}
}
now = System.currentTimeMillis();
timeUntilTrigger = triggerTime - now;
}
// 通知JobStore，這些Triggers將要被觸發(fā)
List<TriggerFiredResult> res = qsRsrcs.getJobStore().triggersFired(triggers);
if(res != null)
bndles = res;
}
// -------------------------------
// 3 觸發(fā)Triggers
// -------------------------------
for (int i = 0; i < bndles.size(); i++) {
TriggerFiredResult result = bndles.get(i);
TriggerFiredBundle bndle = result.getTriggerFiredBundle();
JobRunShell shell = qsRsrcs.getJobRunShellFactory().createJobRunShell(bndle);
shell.initialize(qs);
qsRsrcs.getThreadPool().runInThread(shell);
}
continue; // while (!halted)
} else { // if(availThreadCount > 0)
// should never happen, if threadPool.blockForAvailableThreads() follows contract
continue; // while (!halted)
}
} // while (!halted)
}
1 等待QuartzScheduler啟動
synchronized (sigLock) {
while (paused && !halted.get()) {
// wait until togglePause(false) is called...
sigLock.wait(1000L);
}
}
循環(huán)檢查paused && !halted.get()條件是否滿足，否則釋放sigLock對象的鎖，并等待，一秒后重試。
當(dāng)QuartzScheduler對象創(chuàng)建并調(diào)用start()方法時(shí)，將喚醒QuartzSchedulerThread線程，即可跳出阻塞塊，繼續(xù)執(zhí)行。
/*QuartzScheduler*/
public void start() throws SchedulerException {
....
schedThread.togglePause(false);
....
}
/*QuartzSchedulerThread*/
void togglePause(boolean pause) {
synchronized (sigLock) {
// 更改暫停狀態(tài)
paused = pause;
if (paused) {
signalSchedulingChange(0);
} else {
// 喚醒在sigLock上等待的所有線程
sigLock.notifyAll();
}
}
}
2 查詢待觸發(fā)的Trigger
Quartz未雨綢繆，從JobStore中獲取當(dāng)前時(shí)間后移一段時(shí)間內(nèi)（idle time + time window）將要觸發(fā)的Triggers，以及在當(dāng)前時(shí)間前移一段時(shí)間內(nèi)（misfireThreshold）錯過觸發(fā)的Triggers(這里僅查詢Trigger的主要信息)。被查詢到的Trggers狀態(tài)變化：STATE_WAITING-->STATE_ACQUIRED。結(jié)果集是以觸發(fā)時(shí)間升序、優(yōu)先級降序的集合。
public List<TriggerKey> selectTriggerToAcquire(Connection conn, long noLaterThan, long noEarlierThan, int maxCount)
throws SQLException {
}
SELECT
TRIGGER_NAME,
TRIGGER_GROUP,
NEXT_FIRE_TIME,
PRIORITY
FROM
QRTZ_TRIGGERS
WHERE
SCHED_NAME = 'TestScheduler'
AND TRIGGER_STATE = ?
AND NEXT_FIRE_TIME <= ?
AND (
MISFIRE_INSTR = - 1
OR (
MISFIRE_INSTR != - 1
AND NEXT_FIRE_TIME >= ?
)
)
ORDER BY
NEXT_FIRE_TIME ASC,
PRIORITY DESC
3 等待Trigger觸發(fā)時(shí)間到來
因?yàn)樯弦徊饺〉玫腡riggers是按時(shí)間排序的集合，所以取集合中的第一個，即觸發(fā)時(shí)間最早的Trigger，等待其觸發(fā)時(shí)間的到來。老套路while循環(huán)+wait實(shí)現(xiàn)。
不過需要注意的是，在此期間，可能有一些新的情況發(fā)生，比如說，新增了一個Trigger，并且該新增的Trigger比前面獲取的觸發(fā)時(shí)間都早，那么就需要將上面獲取的Trigger釋放掉(狀態(tài)變化:STATE_ACQUIRED-->STATE_WAITING)，然后重新查詢Trggers
now = System.currentTimeMillis();
long triggerTime = triggers.get(0).getNextFireTime().getTime();
long timeUntilTrigger = triggerTime - now;
// 當(dāng)觸發(fā)時(shí)間距當(dāng)前時(shí)間<=2 ms時(shí)，結(jié)束循環(huán)
while(timeUntilTrigger > 2) {
synchronized (sigLock) {
if (halted.get()) {
break;
}
// 判斷在此過程中是否有新增的并且觸發(fā)時(shí)間更早的Trigger
// 但是此處有個權(quán)衡，為了一個新增的的Trigger而丟棄當(dāng)前已獲取的是否值得？
// 丟棄當(dāng)前獲取的Trigger并重新獲取需要花費(fèi)一定的時(shí)間，時(shí)間的長短與JobStore的實(shí)現(xiàn)有關(guān)。
// 所以此處做了主觀判斷，如果使用的是數(shù)據(jù)庫存儲，查詢時(shí)間假定為70ms，內(nèi)存存儲假定為7ms
// 如果當(dāng)前時(shí)間距已獲得的第一個Trigger觸發(fā)時(shí)間小于查詢時(shí)間，則認(rèn)為丟棄是不合算的。
if (!isCandidateNewTimeEarlierWithinReason(triggerTime, false)) {
try {
// we could have blocked a long while
// on 'synchronize', so we must recompute
now = System.currentTimeMillis();
timeUntilTrigger = triggerTime - now;
// 距觸發(fā)時(shí)間太早，先休息會吧
if(timeUntilTrigger >= 1)
sigLock.wait(timeUntilTrigger);
} catch (InterruptedException ignore) {
}
}
}
// 如果有新增的且觸發(fā)時(shí)間更早的Trigger過來攪局，則釋放上面已獲取的Trigger，等待下一波查詢
if(releaseIfScheduleChangedSignificantly(triggers, triggerTime)) {
break;
}
now = System.currentTimeMillis();
timeUntilTrigger = triggerTime - now;
}
4 觸發(fā)Trigger
前面提到過，先前只是獲取Trigger的主要信息，其關(guān)聯(lián)的Job、Calendar等信息是在觸發(fā)前獲取的。待Trigger所需信息驗(yàn)證、關(guān)聯(lián)完成后，先行將Trigger的狀態(tài)改為STATE_ACQUIRED-->STATE_COMPLETE。而后將Trigger封裝后的TriggerFiredResult對象交由JobRunShell執(zhí)行。
List<TriggerFiredResult> res = qsRsrcs.getJobStore().triggersFired(triggers);
for (int i = 0; i < bndles.size(); i++) {
TriggerFiredResult result = bndles.get(i);
TriggerFiredBundle bndle = result.getTriggerFiredBundle();
JobRunShell shell = qsRsrcs.getJobRunShellFactory().createJobRunShell(bndle);
shell.initialize(qs);
qsRsrcs.getThreadPool().runInThread(shell);
}
