@Override
public void run() {
int acquiresFailed = 0 ;
while (!halted.get()) {
try {
// check if we're supposed to pause...
synchronized (sigLock) {
while (paused && !halted.get()) {
try {
// wait until togglePause(false) is called...
sigLock.wait(1000L);
} catch (InterruptedException ignore) {
}
// reset failure counter when paused, so that we don't
// wait again after unpausing
acquiresFailed = 0 ;
}
if (halted.get()) {
break ;
}
}
// wait a bit, if reading from job store is consistently
// failing (e.g. DB is down or restarting)..
if (acquiresFailed > 1 ) {
try {
long delay = computeDelayForRepeatedErrors(qsRsrcs.getJobStore(), acquiresFailed);
Thread.sleep(delay);
} catch (Exception ignore) {
}
}
// 获取工作线程池中可用的线程数量
int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();
if (availThreadCount > 0 ) { // will always be true, due to semantics of blockForAvailableThreads...
List<OperableTrigger> triggers;
long now = System.currentTimeMillis();
clearSignaledSchedulingChange();
try {
// idleWaitTime默认是30秒,
// maxBatchSize默认为1 , 每次获取一个Trigger来执行,
// 最后一个参数,默认为0
// 总的来说,这个方法是将最近30秒内最近的一批需要执行的JOB给抓出来,抓取数量为maxBatchSize,同时更新JOB的状态为ACQUIRED
// 同时更新JOB的下次执行时间。 这里如果maxBatchSize等于1 的话,则默认不加悲观锁
// 该方法后面重点讲。
triggers = qsRsrcs.getJobStore().acquireNextTriggers(
now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());
//上一步获取成功将失败标志置为false;
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 ;
}
// triggers不为空
if (triggers != null && !triggers.isEmpty()) {
// 获取
now = System.currentTimeMillis();
// 获取第一个
long triggerTime = triggers.get( 0 ).getNextFireTime().getTime();
//计算距离trigger触发的时间
long timeUntilTrigger = triggerTime - now;
while (timeUntilTrigger > 2 ) {
synchronized (sigLock) {
if (halted.get()) {
break ;
}
if (!isCandidateNewTimeEarlierWithinReason(triggerTime, false )) {
try {
// we could have blocked a long while
// on 'synchronize', so we must recompute
now = System.currentTimeMillis();
timeUntilTrigger = triggerTime - now;
if (timeUntilTrigger >= 1 )
sigLock.wait(timeUntilTrigger);
} catch (InterruptedException ignore) {
}
}
}
//这里的意思,主要是当有定时器信息发生改变的时候,有个新的JOB更加迫切,这里就会判断是否值得问题
// 如果重新获取新的任务的时间,依赖赶不上新任务的触发时间,那么就继续执行当前的任务。否则放弃任务
// 比如: 下次任务的触发时间是0.1秒后,但是获取任务的时候就需要0.2秒,那么即使去获取了,也没有意义,索性直接执行当前的任务。
if (releaseIfScheduleChangedSignificantly(triggers, triggerTime)) {
break ;
}
now = System.currentTimeMillis();
timeUntilTrigger = triggerTime - now;
}
// 这里再次做一次非空判断,防止上面那个循环里面,将triggers清空了
// this happens if releaseIfScheduleChangedSignificantly decided to release triggers
if (triggers.isEmpty())
continue ;
// set triggers to 'executing'
List<TriggerFiredResult> bndles = new ArrayList<TriggerFiredResult>();
boolean goAhead = true ;
synchronized (sigLock) {
goAhead = !halted.get();
}
if (goAhead) {
try {
// 这个地方是获取trigger的详情信息,并且做一系列的状态判断,防止重复执行,是否是串行还是并行执行,在这里面都有处理
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);
//QTZ-179 : a problem occurred interacting with the triggers from the db
//we release them and loop again
for ( int i = 0 ; i < triggers.size(); i++) {
qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));
}
continue ;
}
}
// 循环trigger的结果信息
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 ;
}
// it's possible to get 'null' if the triggers was paused,
// blocked, or other similar occurrences that prevent it being
// fired at this time... or if the scheduler was shutdown (halted)
// 为空的话,则释放状态绑定,将状态从ACQUIRED修改为WAITING
if (bndle == null ) {
qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));
continue ;
}
JobRunShell shell = null ;
try {
// 构建任务执行的脚本信息
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 ;
}
// 将任务丢到线程池里面去处理,至此,任务触发算是完成了
if (qsRsrcs.getThreadPool().runInThread(shell) == false ) {
// this case should never happen, as it is indicative of the
// scheduler being shutdown or a bug in the thread pool or
// a thread pool being used concurrently - which the docs
// say not to do...
getLog().error( "ThreadPool.runInThread() return false!" );
qsRsrcs.getJobStore().triggeredJobComplete(triggers.get(i), bndle.getJobDetail(), CompletedExecutionInstruction.SET_ALL_JOB_TRIGGERS_ERROR);
}
}
continue ; // while (!halted)
}
} else { // if(availThreadCount > 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) {
try {
if (!halted.get()) {
// QTZ-336 A job might have been completed in the mean time and we might have
// missed the scheduled changed signal by not waiting for the notify() yet
// Check that before waiting for too long in case this very job needs to be
// scheduled very soon
if (!isScheduleChanged()) {
sigLock.wait(timeUntilContinue);
}
}
} catch (InterruptedException ignore) {
}
}
} catch (RuntimeException re) {
getLog().error( "Runtime error occurred in main trigger firing loop." , re);
}
} // while (!halted)
// drop references to scheduler stuff to aid garbage collection...
qs = null ;
qsRsrcs = null ;
}
|