前言

上一篇文章Quartz数据库表分析介绍了Quartz默认提供的11张表，本文将具体分析Quartz是如何调度的，是如何通过数据库的方式来现在分布式调度。

调度线程

Quartz内部提供的调度类是QuartzScheduler，而QuartzScheduler会委托QuartzSchedulerThread去实时调度；当调度完需要去执行job的时候QuartzSchedulerThread并没有直接去执行job，
而是交给ThreadPool去执行job，具体使用什么ThreadPool，初始化多线线程，可以在配置文件中进行配置：

org.quartz.threadPool.class: org.quartz.simpl.SimpleThreadPool org.quartz.threadPool.threadCount: 10 org.quartz.threadPool.threadPriority: 5

常用的线程池是SimpleThreadPool，这里默认启动了10个线程，在SimpleThreadPool会创建10个WorkerThread，由WorkerThread去执行具体的job；

调度分析

QuartzSchedulerThread是调度的核心类，具体Quartz是如何实现调度的，可以查看QuartzSchedulerThread核心源码：

public void run() {     boolean lastAcquireFailed = false;       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) {                     }                 }                   if (halted.get()) {                     break;                 }             }               int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();             if(availThreadCount > 0) { // will always be true, due to semantics of blockForAvailableThreads...                   List<OperableTrigger> triggers = null;                   long now = System.currentTimeMillis();                   clearSignaledSchedulingChange();                 try {                     triggers = qsRsrcs.getJobStore().acquireNextTriggers(                             now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());                     lastAcquireFailed = false;                     if (log.isDebugEnabled())                          log.debug("batch acquisition of " + (triggers == null ? 0 : triggers.size()) + " triggers");                 } catch (JobPersistenceException jpe) {                     if(!lastAcquireFailed) {                         qs.notifySchedulerListenersError(                             "An error occurred while scanning for the next triggers to fire.",                             jpe);                     }                     lastAcquireFailed = true;                     continue;                 } catch (RuntimeException e) {                     if(!lastAcquireFailed) {                         getLog().error("quartzSchedulerThreadLoop: RuntimeException "                                 +e.getMessage(), e);                     }                     lastAcquireFailed = true;                     continue;                 }                   if (triggers != null && !triggers.isEmpty()) {                       now = System.currentTimeMillis();                     long triggerTime = triggers.get(0).getNextFireTime().getTime();                     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) {                                 }                             }                         }                         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<TriggerFiredResult> bndles = new ArrayList<TriggerFiredResult>();                       boolean goAhead = true;                     synchronized(sigLock) {                         goAhead = !halted.get();                     }                     if(goAhead) {                         try {                             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;                         }                       }                       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)                         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; }

1.halted和paused

这是两个boolean值的标志参数，分别表示：停止和暂停；halted默认为false，当QuartzScheduler执行shutdown()时才会更新为true；paused默认是true，当QuartzScheduler执行start()时
更新为false；正常启动之后QuartzSchedulerThread就可以往下执行了；

2.availThreadCount

查询SimpleThreadPool是否有可用的WorkerThread，如果availThreadCount>0可以往下继续执行其他逻辑，否则继续检查；

3.acquireNextTriggers

查询一段时间内将要被调度的triggers，这里有3个比较重要的参数分别是：idleWaitTime，maxBatchSize，batchTimeWindow，这3个参数都可以在配置文件中进行配置：

org.quartz.scheduler.idleWaitTime:30000 org.quartz.scheduler.batchTriggerAcquisitionMaxCount:1 org.quartz.scheduler.batchTriggerAcquisitionFireAheadTimeWindow:0

idleWaitTime:在调度程序处于空闲状态时，调度程序将在重新查询可用触发器之前等待的时间量（以毫秒为单位），默认是30秒;
batchTriggerAcquisitionMaxCount:允许调度程序节点一次获取（用于触发）的触发器的最大数量，默认是1;
batchTriggerAcquisitionFireAheadTimeWindow:允许触发器在其预定的火灾时间之前被获取和触发的时间（毫秒）的时间量，默认是0;

往下继续查看acquireNextTriggers方法源码：

public List<OperableTrigger> acquireNextTriggers(final long noLaterThan, final int maxCount, final long timeWindow)     throws JobPersistenceException {           String lockName;     if(isAcquireTriggersWithinLock() || maxCount > 1) {          lockName = LOCK_TRIGGER_ACCESS;     } else {         lockName = null;     }     return executeInNonManagedTXLock(lockName,              new TransactionCallback<List<OperableTrigger>>() {                 public List<OperableTrigger> execute(Connection conn) throws JobPersistenceException {                     return acquireNextTrigger(conn, noLaterThan, maxCount, timeWindow);                 }             },             ......             }); }

