一、Quartz概念

1. 强大的调度功能，例如支持丰富多样的调度方法，可以满足各种常规及特殊需求；
2. 负载均衡
3. 高可用

调度器：scheduler

任务：job

触发器：trigger

• SimpleTrigger：简单触发器，从某个时间开始，每隔多少时间触发，重复多少次。
• CronTrigger：使用cron表达式定义触发的时间规则，如"0 0 0,2,4 1/1 * ? *" 表示每天的0，2，4点触发。
• DailyTimeIntervalTrigger：每天中的一个时间段，每N个时间单元触发，时间单元可以是毫秒，秒，分，小时
• CalendarIntervalTrigger：每N个时间单元触发，时间单元可以是毫秒，秒，分，小时，日，月，年。

未正常触发的任务：misfire job

1. 触发时间在应用不可用的时间内，比如重启
2. 上次的执行时间过长，超过了下次触发的时间
3. 任务被暂停一段时间后，重新被调度的时间在下次触发时间之后

之间的关系：

• scheduler由工厂类SchedulerFactory创建，主要负责job和trigger的持久化管理，包括新增、删除、修改、触发、暂停、恢复调度、停止调度等；
• 一个job可以关联多个trigger，但是一个trigger只能关联一个job。

二、集群模式

spring配置

<bean id="quartzJobFactory" class="com.taobao.trip.hcs.service.gubei.job.AutowiringSpringBeanJobFactory"><property name="ignoredUnknownProperties" value="applicationContext"/>
</bean><!-- Scheduler Factory -->
<bean id="quartzScheduler"class="org.springframework.scheduling.quartz.SchedulerFactoryBean"><property name="jobFactory" ref="quartzJobFactory"/><!-- 数据源 --><property name="dataSource" ref="dataSource" /><!-- 配置项--><property name="quartzProperties"><props><prop key="org.quartz.scheduler.instanceName">${quartz.scheduler.instanceName}}</prop><prop key="org.quartz.scheduler.instanceId">hcs-crawl</prop><!-- 执行任务的线程池配置 --><prop key="org.quartz.threadPool.class">org.quartz.simpl.SimpleThreadPool</prop><prop key="org.quartz.threadPool.threadCount">20</prop><prop key="org.quartz.threadPool.threadPriority">5</prop><!-- JobStore config --><!--单机模式：org.quartz.simpl.RAMJobStore --><!--集群模式：org.quartz.impl.jdbcjobstore.JobStoreTX, org.quartz.impl.jdbcjobstore.JobStoreCMT --><prop key="org.quartz.jobStore.class">org.quartz.impl.jdbcjobstore.JobStoreTX</prop><prop key="org.quartz.jobStore.misfireThreshold">120000</prop><prop key="org.quartz.jobStore.tablePrefix">QRTZ_</prop></props></property><!-- scheduler名称，同一个集群需要一致 --><property name="schedulerName" value="${quartz.scheduler.schedulerName}" /><!-- 延迟启动 --><property name="startupDelay" value="30" /><property name="applicationContextSchedulerContextKey" value="applicationContextKey" /><property name="overwriteExistingJobs" value="true" /><property name="autoStartup" value="true" /><property name="triggers"><list><ref bean="cronTrigger" /><ref bean="simpleTrigger" /></list></property><property name="jobDetails"><list><ref bean="jobDetail" /></list></property>
</bean><bean id="jobDetail" class="org.springframework.scheduling.quartz.JobDetailFactoryBean"><property name="name" value="testJob" /><property name="jobClass" value="com.taobao.trip.hcs.service.common.UniversalActionJob"/></bean>
<bean id="cronTrigger" class="org.springframework.scheduling.quartz.CronTriggerFactoryBean"><property name="jobDetail" ref="jobDetail1" /><property name="cronExpression" value="0/10 * * * * ?" />
</bean><bean id="simpleTrigger" class="org.springframework.scheduling.quartz.SimpleTriggerFactoryBean"><property name="repeatInterval" value="30000" /><property name="repeatCount" value="1000" /><property name="jobDetail" ref="jobDetail1" />
</bean>

public class AutowiringSpringBeanJobFactoryextends SpringBeanJobFactory implements ApplicationContextAware {private transient AutowireCapableBeanFactory beanFactory;public void setApplicationContext(final ApplicationContext context) {beanFactory = context.getAutowireCapableBeanFactory();}@Overridepublic Object createJobInstance(final TriggerFiredBundle bundle) throws Exception {final Object job = super.createJobInstance(bundle);beanFactory.autowireBean(job);  //the magic is done herereturn job;}
}

三、集群模式原理分析

数据库表：

线程

源码分析

@Override
public void run() {boolean lastAcquireFailed = false;while (!halted.get()) {int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();if(availThreadCount > 0) {...// 获取时间idleWaitTime（默认30s）内一定数量的trigger，距下次触发时间不超过TimeWindow（默认0）// 同时修改trigger的状态，从WAITING更新为ACQUIREDtriggers = qsRsrcs.getJobStore().acquireNextTriggers(now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());...// 这里修改trigger的状态，从ACQUIRED改为EXECUTING，更新trigger下次触发时间// 再把状态改为WAITING或者BLOCKED或者COMPLETEList<TriggerFiredResult> res = qsRsrcs.getJobStore().triggersFired(triggers);...// 创建具体执行任务的JobRunShellshell = qsRsrcs.getJobRunShellFactory().createJobRunShell(bndle);...// 提交给线程池执行任务qsRsrcs.getThreadPool().runInThread(shell)}} // while (!halted)// drop references to scheduler stuff to aid garbage collection...qs = null;qsRsrcs = null;
}

public void run() {...// 从上下文获取任务Job job = jec.getJobInstance();...// 任务实际执行job.execute(jec);...// 执行结束，修改trigger状态qs.notifyJobStoreJobComplete(trigger, jobDetail, instCode);
}

<!-- JobStore config -->
<!--单机模式：org.quartz.simpl.RAMJobStore -->
<!--集群模式：org.quartz.impl.jdbcjobstore.JobStoreTX, org.quartz.impl.jdbcjobstore.JobStoreCMT -->
<prop key="org.quartz.jobStore.class">org.quartz.impl.jdbcjobstore.JobStoreTX</prop>

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>() {@Overridepublic 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);}}
}

public boolean obtainLock(Connection conn, String lockName)throws LockException {if(log.isDebugEnabled()) {log.debug("Lock '" + lockName + "' is desired by: "+ Thread.currentThread().getName());}if (!isLockOwner(lockName)) {executeSQL(conn, lockName, expandedSQL, expandedInsertSQL);if(log.isDebugEnabled()) {log.debug("Lock '" + lockName + "' given to: "+ Thread.currentThread().getName());}getThreadLocks().add(lockName);//getThreadLocksObtainer().put(lockName, new// Exception("Obtainer..."));} else if(log.isDebugEnabled()) {log.debug("Lock '" + lockName + "' Is already owned by: "+ Thread.currentThread().getName());}return true;
}

四、参考资料