quartz的schedulerFactory实现解析

由于需要实现一个scheduler，用于添加job，更改job等功能，但是没有调度功能。没错主要是用于监控和管理集群用的。所以需要仔细看scheduler的接口规范，于是又了这篇文章。

SchedulerFactory 是一个接口，用于Scheduler的创建和管理。接口很简单，这里主要是解析StdSchedulerFactory 的实现方式。

1、构造函数：

 public StdSchedulerFactory(Properties props) throws SchedulerException {

        //把初始化参数代理给<span style="font-family: Arial, Helvetica, sans-serif;">initialize，通过Properties来做初始化</span>initialize(props);}public StdSchedulerFactory(String fileName) throws SchedulerException {

        //通过配置文件初始化，这也会把文件解析为Propertiesinitialize(fileName);}

2、初始化过程

初始化过程，首先会看是否传入了配置文件或Properties属性，没有的话，会根据系统变量--》工程classpath--》quartz的包的顺序查找quartz.properties文件

查找配置文件过程如下：

public void initialize() throws SchedulerException {// short-circuit if already initializedif (cfg != null) {//找到文件后会构建一个cfg属性，所以如果cfg！=null说明已经初始化过了return;}if (initException != null) {throw initException;}//通过查找系统变量查找配置文件String requestedFile = System.getProperty(PROPERTIES_FILE);

       //如果没有找到就用默认的文件名String propFileName = requestedFile != null ? requestedFile: "quartz.properties";

      File propFile = new File(propFileName);Properties props = new Properties();InputStream in = null;try {

           //先查看指定的配置文件是否存在，存在就读取，注意这里的读取方式用文件系统的路径来读取的if (propFile.exists()) {try {if (requestedFile != null) {propSrc = "specified file: '" + requestedFile + "'";} else {propSrc = "default file in current working dir: 'quartz.properties'";}in = new BufferedInputStream(new FileInputStream(propFileName));props.load(in);} catch (IOException ioe) {initException = new SchedulerException("Properties file: '"+ propFileName + "' could not be read.", ioe);throw initException;}

           //如果通过文件系统的路径读不到，那么就通过<span style="font-family: Arial, Helvetica, sans-serif;">ContextClassLoader的</span>路径来读取} else if (requestedFile != null) {in =Thread.currentThread().getContextClassLoader().getResourceAsStream(requestedFile);if(in == null) {initException = new SchedulerException("Properties file: '"+ requestedFile + "' could not be found.");throw initException;}propSrc = "specified file: '" + requestedFile + "' in the class resource path.";in = new BufferedInputStream(in);try {props.load(in);} catch (IOException ioe) {initException = new SchedulerException("Properties file: '"+ requestedFile + "' could not be read.", ioe);throw initException;}} else {propSrc = "default resource file in Quartz package: 'quartz.properties'";//通过当前的classLoader来读取，但是一般情况下这个和当前线程的classLoader基本是一个ClassLoader cl = getClass().getClassLoader();if(cl == null)cl = findClassloader();if(cl == null)throw new SchedulerConfigException("Unable to find a class loader on the current thread or class.");in = cl.getResourceAsStream("quartz.properties");if (in == null) {in = cl.getResourceAsStream("/quartz.properties");}

<span style="font-family: Arial, Helvetica, sans-serif;">                //是在找不到，那么就从quartz的包找了</span>

                if (in == null) {in = cl.getResourceAsStream("org/quartz/quartz.properties");}if (in == null) {initException = new SchedulerException("Default quartz.properties not found in class path");throw initException;}try {props.load(in);} catch (IOException ioe) {initException = new SchedulerException("Resource properties file: 'org/quartz/quartz.properties' "+ "could not be read from the classpath.", ioe);throw initException;}}} finally {if(in != null) {try { in.close(); } catch(IOException ignore) { /* ignore */ }}}//最后解析为一个Properties去配置schudulerinitialize(overrideWithSysProps(props));

上面都在准备配置信息，这些最后会生成一个Cfg对象，用它去获取属性，配置Scheduler，在调用

Scheduler getScheduler()

的时候根据会首先检查是否已经创建了，没有创建会根据这些属性创建新的Scheduler：

instantiate()

instantiate的关键代码：
关键属性：

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 = 15000L; // 15 secsString classLoadHelperClass;String jobFactoryClass;ThreadExecutor threadExecutor;

schedulerId的生成：

 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");}

        //如果配置成系统属性Id，那么就用相应的class生成else if (schedInstId.equals(SYSTEM_PROPERTY_AS_INSTANCE_ID)) {autoId = true;instanceIdGeneratorClass = "org.quartz.simpl.SystemPropertyInstanceIdGenerator";}

获取UserTansaction的Jndi url，在分布式事务中使用

 userTXLocation = cfg.getStringProperty(PROP_SCHED_USER_TX_URL,userTXLocation);

是否用UserTansaction来执行Job：

wrapJobInTx = cfg.getBooleanProperty(PROP_SCHED_WRAP_JOB_IN_USER_TX,wrapJobInTx);

job实例化factory:

 jobFactoryClass = cfg.getStringProperty(PROP_SCHED_JOB_FACTORY_CLASS, null);

jmx设置相关：

 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);

