Quartz任務調度分析：http://donald-draper.iteye.com/blog/2323118
Spring與Quartz集成詳解：http://donald-draper.iteye.com/blog/2323591
ThreadLocal徹底理解：http://blog.csdn.net/lufeng20/article/details/24314381/
public class SchedulerFactoryBean extends SchedulerAccessor
implements FactoryBean, BeanNameAware, ApplicationContextAware, InitializingBean, DisposableBean, SmartLifecycle
{
public SchedulerFactoryBean()
{
//Quartz的標準調度工場類
schedulerFactoryClass = org/quartz/impl/StdSchedulerFactory;
jobFactorySet = false;
autoStartup = true;
startupDelay = 0;
phase = 2147483647;
exposeSchedulerInRepository = false;
waitForJobsToCompleteOnShutdown = false;
}
//設置調度器工廠類
public void setSchedulerFactoryClass(Class schedulerFactoryClass)
{
Assert.isAssignable(org/quartz/SchedulerFactory, schedulerFactoryClass);
this.schedulerFactoryClass = schedulerFactoryClass;
}
//設置調度器Name
public void setSchedulerName(String schedulerName)
{
this.schedulerName = schedulerName;
}
//Quartz屬性
public void setQuartzProperties(Properties quartzProperties)
{
this.quartzProperties = quartzProperties;
}
//設置任務執行器
public void setTaskExecutor(Executor taskExecutor)
{
this.taskExecutor = taskExecutor;
}
//設置beanName
public void setBeanName(String name)
{
if(schedulerName == null)
schedulerName = name;
}
//設置應用上下文
public void setApplicationContext(ApplicationContext applicationContext)
{
this.applicationContext = applicationContext;
}
//設置job工廠
public void setJobFactory(JobFactory jobFactory)
{
this.jobFactory = jobFactory;
jobFactorySet = true;
}
public void setAutoStartup(boolean autoStartup)
{
this.autoStartup = autoStartup;
}
public void afterPropertiesSet()
throws Exception
{
SchedulerFactory schedulerFactory;
if(dataSource == null && nonTransactionalDataSource != null)
dataSource = nonTransactionalDataSource;
if(applicationContext != null && resourceLoader == null)
resourceLoader = applicationContext;
//實例化調度器工廠
schedulerFactory = (SchedulerFactory)BeanUtils.instantiateClass(schedulerFactoryClass);
//初始化調度器工廠
initSchedulerFactory(schedulerFactory);
if(resourceLoader != null)
configTimeResourceLoaderHolder.set(resourceLoader);
if(taskExecutor != null)
configTimeTaskExecutorHolder.set(taskExecutor);
if(dataSource != null)
configTimeDataSourceHolder.set(dataSource);
if(nonTransactionalDataSource != null)
configTimeNonTransactionalDataSourceHolder.set(nonTransactionalDataSource);
//創建調度器
scheduler = createScheduler(schedulerFactory, schedulerName);
//初始化調度器上下文
populateSchedulerContext();
if(!jobFactorySet && !(scheduler instanceof RemoteScheduler))
jobFactory = new AdaptableJobFactory();
if(jobFactory != null)
{
if(jobFactory instanceof SchedulerContextAware)
((SchedulerContextAware)jobFactory).setSchedulerContext(scheduler.getContext());
//設置調度器job工廠類
scheduler.setJobFactory(jobFactory);
}
if(resourceLoader != null)
configTimeResourceLoaderHolder.remove();
if(taskExecutor != null)
configTimeTaskExecutorHolder.remove();
if(dataSource != null)
configTimeDataSourceHolder.remove();
if(nonTransactionalDataSource != null)
configTimeNonTransactionalDataSourceHolder.remove();
break MISSING_BLOCK_LABEL_334;
if(resourceLoader != null)
configTimeResourceLoaderHolder.remove();
if(taskExecutor != null)
configTimeTaskExecutorHolder.remove();
if(dataSource != null)
configTimeDataSourceHolder.remove();
if(nonTransactionalDataSource != null)
configTimeNonTransactionalDataSourceHolder.remove();
//註冊監聽器
registerListeners();
//註冊job&Trriger
registerJobsAndTriggers();
return;
}
//初始化調度工廠類
private void initSchedulerFactory(SchedulerFactory schedulerFactory)
throws SchedulerException, IOException
{
//判斷標準調度工廠的屬性是否完整
if(!(schedulerFactory instanceof StdSchedulerFactory))
if(configLocation != null || quartzProperties != null || taskExecutor != null || dataSource != null)
throw new IllegalArgumentException((new StringBuilder()).append("StdSchedulerFactory required for applying Quartz properties: ").append(schedulerFactory).toString());
Properties mergedProps = new Properties();
if(resourceLoader != null)
//設置classLoadHelper類
mergedProps.setProperty("org.quartz.scheduler.classLoadHelper.class", org/springframework/scheduling/quartz/ResourceLoaderClassLoadHelper.getName());
if(taskExecutor != null)
{ //設置線程池類
mergedProps.setProperty("org.quartz.threadPool.class", org/springframework/scheduling/quartz/LocalTaskExecutorThreadPool.getName());
} else
{
mergedProps.setProperty("org.quartz.threadPool.class", org/quartz/simpl/SimpleThreadPool.getName());
mergedProps.setProperty("org.quartz.threadPool.threadCount", Integer.toString(10));
}
if(configLocation != null)
{
PropertiesLoaderUtils.fillProperties(mergedProps, configLocation);
}
//調度器工廠屬性合並重組
CollectionUtils.mergePropertiesIntoMap(quartzProperties, mergedProps);
if(dataSource != null)
mergedProps.put("org.quartz.jobStore.class", org/springframework/scheduling/quartz/LocalDataSourceJobStore.getName());
if(schedulerName != null)
mergedProps.put("org.quartz.scheduler.instanceName", schedulerName);
//標準調度工廠初始化
((StdSchedulerFactory)schedulerFactory).initialize(mergedProps);
}
//創建調度器
protected Scheduler createScheduler(SchedulerFactory schedulerFactory, String schedulerName)
throws SchedulerException
{
Thread currentThread;
ClassLoader threadContextClassLoader;
boolean overrideClassLoader;
currentThread = Thread.currentThread();
threadContextClassLoader = currentThread.getContextClassLoader();
overrideClassLoader = resourceLoader != null && !resourceLoader.getClassLoader().equals(threadContextClassLoader);
if(overrideClassLoader)
//如果資源加載器不為空，則設置當前線程上下文加載器
currentThread.setContextClassLoader(resourceLoader.getClassLoader());
//獲取調度器倉庫實例
SchedulerRepository repository = SchedulerRepository.getInstance();
Scheduler scheduler1;
synchronized(repository)
{
//從調度器倉庫尋找schedulerName的調度器
Scheduler existingScheduler = schedulerName == null ? null : repository.lookup(schedulerName);
//從調度器工廠獲取調度器
Scheduler newScheduler = schedulerFactory.getScheduler();
if(newScheduler == existingScheduler)
throw new IllegalStateException((new StringBuilder()).append("Active Scheduler of name '").append(schedulerName).append("' already registered ").append("in Quartz SchedulerRepository. Cannot create a new Spring-managed Scheduler of the same name!").toString());
if(!exposeSchedulerInRepository)
//從倉庫中移除newScheduler
SchedulerRepository.getInstance().remove(newScheduler.getSchedulerName());
scheduler1 = newScheduler;
}
if(overrideClassLoader)
currentThread.setContextClassLoader(threadContextClassLoader);
return scheduler1;
}
//初始化調度器上下文，SchedulerContext,實際上是Map
private void populateSchedulerContext()
throws SchedulerException
{
//如果schedulerContextMap不為空，則將schedulerContextMap放入調度器上下文中
if(schedulerContextMap != null)
scheduler.getContext().putAll(schedulerContextMap);
//如果applicationContextSchedulerContextKey不為空，
//則將applicationContext放入到調度器上下文中
if(applicationContextSchedulerContextKey != null)
{
if(applicationContext == null)
throw new IllegalStateException("SchedulerFactoryBean needs to be set up in an ApplicationContext to be able to handle an 'applicationContextSchedulerContextKey'");
scheduler.getContext().put(applicationContextSchedulerContextKey, applicationContext);
}
}
//啟動調度器
protected void startScheduler(final Scheduler scheduler, final int startupDelay)
throws SchedulerException
{
if(startupDelay <= 0)
{
logger.info("Starting Quartz Scheduler now");
scheduler.start();
} else
{
if(logger.isInfoEnabled())
logger.info((new StringBuilder()).append("Will start Quartz Scheduler [").append(scheduler.getSchedulerName()).append("] in ").append(startupDelay).append(" seconds").toString());
Thread schedulerThread = new Thread() {
public void run()
{
try
{
//如果延遲加載，則休眠startupDelay秒鐘
Thread.sleep(startupDelay * 1000);
}
if(logger.isInfoEnabled())
logger.info((new StringBuilder()).append("Starting Quartz Scheduler now, after delay of ").append(startupDelay).append(" seconds").toString());
try
{
scheduler.start();
}
}
final int val$startupDelay;
final Scheduler val$scheduler;
final SchedulerFactoryBean this$0;
{
this.this$0 = SchedulerFactoryBean.this;
startupDelay = i;
scheduler = scheduler1;
super();
}
};
schedulerThread.setName((new StringBuilder()).append("Quartz Scheduler [").append(scheduler.getSchedulerName()).append("]").toString());
schedulerThread.setDaemon(true);
schedulerThread.start();
}
}
public Scheduler getScheduler()
{
return scheduler;
}
//返回bean對象為Scheduler
public Scheduler getObject()
{
return scheduler;
}
public Class getObjectType()
{
return scheduler == null ? org/quartz/Scheduler : scheduler.getClass();
}
//是否是單例模式，是
public boolean isSingleton()
{
return true;
}
//啟動
public void start()
throws SchedulingException
{
if(scheduler != null)
try
{
startScheduler(scheduler, startupDelay);
}
catch(SchedulerException ex)
{
throw new SchedulingException("Could not start Quartz Scheduler", ex);
}
}
//關閉操作
public void stop()
throws SchedulingException
{
if(scheduler != null)
try
{
scheduler.standby();
}
}
//關閉操作
public void stop(Runnable callback)
throws SchedulingException
{
stop();
callback.run();
}
//獲取調度器是否運行信息
public boolean isRunning()
throws SchedulingException
{
if(scheduler == null)
break MISSING_BLOCK_LABEL_28;
return !scheduler.isInStandbyMode();
return false;
}
//bean銷毀，關閉調度器
public void destroy()
throws SchedulerException
{
logger.info("Shutting down Quartz Scheduler");
scheduler.shutdown(waitForJobsToCompleteOnShutdown);
}
public volatile Object getObject()
throws Exception
{
return getObject();
}
public static final String PROP_THREAD_COUNT = "org.quartz.threadPool.threadCount";
public static final int DEFAULT_THREAD_COUNT = 10;
private static final ThreadLocal configTimeResourceLoaderHolder = new ThreadLocal();
private static final ThreadLocal configTimeTaskExecutorHolder = new ThreadLocal();
private static final ThreadLocal configTimeDataSourceHolder = new ThreadLocal();
private static final ThreadLocal configTimeNonTransactionalDataSourceHolder = new ThreadLocal();
private Class schedulerFactoryClass;//調度工廠類
private String schedulerName;//調度器名
private Resource configLocation;
private Properties quartzProperties;//quartz屬性
private Executor taskExecutor;//任務執行器
private DataSource dataSource;//數據源
private DataSource nonTransactionalDataSource;//無事務數據源
private Map schedulerContextMap;//調度器上下文
private ApplicationContext applicationContext;//應用上下文
private String applicationContextSchedulerContextKey;
private JobFactory jobFactory;//job工廠
private boolean jobFactorySet;
private boolean autoStartup;
private int startupDelay;//延時加載
private int phase;
private boolean exposeSchedulerInRepository;//是否將調度器存儲在調度器倉庫
private boolean waitForJobsToCompleteOnShutdown;//是否等job完成，才關閉調度器
private Scheduler scheduler;//調度器
}
//SchedulerAccessor
public abstract class SchedulerAccessor
implements ResourceLoaderAware
{
public SchedulerAccessor()
{
overwriteExistingJobs = false;
if(jobKeyClass == null && logger.isInfoEnabled())
logger.info("Spring's Quartz 1.x support is deprecated - please upgrade to Quartz 2.0+");
}
protected void registerListeners()
throws SchedulerException
{
Object target;
boolean quartz2;
try
{
//調度器獲取監聽管理器方法
Method getListenerManager = org/quartz/Scheduler.getMethod("getListenerManager", new Class[0]);
//獲取調度器監聽管理器
target = ReflectionUtils.invokeMethod(getListenerManager, getScheduler());
quartz2 = true;
}
catch(NoSuchMethodException ex)
{
target = getScheduler();
quartz2 = false;
}
Class targetClass = target.getClass();
try
{
if(schedulerListeners != null)
{
Method addSchedulerListener = targetClass.getMethod("addSchedulerListener", new Class[] {
org/quartz/SchedulerListener
});
SchedulerListener aschedulerlistener[] = schedulerListeners;
int k = aschedulerlistener.length;
for(int l1 = 0; l1 < k; l1++)
{
SchedulerListener listener = aschedulerlistener[l1];
//添加調度器監聽器
ReflectionUtils.invokeMethod(addSchedulerListener, target, new Object[] {
listener
});
}
}
//添加全局job監聽器
if(globalJobListeners != null)
{
Method addJobListener;
if(quartz2)
//添加job監聽器方法
addJobListener = targetClass.getMethod("addJobListener", new Class[] {
org/quartz/JobListener, Java/util/List
});
else
//添加全局job監聽器方法
addJobListener = targetClass.getMethod("addGlobalJobListener", new Class[] {
org/quartz/JobListener
});
JobListener ajoblistener1[] = globalJobListeners;
int l = ajoblistener1.length;
for(int i2 = 0; i2 < l; i2++)
{
JobListener listener = ajoblistener1[i2];
if(quartz2)
{
List emptyMatchers = new LinkedList();
//添加job監聽器
ReflectionUtils.invokeMethod(addJobListener, target, new Object[] {
listener, emptyMatchers
});
} else
{
//添加全局job監聽器
ReflectionUtils.invokeMethod(addJobListener, target, new Object[] {
listener
});
}
}
}
if(jobListeners != null)
{
JobListener ajoblistener[] = jobListeners;
int i = ajoblistener.length;
for(int i1 = 0; i1 < i; i1++)
{
JobListener listener = ajoblistener[i1];
if(quartz2)
throw new IllegalStateException("Non-global JobListeners not supported on Quartz 2 - manually register a Matcher against the Quartz ListenerManager instead");
//添加job監聽器
getScheduler().addJobListener(listener);
}
}
//添加全局觸發器監聽器
if(globalTriggerListeners != null)
{
Method addTriggerListener;
if(quartz2)
//添加觸發器監聽器方法
addTriggerListener = targetClass.getMethod("addTriggerListener", new Class[] {
org/quartz/TriggerListener, java/util/List
});
else
//添加全局觸發器監聽器方法
addTriggerListener = targetClass.getMethod("addGlobalTriggerListener", new Class[] {
org/quartz/TriggerListener
});
TriggerListener atriggerlistener1[] = globalTriggerListeners;
int j1 = atriggerlistener1.length;
for(int j2 = 0; j2 < j1; j2++)
{
TriggerListener listener = atriggerlistener1[j2];
if(quartz2)
{
List emptyMatchers = new LinkedList();
//添加觸發器監聽器
ReflectionUtils.invokeMethod(addTriggerListener, target, new Object[] {
listener, emptyMatchers
});
} else
{
//添加全局觸發器監聽器
ReflectionUtils.invokeMethod(addTriggerListener, target, new Object[] {
listener
});
}
}
}
if(triggerListeners != null)
{
TriggerListener atriggerlistener[] = triggerListeners;
int j = atriggerlistener.length;
for(int k1 = 0; k1 < j; k1++)
{
TriggerListener listener = atriggerlistener[k1];
if(quartz2)
throw new IllegalStateException("Non-global TriggerListeners not supported on Quartz 2 - manually register a Matcher against the Quartz ListenerManager instead");
//添加觸發器監聽器
getScheduler().addTriggerListener(listener);
}
}
}
catch(NoSuchMethodException ex)
{
throw new IllegalStateException((new StringBuilder()).append("Expected Quartz API not present: ").append(ex).toString());
}
}
//添加job和trriger
protected void registerJobsAndTriggers()
throws SchedulerException
{
TransactionStatus transactionStatus = null;
if(transactionManager != null)
transactionStatus = transactionManager.getTransaction(new DefaultTransactionDefinition());
try
{
if(jobSchedulingDataLocations != null)
{
ClassLoadHelper clh = new ResourceLoaderClassLoadHelper(resourceLoader);
clh.initialize();
XMLSchedulingDataProcessor dataProcessor = new XMLSchedulingDataProcessor(clh);
String as[] = jobSchedulingDataLocations;
int i = as.length;
for(int j = 0; j < i; j++)
{
String location = as[j];
dataProcessor.processFileAndScheduleJobs(location, getScheduler());
}
}
if(jobDetails != null)
{
JobDetail jobDetail;
//如果jobDetails不為空，則添加jobDetails到調度器
for(Iterator iterator = jobDetails.iterator(); iterator.hasNext(); addJobToScheduler(jobDetail))
jobDetail = (JobDetail)iterator.next();
} else
{
jobDetails = new LinkedList();
}
if(calendars != null)
{
String calendarName;
Calendar calendar;
for(Iterator iterator1 = calendars.keySet().iterator(); iterator1.hasNext(); getScheduler().addCalendar(calendarName, calendar, true, true))
{
calendarName = (String)iterator1.next();
calendar = (Calendar)calendars.get(calendarName);
}
}
if(triggers != null)
{
//添加TrrigerWapprer觸發任務到調度器
Trigger trigger;
for(Iterator iterator2 = triggers.iterator(); iterator2.hasNext(); addTriggerToScheduler(trigger))
trigger = (Trigger)iterator2.next();
}
}
catch(Throwable ex)
{
if(transactionStatus != null)
try
{
transactionManager.rollback(transactionStatus);
}
catch(TransactionException tex)
{
logger.error("Job registration exception overridden by rollback exception", ex);
throw tex;
}
if(ex instanceof SchedulerException)
throw (SchedulerException)ex;
if(ex instanceof Exception)
throw new SchedulerException((new StringBuilder()).append("Registration of jobs and triggers failed: ").append(ex.getmessage()).toString(), ex);
else
throw new SchedulerException((new StringBuilder()).append("Registration of jobs and triggers failed: ").append(ex.getMessage()).toString());
}
if(transactionStatus != null)
transactionManager.commit(transactionStatus);
}
//添加jobDetail到調度器
private boolean addJobToScheduler(JobDetail jobDetail)
throws SchedulerException
{
if(overwriteExistingJobs || !jobDetailExists(jobDetail))
{
getScheduler().addJob(jobDetail, true);
return true;
} else
{
return false;
}
}
//添加TrrigerWapprer觸發任務到調度器
private boolean addTriggerToScheduler(Trigger trigger)
throws SchedulerException
{
boolean triggerExists = triggerExists(trigger);
if(!triggerExists || overwriteExistingJobs)
{
JobDetail jobDetail = findJobDetail(trigger);
if(jobDetail != null && !jobDetails.contains(jobDetail) && addJobToScheduler(jobDetail))
jobDetails.add(jobDetail);
if(!triggerExists)
try
{
//關鍵在這裏，調度觸發任務
getScheduler().scheduleJob(trigger);
}
catch(ObjectAlreadyExistsException ex)
{
if(logger.isDebugEnabled())
logger.debug((new StringBuilder()).append("Unexpectedly found existing trigger, assumably due to cluster race condition: ").append(ex.getMessage()).append(" - can safely be ignored").toString());
if(overwriteExistingJobs)
//如果允許重寫存在的job，則重新調度觸發任務
rescheduleJob(trigger);
}
else
rescheduleJob(trigger);
return true;
} else
{
return false;
}
}
//根據trigger獲取JobDetail
private JobDetail findJobDetail(Trigger trigger)
{
if(trigger instanceof JobDetailAwareTrigger)
return ((JobDetailAwareTrigger)trigger).getJobDetail();
Map jobDataMap = (Map)ReflectionUtils.invokeMethod(org/quartz/Trigger.getMethod("getJobDataMap", new Class[0]), trigger);
return (JobDetail)jobDataMap.remove("jobDetail");
}
//設置JobDetail
public transient void setJobDetails(JobDetail jobDetails[])
{
this.jobDetails = new ArrayList(Arrays.asList(jobDetails));
}
//設置Trigger
public transient void setTriggers(Trigger triggers[])
{
this.triggers = Arrays.asList(triggers);
}
//設置SchedulerListener
public transient void setSchedulerListeners(SchedulerListener schedulerListeners[])
{
this.schedulerListeners = schedulerListeners;
}
protected abstract Scheduler getScheduler();
private static Class jobKeyClass;
private static Class triggerKeyClass;
protected final Log logger = LogFactory.getLog(getClass());
private boolean overwriteExistingJobs;
private String jobSchedulingDataLocations[];
private List jobDetails;
private Map calendars;
private List triggers;
private SchedulerListener schedulerListeners[];
private JobListener globalJobListeners[];
private JobListener jobListeners[];
private TriggerListener globalTriggerListeners[];
private TriggerListener triggerListeners[];
private PlatformTransactionManager transactionManager;
protected ResourceLoader resourceLoader;
static
{
try
{
jobKeyClass = ClassUtils.forName("org.quartz.JobKey", org/springframework/scheduling/quartz/SchedulerAccessor.getClassLoader());
triggerKeyClass = ClassUtils.forName("org.quartz.TriggerKey", org/springframework/scheduling/quartz/SchedulerAccessor.getClassLoader());
}
}
}
//資源加載器
public interface ResourceLoaderAware
extends Aware
{
public abstract void setResourceLoader(ResourceLoader resourceloader);
}
//事務狀態
public interface TransactionStatus
extends SavepointManager
{
public abstract boolean isNewTransaction();
public abstract boolean hasSavepoint();
public abstract void setRollbackOnly();
public abstract boolean isRollbackOnly();
public abstract boolean isCompleted();
}
//bean工場
public interface FactoryBean
{
public abstract Object getObject()
throws Exception;
public abstract Class getObjectType();
public abstract boolean isSingleton();
}
//beanName設置
public interface BeanNameAware
extends Aware
{
public abstract void setBeanName(String s);
}
//應用上下文
public interface ApplicationContextAware
extends Aware
{
public abstract void setApplicationContext(ApplicationContext applicationcontext)
throws BeansException;
}
//bean初始化
public interface InitializingBean
{
public abstract void afterPropertiesSet()
throws Exception;
}
//bean銷毀
public interface DisposableBean
{
public abstract void destroy()
throws Exception;
}
//生命周期管理
public interface SmartLifecycle
extends Lifecycle, Phased
{
public abstract boolean isAutoStartup();
public abstract void stop(Runnable runnable);
}
//job工廠
public interface JobFactory
{
//TriggerFiredBundle,觸發任務包裝類，Scheduler，調度器
public abstract Job newJob(TriggerFiredBundle triggerfiredbundle, Scheduler scheduler)
throws SchedulerException;
}
// JobDetailAwareTrigger-Spring
public interface JobDetailAwareTrigger
{
public abstract JobDetail getJobDetail();
public static final String JOB_DETAIL_KEY = "jobDetail";
}
總結：
Spring與Quartz的整合關鍵在於SchedulerFactoryBean，我們在XML中配SchedulerFactoryBean時候，如要需要任務持久化，我們需要配置datasource，配置Quartz屬性（線程池， 調度器名，線程池優先級，持久化類等信息），配置jobDetails，配置觸發任務trrigers（TrrigerWapper），延遲加載屬性，監聽器，重寫屬性；SchedulerFactoryBean實現了InitializingBean，在SchedulerFactoryBean屬性初始化完畢，調用afterPropertiesSet方法，完成數據源，Quartz-schedulerFactory初始化，創建調度器，初始化調度器上下文，註冊監聽器，註冊job，Trrigers。
Tags: public false 文章 blog
文章來源：