QuartzSchedulerThread詳解
QuartzSchedulerThread是一個線程類纲刀,負責查詢并觸發(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ā)時間到來
- 觸發(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)時間內待觸發(fā)的Triggers
// triggers是按觸發(fā)時間由近及遠排序的集合
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ā)時間
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對象的鎖暂论,并等待面褐,一秒后重試。
當QuartzScheduler
對象創(chuàng)建并調用start()
方法時取胎,將喚醒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中獲取當前時間后移一段時間內(idle time + time window)將要觸發(fā)的Triggers,以及在當前時間前移一段時間內(misfireThreshold)錯過觸發(fā)的Triggers(這里僅查詢Trigger的主要信息)觉痛。被查詢到的Trggers狀態(tài)變化:STATE_WAITING-->STATE_ACQUIRED役衡。結果集是以觸發(fā)時間升序、優(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ā)時間到來
因為上一步取得的Triggers是按時間排序的集合手蝎,所以取集合中的第一個榕莺,即觸發(fā)時間最早的Trigger,等待其觸發(fā)時間的到來柑船。老套路while循環(huán)+wait實現(xiàn)帽撑。
不過需要注意的是,在此期間鞍时,可能有一些新的情況發(fā)生,比如說扣蜻,新增了一個Trigger逆巍,并且該新增的Trigger比前面獲取的觸發(fā)時間都早,那么就需要將上面獲取的Trigger釋放掉(狀態(tài)變化:STATE_ACQUIRED-->STATE_WAITING)莽使,然后重新查詢Trggers
now = System.currentTimeMillis();
long triggerTime = triggers.get(0).getNextFireTime().getTime();
long timeUntilTrigger = triggerTime - now;
// 當觸發(fā)時間距當前時間<=2 ms時锐极,結束循環(huán)
while(timeUntilTrigger > 2) {
synchronized (sigLock) {
if (halted.get()) {
break;
}
// 判斷在此過程中是否有新增的并且觸發(fā)時間更早的Trigger
// 但是此處有個權衡,為了一個新增的的Trigger而丟棄當前已獲取的是否值得芳肌?
// 丟棄當前獲取的Trigger并重新獲取需要花費一定的時間灵再,時間的長短與JobStore的實現(xiàn)有關。
// 所以此處做了主觀判斷亿笤,如果使用的是數(shù)據(jù)庫存儲翎迁,查詢時間假定為70ms,內存存儲假定為7ms
// 如果當前時間距已獲得的第一個Trigger觸發(fā)時間小于查詢時間净薛,則認為丟棄是不合算的汪榔。
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ā)時間太早,先休息會吧
if(timeUntilTrigger >= 1)
sigLock.wait(timeUntilTrigger);
} catch (InterruptedException ignore) {
}
}
}
// 如果有新增的且觸發(fā)時間更早的Trigger過來攪局肃拜,則釋放上面已獲取的Trigger痴腌,等待下一波查詢
if(releaseIfScheduleChangedSignificantly(triggers, triggerTime)) {
break;
}
now = System.currentTimeMillis();
timeUntilTrigger = triggerTime - now;
}
4 觸發(fā)Trigger
前面提到過,先前只是獲取Trigger的主要信息燃领,其關聯(lián)的Job士聪、Calendar等信息是在觸發(fā)前獲取的。待Trigger所需信息驗證猛蔽、關聯(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);
}