rmi相关：

        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);

不过Jmx和rmi是不能同时开启的：

  if (jmxProxy && rmiProxy) {throw new SchedulerConfigException("Cannot proxy both RMI and JMX.");}

如果是rmi代理的就创建remoteScheduler：（不会有集群效果，毕竟所有操作都是远程那边在做）

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;}

如果是jmx远程代理就用jmxscheduler：

  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 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);}//获取jobFactory需要的属性，这里的方法就是定义自定义属性名（固定前缀）的实现方式，比如说常用的datasourcetProps = 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;}}

配置ThreadPool，属性值的获取方式和JobFactory一直：

  String tpClass = cfg.getStringProperty(PROP_THREAD_POOL_CLASS, SimpleThreadPool.class.getName());if (tpClass == null) {initException = new SchedulerException("ThreadPool class not specified. ");throw initException;}try {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;}

jobStore的配置：

  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;}//设置JobStore的scheduleName和scheduleId两个值，就是简单的反射设置而已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;}//如果是JobStoreSupport的话，那么就需要设置锁控制器，用于集群的调度同步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;}}}

获取datasource，可能多个：

String[] dsNames = cfg.getPropertyGroups(PROP_DATASOURCE_PREFIX);//获取多个dataSource，所以需要JTX分布式事务的支持，因为有多数据源的job存在的for (int i = 0; i < dsNames.length; i++) {PropertiesParser pp = new PropertiesParser(cfg.getPropertyGroup(PROP_DATASOURCE_PREFIX + "." + dsNames[i], true));//先检测connectionProvider这种配置方式，本质上所有的datasource都会被封装为某种Provider，Provider是quartz对datasource的一中封装而已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 {//移除Provider这个配置后，把其他配置关联到provider// remove the class name, so it isn't attempted to be setpp.getUnderlyingProperties().remove(PROP_CONNECTION_PROVIDER_CLASS);setBeanProps(cp, pp.getUnderlyingProperties());cp.initialize();} catch (Exception e) {initException = new SchedulerException("ConnectionProvider class '" + cpClass+ "' props could not be configured.", e);throw initException;}//把所有的Datasource加到managerdbMgr = DBConnectionManager.getInstance();dbMgr.addConnectionProvider(dsNames[i], cp);} else {

                //查找Jndi这种配置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 {

                    //检查local driver这种配置String dsDriver = pp.getStringProperty(PoolingConnectionProvider.DB_DRIVER);String dsURL = pp.getStringProperty(PoolingConnectionProvider.DB_URL);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(pp.getUnderlyingProperties());dbMgr = DBConnectionManager.getInstance();dbMgr.addConnectionProvider(dsNames[i], cp);} catch (SQLException sqle) {initException = new SchedulerException("Could not initialize DataSource: " + dsNames[i],sqle);throw initException;}}}}

接下来还有SchedulerPlugins、JobListeners、TriggerListeners、ThreadExecutor等的配置。基本都大同小异，毕竟都是配置文件解析而已。

其中ThreadExecutor是执行池，主要是执行schedule 和refire任务。

然后初始化JobRunShellFactory，这个是一个封装执行Job的factory，主要是为了一些listener的执行和异常处理：

 JobRunShellFactory jrsf = null; // Create correct run-shell factory...if (userTXLocation != null) {UserTransactionHelper.setUserTxLocation(userTXLocation);}//更具JTA的配置情况初始化特定的RunShellFactoryif (wrapJobInTx) {jrsf = new JTAJobRunShellFactory();} else {jrsf = new JTAAnnotationAwareJobRunShellFactory();}

对于db JobStore的话是有一些处理的：

 if (js instanceof JobStoreSupport) {JobStoreSupport jjs = (JobStoreSupport)js;jjs.setDbRetryInterval(dbFailureRetry);if(threadsInheritInitalizersClassLoader)jjs.setThreadsInheritInitializersClassLoadContext(threadsInheritInitalizersClassLoader);jjs.setThreadExecutor(threadExecutor);//所以 ThreadExecutor也用来处理数据库的链接的（比如说定时重试）}

初始化以上对象和资源后，quartz会把他保存到QuartzSchedulerResources里面，所以你会看到代码的很多地方都有用到这个对象：

 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);

