繼承了 InitializingBean 接口會(huì)在類被注入Spring容器后
執(zhí)行 afterPropertiesSet 方法
@Override
public void afterPropertiesSet() throws Exception {
if (this.dataSource == null && this.nonTransactionalDataSource != null) {
this.dataSource = this.nonTransactionalDataSource;
}
if (this.applicationContext != null && this.resourceLoader == null) {
this.resourceLoader = this.applicationContext;
}
// Create SchedulerFactory instance...
//(1)strat 創(chuàng)建 SchedulerFactory 并且初始化
SchedulerFactory schedulerFactory = BeanUtils.instantiateClass(this.schedulerFactoryClass);
initSchedulerFactory(schedulerFactory);
//(1)end
if (this.resourceLoader != null) {
// Make given ResourceLoader available for SchedulerFactory configuration.
configTimeResourceLoaderHolder.set(this.resourceLoader);
}
if (this.taskExecutor != null) {
// Make given TaskExecutor available for SchedulerFactory configuration.
configTimeTaskExecutorHolder.set(this.taskExecutor);
}
if (this.dataSource != null) {
// Make given DataSource available for SchedulerFactory configuration.
configTimeDataSourceHolder.set(this.dataSource);
}
if (this.nonTransactionalDataSource != null) {
// Make given non-transactional DataSource available for SchedulerFactory configuration.
configTimeNonTransactionalDataSourceHolder.set(this.nonTransactionalDataSource);
}
//(2)strat
// Get Scheduler instance from SchedulerFactory.
try {
this.scheduler = createScheduler(schedulerFactory, this.schedulerName);
populateSchedulerContext();
if (!this.jobFactorySet && !(this.scheduler instanceof RemoteScheduler)) {
// Use AdaptableJobFactory as default for a local Scheduler, unless when
// explicitly given a null value through the "jobFactory" bean property.
this.jobFactory = new AdaptableJobFactory();
}
if (this.jobFactory != null) {
if (this.jobFactory instanceof SchedulerContextAware) {
((SchedulerContextAware) this.jobFactory).setSchedulerContext(this.scheduler.getContext());
}
this.scheduler.setJobFactory(this.jobFactory);
}
}
finally {
if (this.resourceLoader != null) {
configTimeResourceLoaderHolder.remove();
}
if (this.taskExecutor != null) {
configTimeTaskExecutorHolder.remove();
}
if (this.dataSource != null) {
configTimeDataSourceHolder.remove();
}
if (this.nonTransactionalDataSource != null) {
configTimeNonTransactionalDataSourceHolder.remove();
}
}
registerListeners();
registerJobsAndTriggers();
//(2)end
}
(1) 創(chuàng)建 SchedulerFactory 并且初始化
- 屬性schedulerFactoryClass的默認(rèn)值是StdSchedulerFactory.class篓吁，因此這里默認(rèn)會(huì)初始化StdSchedulerFactory
private void initSchedulerFactory(SchedulerFactory schedulerFactory) throws SchedulerException, IOException {
if (!(schedulerFactory instanceof StdSchedulerFactory)) {
if (this.configLocation != null || this.quartzProperties != null ||
this.taskExecutor != null || this.dataSource != null) {
throw new IllegalArgumentException(
"StdSchedulerFactory required for applying Quartz properties: " + schedulerFactory);
}
// Otherwise assume that no initialization is necessary...
return;
}
Properties mergedProps = new Properties();
if (this.resourceLoader != null) {
mergedProps.setProperty(StdSchedulerFactory.PROP_SCHED_CLASS_LOAD_HELPER_CLASS,
ResourceLoaderClassLoadHelper.class.getName());
}
if (this.taskExecutor != null) {
mergedProps.setProperty(StdSchedulerFactory.PROP_THREAD_POOL_CLASS,
LocalTaskExecutorThreadPool.class.getName());
}
else {
// Set necessary default properties here, as Quartz will not apply
// its default configuration when explicitly given properties.
mergedProps.setProperty(StdSchedulerFactory.PROP_THREAD_POOL_CLASS, SimpleThreadPool.class.getName());
mergedProps.setProperty(PROP_THREAD_COUNT, Integer.toString(DEFAULT_THREAD_COUNT));
}
if (this.configLocation != null) {
if (logger.isInfoEnabled()) {
logger.info("Loading Quartz config from [" + this.configLocation + "]");
}
PropertiesLoaderUtils.fillProperties(mergedProps, this.configLocation);
}
CollectionUtils.mergePropertiesIntoMap(this.quartzProperties, mergedProps);
if (this.dataSource != null) {
mergedProps.put(StdSchedulerFactory.PROP_JOB_STORE_CLASS, LocalDataSourceJobStore.class.getName());
}
// Make sure to set the scheduler name as configured in the Spring configuration.
if (this.schedulerName != null) {
mergedProps.put(StdSchedulerFactory.PROP_SCHED_INSTANCE_NAME, this.schedulerName);
}
((StdSchedulerFactory) schedulerFactory).initialize(mergedProps);
}
- 對(duì)于非StdSchedulerFactory的其他SchedulerFactory，需要對(duì)參數(shù)進(jìn)行檢查草添；
- 設(shè)置內(nèi)置的屬性并存入mergedProps這個(gè)字典中。這些屬性包括：
- org.quartz.scheduler.classLoadHelper.class：用于Quartz與Spring集成時(shí)加載Spring資源狞贱；
- org.quartz.threadPool.class：執(zhí)行Quartz中Task的線程池；
- org.quartz.threadPool.threadCount：執(zhí)行Quartz中Task的線程池的線程數(shù)量。
- 加載configLocation屬性指定的屬性文件中的屬性并合并到mergedProps中诵盼，這說明屬性文件中的配置可以覆蓋內(nèi)置的屬性參數(shù)迹卢。向mergedProps中設(shè)置其它屬性：
org.quartz.jobStore.class：作業(yè)持久化存儲(chǔ)的類辽故，值為LocalDataSourceJobStore；
org.quartz.scheduler.instanceName：值為Spring配置文件中設(shè)置的值腐碱；
(2) 初始化后的動(dòng)作
- 使用ThreadLocal技術(shù)持有resourceLoader誊垢、taskExecutor、dataSource症见、nonTransactionalDataSource喂走；
- 調(diào)用createScheduler方法創(chuàng)建調(diào)度器
- 調(diào)用populateSchedulerContext，指定調(diào)度上下文（SchedulerContext）的屬性和它有的Spring的ApplicationContext谋作；
- 給調(diào)度器設(shè)置作業(yè)工廠類JobFactory芋肠；
- 調(diào)用registerListeners方法注冊(cè)有關(guān)調(diào)度、作業(yè)瓷们、觸發(fā)器等內(nèi)容的監(jiān)聽器
- 調(diào)用registerJobsAndTriggers方法注冊(cè)作業(yè)和觸發(fā)器
2.1 createScheduler
/**
* Create the Scheduler instance for the given factory and scheduler name.
* Called by {@link #afterPropertiesSet}.
* <p>The default implementation invokes SchedulerFactory's {@code getScheduler}
* method. Can be overridden for custom Scheduler creation.
* @param schedulerFactory the factory to create the Scheduler with
* @param schedulerName the name of the scheduler to create
* @return the Scheduler instance
* @throws SchedulerException if thrown by Quartz methods
* @see #afterPropertiesSet
* @see org.quartz.SchedulerFactory#getScheduler
*/
protected Scheduler createScheduler(SchedulerFactory schedulerFactory, String schedulerName)
throws SchedulerException {
// Override thread context ClassLoader to work around naive Quartz ClassLoadHelper loading.
//(1)start
Thread currentThread = Thread.currentThread();
ClassLoader threadContextClassLoader = currentThread.getContextClassLoader();
boolean overrideClassLoader = (this.resourceLoader != null &&
!this.resourceLoader.getClassLoader().equals(threadContextClassLoader));
if (overrideClassLoader) {
currentThread.setContextClassLoader(this.resourceLoader.getClassLoader());
}
//(1)end
try {
SchedulerRepository repository = SchedulerRepository.getInstance();
synchronized (repository) {
Scheduler existingScheduler = (schedulerName != null ? repository.lookup(schedulerName) : null);
Scheduler newScheduler = schedulerFactory.getScheduler();
if (newScheduler == existingScheduler) {
throw new IllegalStateException("Active Scheduler of name '" + schedulerName + "' already registered " +
"in Quartz SchedulerRepository. Cannot create a new Spring-managed Scheduler of the same name!");
}
if (!this.exposeSchedulerInRepository) {
// Need to remove it in this case, since Quartz shares the Scheduler instance by default!
SchedulerRepository.getInstance().remove(newScheduler.getSchedulerName());
}
return newScheduler;
}
}
finally {
if (overrideClassLoader) {
// Reset original thread context ClassLoader.
currentThread.setContextClassLoader(threadContextClassLoader);
}
}
}
- (1)設(shè)置線程上下文的類加載器业栅；
- (2) 獲取 SchedulerRepository 單例
public static synchronized SchedulerRepository getInstance() {
if (inst == null) {
inst = new SchedulerRepository();
}
return inst;
}
- (3)從調(diào)度倉庫實(shí)例SchedulerRepository中查找已經(jīng)存在的調(diào)度器
獲取調(diào)取器（見代碼清單3），其實(shí)際上首先從調(diào)度器緩存中查找調(diào)度器谬晕，否則調(diào)用instantiate方法創(chuàng)建調(diào)度器碘裕；
public Scheduler getScheduler() throws SchedulerException {
if (cfg == null) {
initialize();
}
SchedulerRepository schedRep = SchedulerRepository.getInstance();
Scheduler sched = schedRep.lookup(getSchedulerName());
if (sched != null) {
if (sched.isShutdown()) {
schedRep.remove(getSchedulerName());
} else {
return sched;
}
}
//重點(diǎn)
sched = instantiate();
return sched;
}
instantiate()
private Scheduler instantiate() throws SchedulerException {
//instantiate方法中包含了很多從PropertiesParser
//讀取屬性 代碼略
//如果當(dāng)前調(diào)度器實(shí)際是代理遠(yuǎn)程RMI調(diào)度器，
//那么創(chuàng)建RemoteScheduler攒钳，并將當(dāng)前調(diào)取器與RemoteScheduler進(jìn)行綁定帮孔，
//最后以此RemoteScheduler作為調(diào)度器
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
如果當(dāng)前調(diào)度器實(shí)際是代理遠(yuǎn)程JMX調(diào)度器，那么創(chuàng)建RemoteMBeanScheduler不撑，并將當(dāng)前調(diào)度器與RemoteMBeanScheduler進(jìn)行綁定文兢，最后以此RemoteMBeanScheduler作為調(diào)度器，
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;
}
實(shí)例化作業(yè)工廠
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;
}
}
實(shí)例化實(shí)例ID生成器
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;
}
}
實(shí)例化線程池
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;
}
實(shí)例化JobStore的具體實(shí)例
jobStore顧名思義焕檬，就是作業(yè)的存儲(chǔ)姆坚，以LocalDataSourceJobStore為例，將通過它對(duì)觸發(fā)器实愚、作業(yè)等內(nèi)容進(jìn)行增刪改查兼呵。
// 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;
}
}
}
獲取數(shù)據(jù)庫管理器并設(shè)置數(shù)據(jù)庫連接池(略)
設(shè)置調(diào)度器插件（略）
設(shè)置作業(yè)監(jiān)聽器
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 = null;
try {
nameSetter = listener.getClass().getMethod("setName", strArg);
}
catch(NoSuchMethodException ignore) {
/* do nothing */
}
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;
}
設(shè)置觸發(fā)器監(jiān)聽器
獲取線程執(zhí)行器(此線程執(zhí)行器用于執(zhí)行定時(shí)調(diào)度線程QuartzSchedulerThrea)
....
//創(chuàng)建 Scheduler
qs = new QuartzScheduler(rsrcs, idleWaitTime, dbFailureRetry);
qsInited = true;
// Create Scheduler ref...
Scheduler scheduler = instantiate(rsrcs, qs);
public QuartzScheduler(QuartzSchedulerResources resources, long idleWaitTime, @Deprecated long dbRetryInterval)
throws SchedulerException {
this.resources = resources;
if (resources.getJobStore() instanceof JobListener) {
addInternalJobListener((JobListener)resources.getJobStore());
}
//創(chuàng)建運(yùn)行定時(shí)任務(wù)核心類
this.schedThread = new QuartzSchedulerThread(this, resources);
ThreadExecutor schedThreadExecutor = resources.getThreadExecutor();
//提交進(jìn)入線程池
schedThreadExecutor.execute(this.schedThread);
if (idleWaitTime > 0) {
this.schedThread.setIdleWaitTime(idleWaitTime);
}
jobMgr = new ExecutingJobsManager();
addInternalJobListener(jobMgr);
errLogger = new ErrorLogger();
addInternalSchedulerListener(errLogger);
signaler = new SchedulerSignalerImpl(this, this.schedThread);
getLog().info("Quartz Scheduler v." + getVersion() + " created.");
}
QuartzSchedulerThread(QuartzScheduler qs, QuartzSchedulerResources qsRsrcs) {
this(qs, qsRsrcs, qsRsrcs.getMakeSchedulerThreadDaemon(), Thread.NORM_PRIORITY);
}
QuartzSchedulerThread(QuartzScheduler qs, QuartzSchedulerResources qsRsrcs, boolean setDaemon, int threadPrio) {
super(qs.getSchedulerThreadGroup(), qsRsrcs.getThreadName());
this.qs = qs;
this.qsRsrcs = qsRsrcs;
this.setDaemon(setDaemon);
if(qsRsrcs.isThreadsInheritInitializersClassLoadContext()) {
log.info("QuartzSchedulerThread Inheriting ContextClassLoader of thread: " + Thread.currentThread().getName());
this.setContextClassLoader(Thread.currentThread().getContextClassLoader());
}
this.setPriority(threadPrio);
// start the underlying thread, but put this object into the 'paused'
// state
// so processing doesn't start yet...
paused = true;
halted = new AtomicBoolean(false); //啟動(dòng)設(shè)置不暫停
}
由于new的時(shí)候交給了線程池
等于 run方法執(zhí)行
@Override
public void run() {
int acquiresFailed = 0;
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) {
}
// reset failure counter when paused, so that we don't
// wait again after unpausing
acquiresFailed = 0;
}
if (halted.get()) {
break;
}
}
每次pause 相當(dāng)于 停1s
在 afterPropertiesSet 的最后設(shè)置了 Spring的上下文
/**
* Expose the specified context attributes and/or the current
* ApplicationContext in the Quartz SchedulerContext.
*/
private void populateSchedulerContext() throws SchedulerException {
// Put specified objects into Scheduler context.
if (this.schedulerContextMap != null) {
this.scheduler.getContext().putAll(this.schedulerContextMap);
}
// Register ApplicationContext in Scheduler context.
if (this.applicationContextSchedulerContextKey != null) {
if (this.applicationContext == null) {
throw new IllegalStateException(
"SchedulerFactoryBean needs to be set up in an ApplicationContext " +
"to be able to handle an 'applicationContextSchedulerContextKey'");
}
this.scheduler.getContext().put(this.applicationContextSchedulerContextKey, this.applicationContext);
}
}
