学步园

Quartz是运用最广的任务调度框架，它最核心的组成部分是Scheduler、Trigger、JobDetail，然后给Scheduler配置个线程QuartzSchedulerThread，此线程在Scheduler初始化时启动，等待Scheduler start，然后从JobStore里拿到最近要触发的Trigger，以线程等待的方式等到trigger触发时间点，之后就是执行trigger所关联的JobDetail，最后清扫战场。Scheduler初始化、start和trigger执行的时序图如下所示：

其中，最核心的地方是QuartzSchedulerThread运行机制。下面解析一下它的run方法：

public void run() {<br /> boolean lastAcquireFailed = false;</p> <p> while (!halted) {<br /> try {<br /> // check if we're supposed to pause...<br /> synchronized (pauseLock) {<br /> while (paused && !halted) {<br /> try {<br /> // wait until togglePause(false) is called...<br /> pauseLock.wait(100L);<br /> } catch (InterruptedException ignore) {<br /> }<br /> }</p> <p> if (halted) {<br /> break;<br /> }<br /> }<br /> ......<br /> }<br /> }

 以上是run的最开头的一段，不难看出这是在等待scheduler的start，实际上Quartz就是通过线程的wait或sleep来实现时间调度。继续看代码：

Trigger trigger = null;<br /> long now = System.currentTimeMillis();<br /> signaled = false;<br /> try {<br /> trigger = qsRsrcs.getJobStore().acquireNextTrigger(<br /> ctxt, now + idleWaitTime);<br /> lastAcquireFailed = false;<br /> } catch (JobPersistenceException jpe) {<br /> if(!lastAcquireFailed) {<br /> qs.notifySchedulerListenersError(<br /> "An error occured while scanning for the next trigger to fire.",<br /> jpe);<br /> }<br /> lastAcquireFailed = true;<br /> } catch (RuntimeException e) {<br /> if(!lastAcquireFailed) {<br /> getLog().error("quartzSchedulerThreadLoop: RuntimeException "<br /> +e.getMessage(), e);<br /> }<br /> lastAcquireFailed = true;<br /> }

这段代码是从jobStore里拿到下一个要执行的trigger，一般情况下jobStore使用的是RAMJobStore，即trigger等相关信息存放在内存里，如果需要把任务持久化就得使用可持久化JobStore。继续看代码：

 now = System.currentTimeMillis();<br /> long triggerTime = trigger.getNextFireTime().getTime();<br /> long timeUntilTrigger = triggerTime - now;<br /> long spinInterval = 10;<br /> int numPauses = (int) (timeUntilTrigger / spinInterval);<br /> while (numPauses >= 0 && !signaled) {<br /> try {<br /> Thread.sleep(spinInterval);<br /> } catch (InterruptedException ignore) {<br /> }<br /> now = System.currentTimeMillis();<br /> timeUntilTrigger = triggerTime - now;<br /> numPauses = (int) (timeUntilTrigger / spinInterval);<br /> }<br /> if (signaled) {<br /> try {<br /> qsRsrcs.getJobStore().releaseAcquiredTrigger(<br /> ctxt, trigger);<br /> } catch (JobPersistenceException jpe) {<br /> qs.notifySchedulerListenersError(<br /> "An error occured while releasing trigger '"<br /> + trigger.getFullName() + "'",<br /> jpe);<br /> // db connection must have failed... keep<br /> // retrying until it's up...<br /> releaseTriggerRetryLoop(trigger);<br /> } catch (RuntimeException e) {<br /> getLog().error(<br /> "releaseTriggerRetryLoop: RuntimeException "<br /> +e.getMessage(), e);<br /> // db connection must have failed... keep<br /> // retrying until it's up...<br /> releaseTriggerRetryLoop(trigger);<br /> }<br /> signaled = false;<br /> continue;<br /> }

此段代码是计算下一个trigger的执行时间和现在系统时间的差，然后通过循环线程sleep的方式暂停住此线程，一直等到trigger的执行时间点。继续看代码：

import org.quartz.core.JobRunShell;<br /> JobRunShell shell = null;<br /> try {<br /> shell = qsRsrcs.getJobRunShellFactory().borrowJobRunShell();<br /> shell.initialize(qs, bndle);<br /> } catch (SchedulerException se) {<br /> try {<br /> qsRsrcs.getJobStore().triggeredJobComplete(ctxt,<br /> trigger, bndle.getJobDetail(), Trigger.INSTRUCTION_SET_ALL_JOB_TRIGGERS_ERROR);<br /> } catch (SchedulerException se2) {<br /> qs.notifySchedulerListenersError(<br /> "An error occured while placing job's triggers in error state '"<br /> + trigger.getFullName() + "'", se2);<br /> // db connection must have failed... keep retrying<br /> // until it's up...<br /> errorTriggerRetryLoop(bndle);<br /> }<br /> continue;<br /> }<br /> if (qsRsrcs.getThreadPool().runInThread(shell) == false) {<br /> try {<br /> getLog().error("ThreadPool.runInThread() return false!");<br /> qsRsrcs.getJobStore().triggeredJobComplete(ctxt,<br /> trigger, bndle.getJobDetail(), Trigger.INSTRUCTION_SET_ALL_JOB_TRIGGERS_ERROR);<br /> } catch (SchedulerException se2) {<br /> qs.notifySchedulerListenersError(<br /> "An error occured while placing job's triggers in error state '"<br /> + trigger.getFullName() + "'", se2);<br /> // db connection must have failed... keep retrying<br /> // until it's up...<br /> releaseTriggerRetryLoop(trigger);<br /> }<br /> }

此段代码就是包装trigger，然后通过以JobRunShell为载体，在threadpool里执行trigger所关联的jobDetail。

之后的代码就是清扫战场，就不在累述。