Sheduler的初始化和设置在下面：

qs = new QuartzScheduler(rsrcs, idleWaitTime, dbFailureRetry);//实例化SchedulerqsInited = true;// Create Scheduler ref...Scheduler scheduler = instantiate(rsrcs, qs);//封装QuartzScheduler为StdQuartzScheduler，其实<span style="font-family: Arial, Helvetica, sans-serif;">StdQuartzScheduler就是一个装饰模式而已，QuartzSch//eduler是没有实现Sheduler接口的</span>
// set job factory if specifiedif(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, loadHelper);}// add listenersfor (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((String)key, val);}// fire up job store, and runshell factory//启动jobjs.setInstanceId(schedInstId);js.setInstanceName(schedName);js.setThreadPoolSize(tp.getPoolSize());js.initialize(loadHelper, qs.getSchedulerSignaler());//是一个回掉Job Scheduler的接口<span style="font-family: Arial, Helvetica, sans-serif;">SchedulerSignaler</span>
//启动jobRunShellFactory,<span style="font-family: Arial, Helvetica, sans-serif;">initialize方法就是标准的初始化方法，是quartz的代码习惯</span>
jrsf.initialize(scheduler);//启动shedulerqs.initialize();getLog().info("Quartz scheduler '" + scheduler.getSchedulerName()+ "' initialized from " + propSrc);getLog().info("Quartz scheduler version: " + qs.getVersion());//添加资源引用防止被垃圾回收// prevents the repository from being garbage collectedqs.addNoGCObject(schedRep);// prevents the db manager from being garbage collectedif (dbMgr != null) {qs.addNoGCObject(dbMgr);}

总结下来，其实挺简单的：读取配置文件，安装前缀的风格初始化各种类和属性，然后包装到 QuartzSchedulerResources 给Sheduler用，然后初始化：

jobStore、ThreadPool、JobRunShellFactory、ThreadExecutor、jobFactory、开启各种mananger、初始化Scheduler并为其添加pluger和listener，

quartz的schedulerFactory实现解析相关推荐

【定时任务】quartz表达式Cron Expression解析
每个月的第5天和第20天的上午8点到10点执行,每隔半小时执行一次,上午10:00不会执行 cron 表达式的格式 Quartz cron 表达式的格式十分类似于 UNIX cron 格式,但还是有少 ...
Quartz定时任务调度机制解析(CronTirgger、SimpleTrigger )
一.Quartz的介绍 Quartz 是 OpenSymphony 开源组织在任务调度领域的一个开源项目,完全基于 Java 实现.该项目于 2009 年被 Terracotta 收购,目前是 Ter ...
quartz源码解析--转
quartz源码解析(一) . http://ssuupv.blog.163.com/blog//146156722013829111028966/ 任何个人.任何企业.任何行业都会有作业调度的需求 ...
一文揭秘定时任务调度框架quartz
之前写过quartz或者引用过quartz的一些文章,有很多人给我发消息问quartz的相关问题, quartz 报错:java.lang.classNotFoundException quartz源 ...
Quartz框架多个trigger任务执行出现漏执行的问题分析--转
原文地址:http://blog.csdn.net/dailywater/article/details/51470779 一.问题描述使用Quartz配置定时任务,配置了超过10个定时任务,这些 ...
Quartz-Spring集成Quartz通过XML配置的方式
文章目录概述 Maven依赖步骤 1 创建JobDteail 2 创建Trigger 3 创建Scheduler 示例-MethodInvokingJobDetailFactoryBean 示例- ...
java 动态添加定时器_Spring整合Quartz实现动态定时器的示例代码
一.版本说明 spring3.1以下的版本必须使用quartz1.x系列,3.1以上的版本才支持quartz 2.x,不然会出错. 原因:spring对于quartz的支持实现,org.springf ...
001-spring结合quartz使用
一.添加pom 二.定义业务类 public class TestJobTask{ /** *业务逻辑处理 */ public void service(){ /**业务逻辑*/ .. } } 二.配 ...
SpringBoot结合Quartz实现定时任务
<从零打造项目>系列文章工具比MyBatis Generator更强大的代码生成器 ORM框架选型 SpringBoot项目基础设施搭建 SpringBoot集成Mybatis项目实操 ...

quartz的schedulerFactory实现解析

quartz的schedulerFactory实现解析相关推荐

最新文章

热门文章