1.简介
Quartz 是一个开源的作业调度框架,它完全由 Java 写成,并设计用于 J2SE 和 J2EE 应用中。它提供了巨大的灵活性而不牺牲简单性。你能够用它来为执行一个作业而创建简单的或复杂的调度。
2.原理
参考http://ssuupv.blog.163.com/blog/static/146156722013829111028966/
3.以内存运行信息方式分析源码
3.1 测试代码
public static void main(String[] args) { SimpleDateFormat DateFormat = new SimpleDateFormat("yyyyMMddHHmmss"); Date d = new Date(); String returnstr = DateFormat.format(d); try { System.out.println(returnstr + "【系统启动】"); QuartzManager.addJob("11", TestJob.class, "0/2 * * * * ?"); //每2秒钟执行一次 } catch (Exception e) { e.printStackTrace(); } try { System.out.println("【系统启动】"); QuartzManager.start(); } catch (Exception e) { e.printStackTrace(); } }
QuartzManager:
/** */ /** * 添加一个定时任务,使用默认的任务组名,触发器名,触发器组名 * @param jobName 任务名 * @param job 任务 * @param time 时间设置,参考quartz说明文档 * @throws SchedulerException * @throws ParseException */ @SuppressWarnings("deprecation") public static void addJob(String jobName, Class<? extends Job> jobClazz, String time) throws SchedulerException, ParseException { Scheduler sched = sf.getScheduler(); JobDetail jobDetail = new JobDetailImpl(jobName, JOB_GROUP_NAME, jobClazz);//任务名,任务组,任务执行类 //触发器 CronTriggerImpl trigger = new CronTriggerImpl(jobName, TRIGGER_GROUP_NAME);//触发器名,触发器组 trigger.setCronExpression(time);//触发器时间设定 sched.scheduleJob(jobDetail, trigger); //启动 if (!sched.isShutdown()) sched.start(); } public static void start() throws SchedulerException, ParseException { Scheduler sched = sf.getScheduler(); //启动 if (!sched.isShutdown()) sched.start(); }
TestJob:
public class TestJob implements Job { private final static SimpleDateFormat DateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); public void execute(JobExecutionContext context) throws JobExecutionException { System.out.println(DateFormat.format(new Date()) + "★★★★★★★★★★★"); } }
3.2代码分析
3.2.1首先通过SchedulerFactory得到StdScheduler
private Scheduler instantiate() throws SchedulerException { if (cfg == null) { initialize(); } if (initException != null) { throw initException; } JobStore js = null; ThreadPool tp = null; QuartzScheduler qs = null; DBConnectionManager dbMgr = null; String instanceIdGeneratorClass = null; Properties tProps = null; String userTXLocation = null; boolean wrapJobInTx = false; boolean autoId = false; long idleWaitTime = -1; long dbFailureRetry = -1; String classLoadHelperClass; String jobFactoryClass; ThreadExecutor threadExecutor; SchedulerRepository schedRep = SchedulerRepository.getInstance(); // Get Scheduler Properties // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ String schedName = cfg.getStringProperty(PROP_SCHED_INSTANCE_NAME, "QuartzScheduler"); String threadName = cfg.getStringProperty(PROP_SCHED_THREAD_NAME, schedName + "_QuartzSchedulerThread"); String schedInstId = cfg.getStringProperty(PROP_SCHED_INSTANCE_ID, DEFAULT_INSTANCE_ID); if (schedInstId.equals(AUTO_GENERATE_INSTANCE_ID)) { autoId = true; instanceIdGeneratorClass = cfg.getStringProperty( PROP_SCHED_INSTANCE_ID_GENERATOR_CLASS, "org.quartz.simpl.SimpleInstanceIdGenerator"); } else if (schedInstId.equals(SYSTEM_PROPERTY_AS_INSTANCE_ID)) { autoId = true; instanceIdGeneratorClass = "org.quartz.simpl.SystemPropertyInstanceIdGenerator"; } userTXLocation = cfg.getStringProperty(PROP_SCHED_USER_TX_URL, userTXLocation); if (userTXLocation != null && userTXLocation.trim().length() == 0) { userTXLocation = null; } classLoadHelperClass = cfg.getStringProperty( PROP_SCHED_CLASS_LOAD_HELPER_CLASS, "org.quartz.simpl.CascadingClassLoadHelper"); wrapJobInTx = cfg.getBooleanProperty(PROP_SCHED_WRAP_JOB_IN_USER_TX, wrapJobInTx); jobFactoryClass = cfg.getStringProperty( PROP_SCHED_JOB_FACTORY_CLASS, null); idleWaitTime = cfg.getLongProperty(PROP_SCHED_IDLE_WAIT_TIME, idleWaitTime); dbFailureRetry = cfg.getLongProperty( PROP_SCHED_DB_FAILURE_RETRY_INTERVAL, dbFailureRetry); boolean makeSchedulerThreadDaemon = cfg.getBooleanProperty(PROP_SCHED_MAKE_SCHEDULER_THREAD_DAEMON); boolean threadsInheritInitalizersClassLoader = cfg.getBooleanProperty(PROP_SCHED_SCHEDULER_THREADS_INHERIT_CONTEXT_CLASS_LOADER_OF_INITIALIZING_THREAD); boolean skipUpdateCheck = cfg.getBooleanProperty(PROP_SCHED_SKIP_UPDATE_CHECK, false); long batchTimeWindow = cfg.getLongProperty(PROP_SCHED_BATCH_TIME_WINDOW, 0L); int maxBatchSize = cfg.getIntProperty(PROP_SCHED_MAX_BATCH_SIZE, 1); boolean interruptJobsOnShutdown = cfg.getBooleanProperty(PROP_SCHED_INTERRUPT_JOBS_ON_SHUTDOWN, false); boolean interruptJobsOnShutdownWithWait = cfg.getBooleanProperty(PROP_SCHED_INTERRUPT_JOBS_ON_SHUTDOWN_WITH_WAIT, false); boolean jmxExport = cfg.getBooleanProperty(PROP_SCHED_JMX_EXPORT); String jmxObjectName = cfg.getStringProperty(PROP_SCHED_JMX_OBJECT_NAME); boolean jmxProxy = cfg.getBooleanProperty(PROP_SCHED_JMX_PROXY); String jmxProxyClass = cfg.getStringProperty(PROP_SCHED_JMX_PROXY_CLASS); boolean rmiExport = cfg.getBooleanProperty(PROP_SCHED_RMI_EXPORT, false); boolean rmiProxy = cfg.getBooleanProperty(PROP_SCHED_RMI_PROXY, false); String rmiHost = cfg.getStringProperty(PROP_SCHED_RMI_HOST, "localhost"); int rmiPort = cfg.getIntProperty(PROP_SCHED_RMI_PORT, 1099); int rmiServerPort = cfg.getIntProperty(PROP_SCHED_RMI_SERVER_PORT, -1); String rmiCreateRegistry = cfg.getStringProperty( PROP_SCHED_RMI_CREATE_REGISTRY, QuartzSchedulerResources.CREATE_REGISTRY_NEVER); String rmiBindName = cfg.getStringProperty(PROP_SCHED_RMI_BIND_NAME); if (jmxProxy && rmiProxy) { throw new SchedulerConfigException("Cannot proxy both RMI and JMX."); } Properties schedCtxtProps = cfg.getPropertyGroup(PROP_SCHED_CONTEXT_PREFIX, true); // If Proxying to remote scheduler, short-circuit here... // ~~~~~~~~~~~~~~~~~~ if (rmiProxy) { if (autoId) { schedInstId = DEFAULT_INSTANCE_ID; } String uid = (rmiBindName == null) ? QuartzSchedulerResources.getUniqueIdentifier( schedName, schedInstId) : rmiBindName; RemoteScheduler remoteScheduler = new RemoteScheduler(uid, rmiHost, rmiPort); schedRep.bind(remoteScheduler); return remoteScheduler; } // Create class load helper ClassLoadHelper loadHelper = null; try { loadHelper = (ClassLoadHelper) loadClass(classLoadHelperClass) .newInstance(); } catch (Exception e) { throw new SchedulerConfigException( "Unable to instantiate class load helper class: " + e.getMessage(), e); } loadHelper.initialize(); // If Proxying to remote JMX scheduler, short-circuit here... // ~~~~~~~~~~~~~~~~~~ if (jmxProxy) { if (autoId) { schedInstId = DEFAULT_INSTANCE_ID; } if (jmxProxyClass == null) { throw new SchedulerConfigException("No JMX Proxy Scheduler class provided"); } RemoteMBeanScheduler jmxScheduler = null; try { jmxScheduler = (RemoteMBeanScheduler)loadHelper.loadClass(jmxProxyClass) .newInstance(); } catch (Exception e) { throw new SchedulerConfigException( "Unable to instantiate RemoteMBeanScheduler class.", e); } if (jmxObjectName == null) { jmxObjectName = QuartzSchedulerResources.generateJMXObjectName(schedName, schedInstId); } jmxScheduler.setSchedulerObjectName(jmxObjectName); tProps = cfg.getPropertyGroup(PROP_SCHED_JMX_PROXY, true); try { setBeanProps(jmxScheduler, tProps); } catch (Exception e) { initException = new SchedulerException("RemoteMBeanScheduler class '" + jmxProxyClass + "' props could not be configured.", e); throw initException; } jmxScheduler.initialize(); schedRep.bind(jmxScheduler); return jmxScheduler; } JobFactory jobFactory = null; if(jobFactoryClass != null) { try { jobFactory = (JobFactory) loadHelper.loadClass(jobFactoryClass) .newInstance(); } catch (Exception e) { throw new SchedulerConfigException( "Unable to instantiate JobFactory class: " + e.getMessage(), e); } tProps = cfg.getPropertyGroup(PROP_SCHED_JOB_FACTORY_PREFIX, true); try { setBeanProps(jobFactory, tProps); } catch (Exception e) { initException = new SchedulerException("JobFactory class '" + jobFactoryClass + "' props could not be configured.", e); throw initException; } } InstanceIdGenerator instanceIdGenerator = null; if(instanceIdGeneratorClass != null) { try { instanceIdGenerator = (InstanceIdGenerator) loadHelper.loadClass(instanceIdGeneratorClass) .newInstance(); } catch (Exception e) { throw new SchedulerConfigException( "Unable to instantiate InstanceIdGenerator class: " + e.getMessage(), e); } tProps = cfg.getPropertyGroup(PROP_SCHED_INSTANCE_ID_GENERATOR_PREFIX, true); try { setBeanProps(instanceIdGenerator, tProps); } catch (Exception e) { initException = new SchedulerException("InstanceIdGenerator class '" + instanceIdGeneratorClass + "' props could not be configured.", e); throw initException; } } // Get ThreadPool Properties // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ String tpClass = cfg.getStringProperty(PROP_THREAD_POOL_CLASS, SimpleThreadPool.class.getName()); if (tpClass == null) { initException = new SchedulerException( "ThreadPool class not specified. "); throw initException; } try { //得到ThreadPool tp = (ThreadPool) loadHelper.loadClass(tpClass).newInstance(); } catch (Exception e) { initException = new SchedulerException("ThreadPool class '" + tpClass + "' could not be instantiated.", e); throw initException; } tProps = cfg.getPropertyGroup(PROP_THREAD_POOL_PREFIX, true); try { setBeanProps(tp, tProps); } catch (Exception e) { initException = new SchedulerException("ThreadPool class '" + tpClass + "' props could not be configured.", e); throw initException; } // Get JobStore Properties // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ String jsClass = cfg.getStringProperty(PROP_JOB_STORE_CLASS, RAMJobStore.class.getName()); if (jsClass == null) { initException = new SchedulerException( "JobStore class not specified. "); throw initException; } try { js = (JobStore) loadHelper.loadClass(jsClass).newInstance(); } catch (Exception e) { initException = new SchedulerException("JobStore class '" + jsClass + "' could not be instantiated.", e); throw initException; } SchedulerDetailsSetter.setDetails(js, schedName, schedInstId); tProps = cfg.getPropertyGroup(PROP_JOB_STORE_PREFIX, true, new String[] {PROP_JOB_STORE_LOCK_HANDLER_PREFIX}); try { setBeanProps(js, tProps); } catch (Exception e) { initException = new SchedulerException("JobStore class '" + jsClass + "' props could not be configured.", e); throw initException; } if (js instanceof JobStoreSupport) { // Install custom lock handler (Semaphore) String lockHandlerClass = cfg.getStringProperty(PROP_JOB_STORE_LOCK_HANDLER_CLASS); if (lockHandlerClass != null) { try { Semaphore lockHandler = (Semaphore)loadHelper.loadClass(lockHandlerClass).newInstance(); tProps = cfg.getPropertyGroup(PROP_JOB_STORE_LOCK_HANDLER_PREFIX, true); // If this lock handler requires the table prefix, add it to its properties. if (lockHandler instanceof TablePrefixAware) { tProps.setProperty( PROP_TABLE_PREFIX, ((JobStoreSupport)js).getTablePrefix()); tProps.setProperty( PROP_SCHED_NAME, schedName); } try { setBeanProps(lockHandler, tProps); } catch (Exception e) { initException = new SchedulerException("JobStore LockHandler class '" + lockHandlerClass + "' props could not be configured.", e); throw initException; } ((JobStoreSupport)js).setLockHandler(lockHandler); getLog().info("Using custom data access locking (synchronization): " + lockHandlerClass); } catch (Exception e) { initException = new SchedulerException("JobStore LockHandler class '" + lockHandlerClass + "' could not be instantiated.", e); throw initException; } } } // Set up any DataSources // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ String[] dsNames = cfg.getPropertyGroups(PROP_DATASOURCE_PREFIX); for (int i = 0; i < dsNames.length; i++) { PropertiesParser pp = new PropertiesParser(cfg.getPropertyGroup( PROP_DATASOURCE_PREFIX + "." + dsNames[i], true)); String cpClass = pp.getStringProperty(PROP_CONNECTION_PROVIDER_CLASS, null); // custom connectionProvider... if(cpClass != null) { ConnectionProvider cp = null; try { cp = (ConnectionProvider) loadHelper.loadClass(cpClass).newInstance(); } catch (Exception e) { initException = new SchedulerException("ConnectionProvider class '" + cpClass + "' could not be instantiated.", e); throw initException; } try { // remove the class name, so it isn't attempted to be set pp.getUnderlyingProperties().remove( PROP_CONNECTION_PROVIDER_CLASS); setBeanProps(cp, pp.getUnderlyingProperties()); } catch (Exception e) { initException = new SchedulerException("ConnectionProvider class '" + cpClass + "' props could not be configured.", e); throw initException; } dbMgr = DBConnectionManager.getInstance(); dbMgr.addConnectionProvider(dsNames[i], cp); } else { String dsJndi = pp.getStringProperty(PROP_DATASOURCE_JNDI_URL, null); if (dsJndi != null) { boolean dsAlwaysLookup = pp.getBooleanProperty( PROP_DATASOURCE_JNDI_ALWAYS_LOOKUP); String dsJndiInitial = pp.getStringProperty( PROP_DATASOURCE_JNDI_INITIAL); String dsJndiProvider = pp.getStringProperty( PROP_DATASOURCE_JNDI_PROVDER); String dsJndiPrincipal = pp.getStringProperty( PROP_DATASOURCE_JNDI_PRINCIPAL); String dsJndiCredentials = pp.getStringProperty( PROP_DATASOURCE_JNDI_CREDENTIALS); Properties props = null; if (null != dsJndiInitial || null != dsJndiProvider || null != dsJndiPrincipal || null != dsJndiCredentials) { props = new Properties(); if (dsJndiInitial != null) { props.put(PROP_DATASOURCE_JNDI_INITIAL, dsJndiInitial); } if (dsJndiProvider != null) { props.put(PROP_DATASOURCE_JNDI_PROVDER, dsJndiProvider); } if (dsJndiPrincipal != null) { props.put(PROP_DATASOURCE_JNDI_PRINCIPAL, dsJndiPrincipal); } if (dsJndiCredentials != null) { props.put(PROP_DATASOURCE_JNDI_CREDENTIALS, dsJndiCredentials); } } JNDIConnectionProvider cp = new JNDIConnectionProvider(dsJndi, props, dsAlwaysLookup); dbMgr = DBConnectionManager.getInstance(); dbMgr.addConnectionProvider(dsNames[i], cp); } else { String dsDriver = pp.getStringProperty(PROP_DATASOURCE_DRIVER); String dsURL = pp.getStringProperty(PROP_DATASOURCE_URL); String dsUser = pp.getStringProperty(PROP_DATASOURCE_USER, ""); String dsPass = pp.getStringProperty(PROP_DATASOURCE_PASSWORD, ""); int dsCnt = pp.getIntProperty(PROP_DATASOURCE_MAX_CONNECTIONS, 10); String dsValidation = pp.getStringProperty(PROP_DATASOURCE_VALIDATION_QUERY); if (dsDriver == null) { initException = new SchedulerException( "Driver not specified for DataSource: " + dsNames[i]); throw initException; } if (dsURL == null) { initException = new SchedulerException( "DB URL not specified for DataSource: " + dsNames[i]); throw initException; } try { PoolingConnectionProvider cp = new PoolingConnectionProvider( dsDriver, dsURL, dsUser, dsPass, dsCnt, dsValidation); dbMgr = DBConnectionManager.getInstance(); dbMgr.addConnectionProvider(dsNames[i], cp); } catch (SQLException sqle) { initException = new SchedulerException( "Could not initialize DataSource: " + dsNames[i], sqle); throw initException; } } } } // Set up any SchedulerPlugins // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ String[] pluginNames = cfg.getPropertyGroups(PROP_PLUGIN_PREFIX); SchedulerPlugin[] plugins = new SchedulerPlugin[pluginNames.length]; for (int i = 0; i < pluginNames.length; i++) { Properties pp = cfg.getPropertyGroup(PROP_PLUGIN_PREFIX + "." + pluginNames[i], true); String plugInClass = pp.getProperty(PROP_PLUGIN_CLASS, null); if (plugInClass == null) { initException = new SchedulerException( "SchedulerPlugin class not specified for plugin '" + pluginNames[i] + "'"); throw initException; } SchedulerPlugin plugin = null; try { plugin = (SchedulerPlugin) loadHelper.loadClass(plugInClass).newInstance(); } catch (Exception e) { initException = new SchedulerException( "SchedulerPlugin class '" + plugInClass + "' could not be instantiated.", e); throw initException; } try { setBeanProps(plugin, pp); } catch (Exception e) { initException = new SchedulerException( "JobStore SchedulerPlugin '" + plugInClass + "' props could not be configured.", e); throw initException; } plugins[i] = plugin; } // Set up any JobListeners // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Class<?>[] strArg = new Class[] { String.class }; String[] jobListenerNames = cfg.getPropertyGroups(PROP_JOB_LISTENER_PREFIX); JobListener[] jobListeners = new JobListener[jobListenerNames.length]; for (int i = 0; i < jobListenerNames.length; i++) { Properties lp = cfg.getPropertyGroup(PROP_JOB_LISTENER_PREFIX + "." + jobListenerNames[i], true); String listenerClass = lp.getProperty(PROP_LISTENER_CLASS, null); if (listenerClass == null) { initException = new SchedulerException( "JobListener class not specified for listener '" + jobListenerNames[i] + "'"); throw initException; } JobListener listener = null; try { listener = (JobListener) loadHelper.loadClass(listenerClass).newInstance(); } catch (Exception e) { initException = new SchedulerException( "JobListener class '" + listenerClass + "' could not be instantiated.", e); throw initException; } try { Method nameSetter = listener.getClass().getMethod("setName", strArg); if(nameSetter != null) { nameSetter.invoke(listener, new Object[] {jobListenerNames[i] } ); } setBeanProps(listener, lp); } catch (Exception e) { initException = new SchedulerException( "JobListener '" + listenerClass + "' props could not be configured.", e); throw initException; } jobListeners[i] = listener; } // Set up any TriggerListeners // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ String[] triggerListenerNames = cfg.getPropertyGroups(PROP_TRIGGER_LISTENER_PREFIX); TriggerListener[] triggerListeners = new TriggerListener[triggerListenerNames.length]; for (int i = 0; i < triggerListenerNames.length; i++) { Properties lp = cfg.getPropertyGroup(PROP_TRIGGER_LISTENER_PREFIX + "." + triggerListenerNames[i], true); String listenerClass = lp.getProperty(PROP_LISTENER_CLASS, null); if (listenerClass == null) { initException = new SchedulerException( "TriggerListener class not specified for listener '" + triggerListenerNames[i] + "'"); throw initException; } TriggerListener listener = null; try { listener = (TriggerListener) loadHelper.loadClass(listenerClass).newInstance(); } catch (Exception e) { initException = new SchedulerException( "TriggerListener class '" + listenerClass + "' could not be instantiated.", e); throw initException; } try { Method nameSetter = listener.getClass().getMethod("setName", strArg); if(nameSetter != null) { nameSetter.invoke(listener, new Object[] {triggerListenerNames[i] } ); } setBeanProps(listener, lp); } catch (Exception e) { initException = new SchedulerException( "TriggerListener '" + listenerClass + "' props could not be configured.", e); throw initException; } triggerListeners[i] = listener; } boolean tpInited = false; boolean qsInited = false; // Get ThreadExecutor Properties // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ String threadExecutorClass = cfg.getStringProperty(PROP_THREAD_EXECUTOR_CLASS); if (threadExecutorClass != null) { tProps = cfg.getPropertyGroup(PROP_THREAD_EXECUTOR, true); try { threadExecutor = (ThreadExecutor) loadHelper.loadClass(threadExecutorClass).newInstance(); log.info("Using custom implementation for ThreadExecutor: " + threadExecutorClass); setBeanProps(threadExecutor, tProps); } catch (Exception e) { initException = new SchedulerException( "ThreadExecutor class '" + threadExecutorClass + "' could not be instantiated.", e); throw initException; } } else { log.info("Using default implementation for ThreadExecutor"); threadExecutor = new DefaultThreadExecutor(); } // Fire everything up // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ try { JobRunShellFactory jrsf = null; // Create correct run-shell factory... if (userTXLocation != null) { UserTransactionHelper.setUserTxLocation(userTXLocation); } if (wrapJobInTx) { jrsf = new JTAJobRunShellFactory(); } else { jrsf = new JTAAnnotationAwareJobRunShellFactory(); } if (autoId) { try { schedInstId = DEFAULT_INSTANCE_ID; if (js.isClustered()) { schedInstId = instanceIdGenerator.generateInstanceId(); } } catch (Exception e) { getLog().error("Couldn't generate instance Id!", e); throw new IllegalStateException("Cannot run without an instance id."); } } if (js.getClass().getName().startsWith("org.terracotta.quartz")) { try { String uuid = (String) js.getClass().getMethod("getUUID").invoke(js); if(schedInstId.equals(DEFAULT_INSTANCE_ID)) { schedInstId = "TERRACOTTA_CLUSTERED,node=" + uuid; if (jmxObjectName == null) { jmxObjectName = QuartzSchedulerResources.generateJMXObjectName(schedName, schedInstId); } } else if(jmxObjectName == null) { jmxObjectName = QuartzSchedulerResources.generateJMXObjectName(schedName, schedInstId + ",node=" + uuid); } } catch(Exception e) { throw new RuntimeException("Problem obtaining node id from TerracottaJobStore.", e); } if(null == cfg.getStringProperty(PROP_SCHED_JMX_EXPORT)) { jmxExport = true; } } if (js instanceof JobStoreSupport) { JobStoreSupport jjs = (JobStoreSupport)js; jjs.setDbRetryInterval(dbFailureRetry); if(threadsInheritInitalizersClassLoader) jjs.setThreadsInheritInitializersClassLoadContext(threadsInheritInitalizersClassLoader); } QuartzSchedulerResources rsrcs = new QuartzSchedulerResources(); rsrcs.setName(schedName); rsrcs.setThreadName(threadName); rsrcs.setInstanceId(schedInstId); rsrcs.setJobRunShellFactory(jrsf); rsrcs.setMakeSchedulerThreadDaemon(makeSchedulerThreadDaemon); rsrcs.setThreadsInheritInitializersClassLoadContext(threadsInheritInitalizersClassLoader); rsrcs.setRunUpdateCheck(!skipUpdateCheck); rsrcs.setBatchTimeWindow(batchTimeWindow); rsrcs.setMaxBatchSize(maxBatchSize); rsrcs.setInterruptJobsOnShutdown(interruptJobsOnShutdown); rsrcs.setInterruptJobsOnShutdownWithWait(interruptJobsOnShutdownWithWait); rsrcs.setJMXExport(jmxExport); rsrcs.setJMXObjectName(jmxObjectName); if (rmiExport) { rsrcs.setRMIRegistryHost(rmiHost); rsrcs.setRMIRegistryPort(rmiPort); rsrcs.setRMIServerPort(rmiServerPort); rsrcs.setRMICreateRegistryStrategy(rmiCreateRegistry); rsrcs.setRMIBindName(rmiBindName); } SchedulerDetailsSetter.setDetails(tp, schedName, schedInstId); rsrcs.setThreadExecutor(threadExecutor); threadExecutor.initialize(); rsrcs.setThreadPool(tp); if(tp instanceof SimpleThreadPool) { ((SimpleThreadPool)tp).setThreadNamePrefix(schedName + "_Worker"); if(threadsInheritInitalizersClassLoader) ((SimpleThreadPool)tp).setThreadsInheritContextClassLoaderOfInitializingThread(threadsInheritInitalizersClassLoader); } //ThreadPool初始化 tp.initialize(); tpInited = true; rsrcs.setJobStore(js); // add plugins for (int i = 0; i < plugins.length; i++) { rsrcs.addSchedulerPlugin(plugins[i]); } //得到QuartzSchedulerThread 默认paused=true qs = new QuartzScheduler(rsrcs, idleWaitTime, dbFailureRetry); qsInited = true; // Create Scheduler ref... Scheduler scheduler = instantiate(rsrcs, qs); // set job factory if specified if(jobFactory != null) { qs.setJobFactory(jobFactory); } // Initialize plugins now that we have a Scheduler instance. for (int i = 0; i < plugins.length; i++) { plugins[i].initialize(pluginNames[i], scheduler); } // add listeners for (int i = 0; i < jobListeners.length; i++) { qs.getListenerManager().addJobListener(jobListeners[i], EverythingMatcher.allJobs()); } for (int i = 0; i < triggerListeners.length; i++) { qs.getListenerManager().addTriggerListener(triggerListeners[i], EverythingMatcher.allTriggers()); } // set scheduler context data... for(Object key: schedCtxtProps.keySet()) { String val = schedCtxtProps.getProperty((String) key); scheduler.getContext().put(key, val); } // fire up job store, and runshell factory js.setInstanceId(schedInstId); js.setInstanceName(schedName); js.initialize(loadHelper, qs.getSchedulerSignaler()); js.setThreadPoolSize(tp.getPoolSize()); jrsf.initialize(scheduler); qs.initialize(); getLog().info( "Quartz scheduler '" + scheduler.getSchedulerName() + "' initialized from " + propSrc); getLog().info("Quartz scheduler version: " + qs.getVersion()); // prevents the repository from being garbage collected qs.addNoGCObject(schedRep); // prevents the db manager from being garbage collected if (dbMgr != null) { qs.addNoGCObject(dbMgr); } schedRep.bind(scheduler); return scheduler; } catch(SchedulerException e) { if(qsInited) qs.shutdown(false); else if(tpInited) tp.shutdown(false); throw e; } catch(RuntimeException re) { if(qsInited) qs.shutdown(false); else if(tpInited) tp.shutdown(false); throw re; } catch(Error re) { if(qsInited) qs.shutdown(false); else if(tpInited) tp.shutdown(false); throw re; } }
3.2在QuartzSchedulerThread线程中循环通过JobStore取得List<OperableTrigger>后通过ThreadPool执行JobRunShell。
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 { //通过JobStore取得List<OperableTrigger> 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; } catch (RuntimeException e) { if(!lastAcquireFailed) { getLog().error("quartzSchedulerThreadLoop: RuntimeException " +e.getMessage(), e); } lastAcquireFailed = true; } 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 { //告知JobStore开始执行trigger,内存则移除,表则更新qrtz_triggers信息 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); } } 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); // db connection must have failed... keep // retrying until it's up... releaseTriggerRetryLoop(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) { try { qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i)); } catch (SchedulerException se) { qs.notifySchedulerListenersError( "An error occurred while releasing triggers '" + triggers.get(i).getKey() + "'", se); // db connection must have failed... keep retrying // until it's up... releaseTriggerRetryLoop(triggers.get(i)); } continue; } // TODO: improvements: // // 2- make sure we can get a job runshell before firing triggers, or // don't let that throw an exception (right now it never does, // but the signature says it can). // 3- acquire more triggers at a time (based on num threads available?) //得到JobRunShell JobRunShell shell = null; try { shell = qsRsrcs.getJobRunShellFactory().createJobRunShell(bndle); //初始化JobRunShell shell.initialize(qs); } catch (SchedulerException se) { try { qsRsrcs.getJobStore().triggeredJobComplete( triggers.get(i), bndle.getJobDetail(), CompletedExecutionInstruction.SET_ALL_JOB_TRIGGERS_ERROR); } catch (SchedulerException se2) { qs.notifySchedulerListenersError( "An error occurred while placing job's triggers in error state '" + triggers.get(i).getKey() + "'", se2); // db connection must have failed... keep retrying // until it's up... errorTriggerRetryLoop(bndle); } continue; } //通过ThreadPool执行JobRunShell if (qsRsrcs.getThreadPool().runInThread(shell) == false) { try { // 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); } catch (SchedulerException se2) { qs.notifySchedulerListenersError( "An error occurred while placing job's triggers in error state '" + triggers.get(i).getKey() + "'", se2); // db connection must have failed... keep retrying // until it's up... releaseTriggerRetryLoop(triggers.get(i)); } } } 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 { 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; }
3.3JobStore得到List<OperableTrigger>
RAMJobStore取得List<OperableTrigger>方法使用hashTree
public List<OperableTrigger> acquireNextTriggers(long noLaterThan, int maxCount, long timeWindow) { synchronized (lock) { List<OperableTrigger> result = new ArrayList<OperableTrigger>(); while (true) { TriggerWrapper tw; try { tw = (TriggerWrapper) timeTriggers.first(); if (tw == null) return result; timeTriggers.remove(tw); } catch (java.util.NoSuchElementException nsee) { return result; } if (tw.trigger.getNextFireTime() == null) { continue; } if (applyMisfire(tw)) { if (tw.trigger.getNextFireTime() != null) { timeTriggers.add(tw); } continue; } if (tw.getTrigger().getNextFireTime().getTime() > noLaterThan + timeWindow) { timeTriggers.add(tw); return result; } tw.state = TriggerWrapper.STATE_ACQUIRED; tw.trigger.setFireInstanceId(getFiredTriggerRecordId()); OperableTrigger trig = (OperableTrigger) tw.trigger.clone(); result.add(trig); if (result.size() == maxCount) return result; } } }
3.4JobRunShell
初始化:
public void initialize(QuartzScheduler qs) throws SchedulerException { this.qs = qs; Job job = null; JobDetail jobDetail = firedTriggerBundle.getJobDetail(); try { job = qs.getJobFactory().newJob(firedTriggerBundle, scheduler); } catch (SchedulerException se) { qs.notifySchedulerListenersError( "An error occured instantiating job to be executed. job= '" + jobDetail.getKey() + "'", se); throw se; } catch (Throwable ncdfe) { // such as NoClassDefFoundError SchedulerException se = new SchedulerException( "Problem instantiating class '" + jobDetail.getJobClass().getName() + "' - ", ncdfe); qs.notifySchedulerListenersError( "An error occured instantiating job to be executed. job= '" + jobDetail.getKey() + "'", se); throw se; } this.jec = new JobExecutionContextImpl(scheduler, firedTriggerBundle, job); }
run方法
public void run() { qs.addInternalSchedulerListener(this); try { OperableTrigger trigger = (OperableTrigger) jec.getTrigger(); JobDetail jobDetail = jec.getJobDetail(); do { JobExecutionException jobExEx = null; Job job = jec.getJobInstance(); try { begin(); } catch (SchedulerException se) { qs.notifySchedulerListenersError("Error executing Job (" + jec.getJobDetail().getKey() + ": couldn't begin execution.", se); break; } // notify job & trigger listeners... try { if (!notifyListenersBeginning(jec)) { break; } } catch(VetoedException ve) { try { CompletedExecutionInstruction instCode = trigger.executionComplete(jec, null); try { qs.notifyJobStoreJobVetoed(trigger, jobDetail, instCode); } catch(JobPersistenceException jpe) { vetoedJobRetryLoop(trigger, jobDetail, instCode); } complete(true); } catch (SchedulerException se) { qs.notifySchedulerListenersError("Error during veto of Job (" + jec.getJobDetail().getKey() + ": couldn't finalize execution.", se); } break; } long startTime = System.currentTimeMillis(); long endTime = startTime; // execute the job try { log.debug("Calling execute on job " + jobDetail.getKey()); job.execute(jec); endTime = System.currentTimeMillis(); } catch (JobExecutionException jee) { endTime = System.currentTimeMillis(); jobExEx = jee; getLog().info("Job " + jobDetail.getKey() + " threw a JobExecutionException: ", jobExEx); } catch (Throwable e) { endTime = System.currentTimeMillis(); getLog().error("Job " + jobDetail.getKey() + " threw an unhandled Exception: ", e); SchedulerException se = new SchedulerException( "Job threw an unhandled exception.", e); qs.notifySchedulerListenersError("Job (" + jec.getJobDetail().getKey() + " threw an exception.", se); jobExEx = new JobExecutionException(se, false); } jec.setJobRunTime(endTime - startTime); // notify all job listeners if (!notifyJobListenersComplete(jec, jobExEx)) { break; } CompletedExecutionInstruction instCode = CompletedExecutionInstruction.NOOP; // update the trigger try { instCode = trigger.executionComplete(jec, jobExEx); } catch (Exception e) { // If this happens, there's a bug in the trigger... SchedulerException se = new SchedulerException( "Trigger threw an unhandled exception.", e); qs.notifySchedulerListenersError( "Please report this error to the Quartz developers.", se); } // notify all trigger listeners if (!notifyTriggerListenersComplete(jec, instCode)) { break; } // update job/trigger or re-execute job if (instCode == CompletedExecutionInstruction.RE_EXECUTE_JOB) { jec.incrementRefireCount(); try { complete(false); } catch (SchedulerException se) { qs.notifySchedulerListenersError("Error executing Job (" + jec.getJobDetail().getKey() + ": couldn't finalize execution.", se); } continue; } try { complete(true); } catch (SchedulerException se) { qs.notifySchedulerListenersError("Error executing Job (" + jec.getJobDetail().getKey() + ": couldn't finalize execution.", se); continue; } try { qs.notifyJobStoreJobComplete(trigger, jobDetail, instCode); } catch (JobPersistenceException jpe) { qs.notifySchedulerListenersError( "An error occured while marking executed job complete. job= '" + jobDetail.getKey() + "'", jpe); if (!completeTriggerRetryLoop(trigger, jobDetail, instCode)) { return; } } break; } while (true); } finally { qs.removeInternalSchedulerListener(this); } }
3.5SimpleThreadPool
初始化方法:
public void initialize() throws SchedulerConfigException { if(workers != null && workers.size() > 0) // already initialized... return; if (count <= 0) { throw new SchedulerConfigException( "Thread count must be > 0"); } if (prio <= 0 || prio > 9) { throw new SchedulerConfigException( "Thread priority must be > 0 and <= 9"); } if(isThreadsInheritGroupOfInitializingThread()) { threadGroup = Thread.currentThread().getThreadGroup(); } else { // follow the threadGroup tree to the root thread group. threadGroup = Thread.currentThread().getThreadGroup(); ThreadGroup parent = threadGroup; while ( !parent.getName().equals("main") ) { threadGroup = parent; parent = threadGroup.getParent(); } threadGroup = new ThreadGroup(parent, schedulerInstanceName + "-SimpleThreadPool"); if (isMakeThreadsDaemons()) { threadGroup.setDaemon(true); } } if (isThreadsInheritContextClassLoaderOfInitializingThread()) { getLog().info( "Job execution threads will use class loader of thread: " + Thread.currentThread().getName()); } // create the worker threads and start them Iterator workerThreads = createWorkerThreads(count).iterator(); while(workerThreads.hasNext()) { WorkerThread wt = (WorkerThread) workerThreads.next(); wt.start(); availWorkers.add(wt); } }
runInThread方法
public boolean runInThread(Runnable runnable) { if (runnable == null) { return false; } synchronized (nextRunnableLock) { handoffPending = true; // Wait until a worker thread is available while ((availWorkers.size() < 1) && !isShutdown) { try { nextRunnableLock.wait(500); } catch (InterruptedException ignore) { } } if (!isShutdown) { WorkerThread wt = (WorkerThread)availWorkers.removeFirst(); busyWorkers.add(wt); wt.run(runnable); } else { // If the thread pool is going down, execute the Runnable // within a new additional worker thread (no thread from the pool). WorkerThread wt = new WorkerThread(this, threadGroup, "WorkerThread-LastJob", prio, isMakeThreadsDaemons(), runnable); busyWorkers.add(wt); workers.add(wt); wt.start(); } nextRunnableLock.notifyAll(); handoffPending = false; } return true; }
4..以mysql存储信息方式分析源码
4.1测试配置文件
#============================================================== #Configure Main Scheduler Properties #============================================================== org.quartz.scheduler.instanceName = quartzScheduler org.quartz.scheduler.instanceId = AUTO #============================================================== #Configure JobStore #============================================================== org.quartz.jobStore.class = org.quartz.impl.jdbcjobstore.JobStoreTX org.quartz.jobStore.driverDelegateClass = org.quartz.impl.jdbcjobstore.StdJDBCDelegate org.quartz.jobStore.tablePrefix = QRTZ_ org.quartz.jobStore.isClustered = true org.quartz.jobStore.clusterCheckinInterval = 20000 org.quartz.jobStore.dataSource = myDS org.quartz.jobStore.lockHandler.class=org.quartz.impl.jdbcjobstore.UpdateLockRowSemaphore #============================================================== #Configure DataSource #============================================================== org.quartz.dataSource.myDS.driver = com.mysql.jdbc.Driver org.quartz.dataSource.myDS.URL = jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf-8 org.quartz.dataSource.myDS.user = root org.quartz.dataSource.myDS.password = password org.quartz.dataSource.myDS.maxConnections = 30 #============================================================== #Configure ThreadPool #============================================================== org.quartz.threadPool.class = org.quartz.simpl.SimpleThreadPool org.quartz.threadPool.threadCount = 10 org.quartz.threadPool.threadPriority = 5 org.quartz.threadPool.threadsInheritContextClassLoaderOfInitializingThread = true4.2告知JobStore开始执行triggers,更新triggers时间,是在QuartzSchedulerThread的方法中的qsRsrcs.getJobStore().triggersFired(triggers)(该方法通过qrtz_locks表锁来控制并发)
JobStoreSupport的triggersFired方法
public List<TriggerFiredResult> triggersFired(final List<OperableTrigger> triggers) throws JobPersistenceException { return (List<TriggerFiredResult>)executeInNonManagedTXLock( LOCK_TRIGGER_ACCESS, new TransactionCallback() { public Object execute(Connection conn) throws JobPersistenceException { List<TriggerFiredResult> results = new ArrayList<TriggerFiredResult>(); TriggerFiredResult result; for (OperableTrigger trigger : triggers) { try { TriggerFiredBundle bundle = triggerFired(conn, trigger); result = new TriggerFiredResult(bundle); } catch (JobPersistenceException jpe) { result = new TriggerFiredResult(jpe); } catch(RuntimeException re) { result = new TriggerFiredResult(re); } results.add(result); } return results; } }); }JobStoreSupport的executeInNonManagedTXLock方法
protected Object executeInNonManagedTXLock( String lockName, TransactionCallback txCallback) 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(); } Object result = txCallback.execute(conn); commitConnection(conn); 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(conn, lockName, transOwner); } finally { cleanupConnection(conn); } } }JobStoreSupport的triggersFired方法
public List<TriggerFiredResult> triggersFired(final List<OperableTrigger> triggers) throws JobPersistenceException { return (List<TriggerFiredResult>)executeInNonManagedTXLock( LOCK_TRIGGER_ACCESS, new TransactionCallback() { public Object execute(Connection conn) throws JobPersistenceException { List<TriggerFiredResult> results = new ArrayList<TriggerFiredResult>(); TriggerFiredResult result; for (OperableTrigger trigger : triggers) { try { TriggerFiredBundle bundle = triggerFired(conn, trigger); result = new TriggerFiredResult(bundle); } catch (JobPersistenceException jpe) { result = new TriggerFiredResult(jpe); } catch(RuntimeException re) { result = new TriggerFiredResult(re); } results.add(result); } return results; } }); }JobStoreSupport的triggerFired方法
protected TriggerFiredBundle triggerFired(Connection conn, OperableTrigger trigger) throws JobPersistenceException { JobDetail job = null; 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()); }JobStoreSupport的storeTrigger方法
protected void storeTrigger(Connection conn, OperableTrigger newTrigger, JobDetail job, boolean replaceExisting, String state, boolean forceState, boolean recovering) throws ObjectAlreadyExistsException, JobPersistenceException { boolean existingTrigger = triggerExists(conn, newTrigger.getKey()); if ((existingTrigger) && (!replaceExisting)) { throw new ObjectAlreadyExistsException(newTrigger); } try { boolean shouldBepaused = false; if (!forceState) { shouldBepaused = getDelegate().isTriggerGroupPaused( conn, newTrigger.getKey().getGroup()); if(!shouldBepaused) { shouldBepaused = getDelegate().isTriggerGroupPaused(conn, ALL_GROUPS_PAUSED); if (shouldBepaused) { getDelegate().insertPausedTriggerGroup(conn, newTrigger.getKey().getGroup()); } } if (shouldBepaused && (state.equals(STATE_WAITING) || state.equals(STATE_ACQUIRED))) { state = STATE_PAUSED; } } if(job == null) { job = getDelegate().selectJobDetail(conn, newTrigger.getJobKey(), getClassLoadHelper()); } if (job == null) { throw new JobPersistenceException("The job (" + newTrigger.getJobKey() + ") referenced by the trigger does not exist."); } if (job.isConcurrentExectionDisallowed() && !recovering) { state = checkBlockedState(conn, job.getKey(), state); } if (existingTrigger) { getDelegate().updateTrigger(conn, newTrigger, state, job); } else { getDelegate().insertTrigger(conn, newTrigger, state, job); } } catch (Exception e) { throw new JobPersistenceException("Couldn't store trigger '" + newTrigger.getKey() + "' for '" + newTrigger.getJobKey() + "' job:" + e.getMessage(), e); } }StdJDBCDelegate的updateTrigger方法
public int updateTrigger(Connection conn, OperableTrigger trigger, String state, JobDetail jobDetail) throws SQLException, IOException { // save some clock cycles by unnecessarily writing job data blob ... boolean updateJobData = trigger.getJobDataMap().isDirty(); ByteArrayOutputStream baos = null; if(updateJobData && trigger.getJobDataMap().size() > 0) { baos = serializeJobData(trigger.getJobDataMap()); } PreparedStatement ps = null; int insertResult = 0; try { if(updateJobData) { ps = conn.prepareStatement(rtp(UPDATE_TRIGGER)); } else { ps = conn.prepareStatement(rtp(UPDATE_TRIGGER_SKIP_DATA)); } ps.setString(1, trigger.getJobKey().getName()); ps.setString(2, trigger.getJobKey().getGroup()); ps.setString(3, trigger.getDescription()); long nextFireTime = -1; if (trigger.getNextFireTime() != null) { nextFireTime = trigger.getNextFireTime().getTime(); } ps.setBigDecimal(4, new BigDecimal(String.valueOf(nextFireTime))); long prevFireTime = -1; if (trigger.getPreviousFireTime() != null) { prevFireTime = trigger.getPreviousFireTime().getTime(); } ps.setBigDecimal(5, new BigDecimal(String.valueOf(prevFireTime))); ps.setString(6, state); TriggerPersistenceDelegate tDel = findTriggerPersistenceDelegate(trigger); String type = TTYPE_BLOB; if(tDel != null) type = tDel.getHandledTriggerTypeDiscriminator(); ps.setString(7, type); ps.setBigDecimal(8, new BigDecimal(String.valueOf(trigger .getStartTime().getTime()))); long endTime = 0; if (trigger.getEndTime() != null) { endTime = trigger.getEndTime().getTime(); } ps.setBigDecimal(9, new BigDecimal(String.valueOf(endTime))); ps.setString(10, trigger.getCalendarName()); ps.setInt(11, trigger.getMisfireInstruction()); ps.setInt(12, trigger.getPriority()); if(updateJobData) { setBytes(ps, 13, baos); ps.setString(14, trigger.getKey().getName()); ps.setString(15, trigger.getKey().getGroup()); } else { ps.setString(13, trigger.getKey().getName()); ps.setString(14, trigger.getKey().getGroup()); } insertResult = ps.executeUpdate(); if(tDel == null) updateBlobTrigger(conn, trigger); else tDel.updateExtendedTriggerProperties(conn, trigger, state, jobDetail); } finally { closeStatement(ps); } return insertResult; }
4.3锁机制
通过配置文件配置org.quartz.jobStore.lockHandler.class
DBSemaphore的实现由2个
UpdateLockRowSemaphore以update获取锁
public static final String UPDATE_FOR_LOCK = "UPDATE " + TABLE_PREFIX_SUBST + TABLE_LOCKS + " SET " + COL_LOCK_NAME + " = " + COL_LOCK_NAME + " WHERE " + COL_SCHEDULER_NAME + " = " + SCHED_NAME_SUBST + " AND " + COL_LOCK_NAME + " = ? ";
StdRowLockSemaphore以select ``` from update获取锁
public static final String SELECT_FOR_LOCK = "SELECT * FROM " + TABLE_PREFIX_SUBST + TABLE_LOCKS + " WHERE " + COL_SCHEDULER_NAME + " = " + SCHED_NAME_SUBST + " AND " + COL_LOCK_NAME + " = ? FOR UPDATE";
注:本文quartz版本为2.0.2