可以发现只有在设置了acquireTriggersWithinLock或者batchTriggerAcquisitionMaxCount>1情况下才使用LOCK_TRIGGER_ACCESS锁，也就是说在默认参数配置的情况下，这里是没有使用锁的，
那么如果多个节点同时去执行acquireNextTriggers，会不会出现同一个trigger在多个节点都被执行？
注：acquireTriggersWithinLock可以在配置文件中进行配置：

org.quartz.jobStore.acquireTriggersWithinLock=true

acquireTriggersWithinLock：获取triggers的时候是否需要使用锁，默认是false，如果batchTriggerAcquisitionMaxCount>1最好同时设置acquireTriggersWithinLock为true；

带着问题继续查看TransactionCallback内部的acquireNextTrigger方法源码：

protected List<OperableTrigger> acquireNextTrigger(Connection conn, long noLaterThan, int maxCount, long timeWindow)     throws JobPersistenceException {     if (timeWindow < 0) {       throw new IllegalArgumentException();     }           List<OperableTrigger> acquiredTriggers = new ArrayList<OperableTrigger>();     Set<JobKey> acquiredJobKeysForNoConcurrentExec = new HashSet<JobKey>();     final int MAX_DO_LOOP_RETRY = 3;     int currentLoopCount = 0;     do {         currentLoopCount ++;         try {             List<TriggerKey> keys = getDelegate().selectTriggerToAcquire(conn, noLaterThan + timeWindow, getMisfireTime(), maxCount);                           // No trigger is ready to fire yet.             if (keys == null || keys.size() == 0)                 return acquiredTriggers;               long batchEnd = noLaterThan;               for(TriggerKey triggerKey: keys) {                 // If our trigger is no longer available, try a new one.                 OperableTrigger nextTrigger = retrieveTrigger(conn, triggerKey);                 if(nextTrigger == null) {                     continue; // next trigger                 }                                   // If trigger's job is set as @DisallowConcurrentExecution, and it has already been added to result, then                 // put it back into the timeTriggers set and continue to search for next trigger.                 JobKey jobKey = nextTrigger.getJobKey();                 JobDetail job;                 try {                     job = retrieveJob(conn, jobKey);                 } catch (JobPersistenceException jpe) {                     try {                         getLog().error("Error retrieving job, setting trigger state to ERROR.", jpe);                         getDelegate().updateTriggerState(conn, triggerKey, STATE_ERROR);                     } catch (SQLException sqle) {                         getLog().error("Unable to set trigger state to ERROR.", sqle);                     }                     continue;                 }                                   if (job.isConcurrentExectionDisallowed()) {                     if (acquiredJobKeysForNoConcurrentExec.contains(jobKey)) {                         continue; // next trigger                     } else {                         acquiredJobKeysForNoConcurrentExec.add(jobKey);                     }                 }                                   if (nextTrigger.getNextFireTime().getTime() > batchEnd) {                   break;                 }                 // We now have a acquired trigger, let's add to return list.                 // If our trigger was no longer in the expected state, try a new one.                 int rowsUpdated = getDelegate().updateTriggerStateFromOtherState(conn, triggerKey, STATE_ACQUIRED, STATE_WAITING);                 if (rowsUpdated <= 0) {                     continue; // next trigger                 }                 nextTrigger.setFireInstanceId(getFiredTriggerRecordId());                 getDelegate().insertFiredTrigger(conn, nextTrigger, STATE_ACQUIRED, null);                   if(acquiredTriggers.isEmpty()) {                     batchEnd = Math.max(nextTrigger.getNextFireTime().getTime(), System.currentTimeMillis()) + timeWindow;                 }                 acquiredTriggers.add(nextTrigger);             }               // if we didn't end up with any trigger to fire from that first             // batch, try again for another batch. We allow with a max retry count.             if(acquiredTriggers.size() == 0 && currentLoopCount < MAX_DO_LOOP_RETRY) {                 continue;             }                           // We are done with the while loop.             break;         } catch (Exception e) {             throw new JobPersistenceException(                       "Couldn't acquire next trigger: " + e.getMessage(), e);         }     } while (true);           // Return the acquired trigger list     return acquiredTriggers; }

首先看一下在执行selectTriggerToAcquire方法时引入了新的参数：misfireTime=当前时间-MisfireThreshold，MisfireThreshold可以在配置文件中进行配置：

org.quartz.jobStore.misfireThreshold: 60000

misfireThreshold：叫触发器超时，比如有10个线程，但是有11个任务，这样就有一个任务被延迟执行了，可以理解为调度引擎可以忍受这个超时的时间；具体的查询SQL如下所示：

SELECT TRIGGER_NAME, TRIGGER_GROUP, NEXT_FIRE_TIME, PRIORITY   FROM qrtz_TRIGGERS  WHERE SCHED_NAME = 'myScheduler'    AND TRIGGER_STATE = 'WAITING'    AND NEXT_FIRE_TIME <= noLaterThan    AND (MISFIRE_INSTR = -1 OR        (MISFIRE_INSTR != -1 AND NEXT_FIRE_TIME >= noEarlierThan))  ORDER BY NEXT_FIRE_TIME ASC, PRIORITY DESC

这里的noLaterThan=当前时间+idleWaitTime+batchTriggerAcquisitionFireAheadTimeWindow，
noEarlierThan=当前时间-MisfireThreshold；
在查询完之后，会遍历执行updateTriggerStateFromOtherState()方法更新trigger的状态从STATE_WAITING到STATE_ACQUIRED，并且会判断rowsUpdated是否大于0，这样就算多个节点都查询到相同的trigger，但是肯定只会有一个节点更新成功；更新完状态之后，往qrtz_fired_triggers表中插入一条记录，表示当前trigger已经触发，状态为STATE_ACQUIRED；

4.executeInNonManagedTXLock

Quartz的分布式锁被用在很多地方，下面具体看一下Quartz是如何实现分布式锁的，executeInNonManagedTXLock方法源码如下：

protected <T> T executeInNonManagedTXLock(         String lockName,          TransactionCallback<T> txCallback, final TransactionValidator<T> txValidator) throws JobPersistenceException {     boolean transOwner = false;     Connection conn = null;     try {         if (lockName != null) {             // If we aren't using db locks, then delay getting DB connection              // until after acquiring the lock since it isn't needed.             if (getLockHandler().requiresConnection()) {                 conn = getNonManagedTXConnection();             }                           transOwner = getLockHandler().obtainLock(conn, lockName);         }                   if (conn == null) {             conn = getNonManagedTXConnection();         }                   final T result = txCallback.execute(conn);         try {             commitConnection(conn);         } catch (JobPersistenceException e) {             rollbackConnection(conn);             if (txValidator == null || !retryExecuteInNonManagedTXLock(lockName, new TransactionCallback<Boolean>() {                 @Override                 public Boolean execute(Connection conn) throws JobPersistenceException {                     return txValidator.validate(conn, result);                 }             })) {                 throw e;             }         }           Long sigTime = clearAndGetSignalSchedulingChangeOnTxCompletion();         if(sigTime != null && sigTime >= 0) {             signalSchedulingChangeImmediately(sigTime);         }                   return result;     } catch (JobPersistenceException e) {         rollbackConnection(conn);         throw e;     } catch (RuntimeException e) {         rollbackConnection(conn);         throw new JobPersistenceException("Unexpected runtime exception: "                 + e.getMessage(), e);     } finally {         try {             releaseLock(lockName, transOwner);         } finally {             cleanupConnection(conn);         }     } }

大致分成3个步骤：获取锁，执行逻辑，释放锁；getLockHandler().obtainLock表示获取锁txCallback.execute(conn)表示执行逻辑，commitConnection(conn)表示释放锁
Quartz的分布式锁接口类是Semaphore，默认具体的实现是StdRowLockSemaphore，具体接口如下：

public interface Semaphore {     boolean obtainLock(Connection conn, String lockName) throws LockException;     void releaseLock(String lockName) throws LockException;     boolean requiresConnection(); }

具体看一下obtainLock()是如何获取锁的，源码如下：

public boolean obtainLock(Connection conn, String lockName)     throws LockException {     if (!isLockOwner(lockName)) {         executeSQL(conn, lockName, expandedSQL, expandedInsertSQL);         getThreadLocks().add(lockName);              } else if(log.isDebugEnabled()) {              }     return true; }   protected void executeSQL(Connection conn, final String lockName, final String expandedSQL, final String expandedInsertSQL) throws LockException {     PreparedStatement ps = null;     ResultSet rs = null;     SQLException initCause = null;           int count = 0;     do {         count++;         try {             ps = conn.prepareStatement(expandedSQL);             ps.setString(1, lockName);                           rs = ps.executeQuery();             if (!rs.next()) {                 getLog().debug(                         "Inserting new lock row for lock: '" + lockName + "' being obtained by thread: " +                          Thread.currentThread().getName());                 rs.close();                 rs = null;                 ps.close();                 ps = null;                 ps = conn.prepareStatement(expandedInsertSQL);                 ps.setString(1, lockName);                   int res = ps.executeUpdate();                                   if(res != 1) {                    if(count < 3) {                         try {                             Thread.sleep(1000L);                         } catch (InterruptedException ignore) {                             Thread.currentThread().interrupt();                         }                         continue;                     }                 }             }                           return; // obtained lock, go         } catch (SQLException sqle) {             ......     } while(count < 4);   }

obtainLock首先判断是否已经获取到锁，如果没有执行方法executeSQL，其中有两条重要的SQL，分别是：expandedSQL和expandedInsertSQL，以SCHED_NAME = ‘myScheduler’为例：

SELECT * FROM QRTZ_LOCKS WHERE SCHED_NAME = 'myScheduler' AND LOCK_NAME = ? FOR UPDATE INSERT INTO QRTZ_LOCKS(SCHED_NAME, LOCK_NAME) VALUES ('myScheduler', ?)

select语句后面添加了FOR UPDATE，如果LOCK_NAME存在，当多个节点去执行此SQL时，只有第一个节点会成功，其他的节点都将进入等待；
如果LOCK_NAME不存在，多个节点同时执行expandedInsertSQL，只会有一个节点插入成功，执行插入失败的节点将进入重试，重新执行expandedSQL；
txCallback执行完之后，执行commitConnection操作，这样当前节点就释放了LOCK_NAME，其他节点可以竞争获取锁，最后执行了releaseLock；

5.triggersFired

表示触发trigger，具体代码如下：

protected TriggerFiredBundle triggerFired(Connection conn,         OperableTrigger trigger)     throws JobPersistenceException {     JobDetail job;     Calendar cal = null;       // Make sure trigger wasn't deleted, paused, or completed...     try { // if trigger was deleted, state will be STATE_DELETED         String state = getDelegate().selectTriggerState(conn,                 trigger.getKey());         if (!state.equals(STATE_ACQUIRED)) {             return null;         }     } catch (SQLException e) {         throw new JobPersistenceException("Couldn't select trigger state: "                 + e.getMessage(), e);     }       try {         job = retrieveJob(conn, trigger.getJobKey());         if (job == null) { return null; }     } catch (JobPersistenceException jpe) {         try {             getLog().error("Error retrieving job, setting trigger state to ERROR.", jpe);             getDelegate().updateTriggerState(conn, trigger.getKey(),                     STATE_ERROR);         } catch (SQLException sqle) {             getLog().error("Unable to set trigger state to ERROR.", sqle);         }         throw jpe;     }       if (trigger.getCalendarName() != null) {         cal = retrieveCalendar(conn, trigger.getCalendarName());         if (cal == null) { return null; }     }       try {         getDelegate().updateFiredTrigger(conn, trigger, STATE_EXECUTING, job);     } catch (SQLException e) {         throw new JobPersistenceException("Couldn't insert fired trigger: "                 + e.getMessage(), e);     }       Date prevFireTime = trigger.getPreviousFireTime();       // call triggered - to update the trigger's next-fire-time state...     trigger.triggered(cal);       String state = STATE_WAITING;     boolean force = true;           if (job.isConcurrentExectionDisallowed()) {         state = STATE_BLOCKED;         force = false;         try {             getDelegate().updateTriggerStatesForJobFromOtherState(conn, job.getKey(),                     STATE_BLOCKED, STATE_WAITING);             getDelegate().updateTriggerStatesForJobFromOtherState(conn, job.getKey(),                     STATE_BLOCKED, STATE_ACQUIRED);             getDelegate().updateTriggerStatesForJobFromOtherState(conn, job.getKey(),                     STATE_PAUSED_BLOCKED, STATE_PAUSED);         } catch (SQLException e) {             throw new JobPersistenceException(                     "Couldn't update states of blocked triggers: "                             + e.getMessage(), e);         }     }                if (trigger.getNextFireTime() == null) {         state = STATE_COMPLETE;         force = true;     }       storeTrigger(conn, trigger, job, true, state, force, false);       job.getJobDataMap().clearDirtyFlag();       return new TriggerFiredBundle(job, trigger, cal, trigger.getKey().getGroup()             .equals(Scheduler.DEFAULT_RECOVERY_GROUP), new Date(), trigger             .getPreviousFireTime(), prevFireTime, trigger.getNextFireTime()); }

首先查询trigger的状态是否STATE_ACQUIRED状态，如果不是直接返回null；然后通过通过jobKey获取对应的jobDetail，更新对应的FiredTrigger为EXECUTING状态；最后判定job的DisallowConcurrentExecution是否开启，如果开启了不能并发执行job，那么trigger的状态为STATE_BLOCKED状态，否则为STATE_WAITING；如果状态为STATE_BLOCKED，那么下次调度
对应的trigger不会被拉取，只有等对应的job执行完之后，更新状态为STATE_WAITING之后才可以执行，保证了job的串行；

6.执行job

通过ThreadPool来执行封装job的JobRunShell；

问题解释

在文章Spring整合Quartz分布式调度中，最后做了几次测试分布式调度，现在可以做出相应的解释

1.同一trigger同一时间只会在一个节点执行

上文中可以发现Quartz使用了分布式锁和状态来保证只有一个节点能执行；

2.任务没有执行完，可以重新开始

因为调度线程和任务执行线程是分开的，认为执行在Threadpool中执行，互相不影响；

3.通过DisallowConcurrentExecution注解保证任务的串行

在triggerFired中如果使用了DisallowConcurrentExecution，会引入STATE_BLOCKED状态，保证任务的串行；

总结

本文从源码的角度大致介绍了一下Quartz调度的流程，当然太细节的东西没有去深入；通过本文大致可以对多节点调度产生的现象做一个合理的解释。