上期介绍了job 配置相关信息,
今天介绍ej 如何通过Spring容器调度job.
在通过Spring 的xml配置job信息时,会引入xmlns:job="http://www.dangdang.com/schema/ddframe/job"
在ej 中, 有一个spring.handlers ,配置了分布式作业的命名空间处理器.
http\://www.dangdang.com/schema/ddframe/job=io.elasticjob.lite.spring.job.handler.JobNamespaceHandler
对Simple, Dataflow, ScriptJob 三种job 进行解析,注入Spring IOC.
public final class JobNamespaceHandler extends NamespaceHandlerSupport {
@Override
public void init() {
registerBeanDefinitionParser("simple", new SimpleJobBeanDefinitionParser());
registerBeanDefinitionParser("dataflow", new DataflowJobBeanDefinitionParser());
registerBeanDefinitionParser("script", new ScriptJobBeanDefinitionParser());
}
}
今天从 SimpleJobBeanDefinitionParser :简单作业的命名空间解析器为例说起:
抽象出 基本作业的命名空间解析器. AbstractJobBeanDefinitionParser, 继承自 Spring 的AbstractBeanDefinitionParser.
当Spring 容器启动时,会直接解析并注入容器中.
在AbstractJobBeanDefinitionParser中:
在factory中初始化SpringJobScheduler.class, 并调用 init方法.
public abstract class AbstractJobBeanDefinitionParser extends AbstractBeanDefinitionParser {
@Override
protected AbstractBeanDefinition parseInternal(final Element element, final ParserContext parserContext) {
BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(SpringJobScheduler.class);
factory.setInitMethodName("init");
.............
}
在SpringJobScheduler.class(基于Spring的作业启动器.), 继承自 JobScheduler
在JobScheduler(作业调度器.) 中, init 方法:
/**
* 初始化作业.
*/
public void init() {
LiteJobConfiguration liteJobConfigFromRegCenter = schedulerFacade.updateJobConfiguration(liteJobConfig);
JobRegistry.getInstance().setCurrentShardingTotalCount(liteJobConfigFromRegCenter.getJobName(), liteJobConfigFromRegCenter.getTypeConfig().getCoreConfig().getShardingTotalCount());
JobScheduleController jobScheduleController = new JobScheduleController(
createScheduler(), createJobDetail(liteJobConfigFromRegCenter.getTypeConfig().getJobClass()), liteJobConfigFromRegCenter.getJobName());
JobRegistry.getInstance().registerJob(liteJobConfigFromRegCenter.getJobName(), jobScheduleController, regCenter);
schedulerFacade.registerStartUpInfo(!liteJobConfigFromRegCenter.isDisabled());
jobScheduleController.scheduleJob(liteJobConfigFromRegCenter.getTypeConfig().getCoreConfig().getCron());
}
分析一下 init():
1. LiteJobConfiguration liteJobConfigFromRegCenter = schedulerFacade.updateJobConfiguration(liteJobConfig);
从schedulerFacade(为调度器提供内部服务的门面类.)中调用 updateJobConfiguration(更新作业配置.)方法,获得最新的作业配置.
2.JobRegistry(作业注册表.) 重新设置分片总数,job 名称等属性信息. 这是个单例设计,采用双重锁校验模式.
public static JobRegistry getInstance() {
if (null == instance) {
synchronized (JobRegistry.class) {
if (null == instance) {
instance = new JobRegistry();
}
}
}
return instance;
}
3.初始化 JobScheduleController(作业调度控制器.)
4. 添加作业调度控制器.
5.schedulerFacade.registerStartUpInfo(!liteJobConfigFromRegCenter.isDisabled()); 注册作业启动信息.
6. jobScheduleController.scheduleJob(liteJobConfigFromRegCenter.getTypeConfig().getCoreConfig().getCron()); 调度作业.
public void scheduleJob(final String cron) {
try {
if (!scheduler.checkExists(jobDetail.getKey())) {
scheduler.scheduleJob(jobDetail, createTrigger(cron));
}
scheduler.start();
} catch (final SchedulerException ex) {
throw new JobSystemException(ex);
}
}
在调度作业时, 进行调度的启动. ej 底层,是基于quartz 进行封装的.这里的调度器也是quartz 中的Scheduler.
此节完.
