版本：3.1.0-SNAPSHOT
git地址：https://github.com/apache/shardingsphere-elasticjob
Maven 坐标
<dependency>
<groupId>org.apache.shardingsphere.elasticjob</groupId>
<artifactId>elasticjob-lite-spring-boot-starter</artifactId>
<version>${latest.version}</version>
</dependency>
Spring.factories配置
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.apache.shardingsphere.elasticjob.lite.spring.boot.job.ElasticJobLiteAutoConfiguration
在添加elasticjob-lite-spring-boot-starter启动类的时候，会自动加载ElasticJobLiteAutoConfiguration，接下来看下ElasticJobLiteAutoConfiguration中所做的处理。
ElasticJobLiteAutoConfiguration.java
/** * ElasticJob-Lite auto configuration. */ @Configuration(proxyBeanMethods = false) @AutoConfigureAfter(DataSourceAutoConfiguration.class) /** * elastic job 开关 * elasticjob.enabled.ture默认为true */ @ConditionalOnProperty(name = "elasticjob.enabled", havingValue = "true", matchIfMissing = true) /** * 导入 * ElasticJobRegistryCenterConfiguration.class 注册中心配置 * ElasticJobTracingConfiguration.class job事件追踪配置 * ElasticJobSnapshotServiceConfiguration.class 快照配置 */ @Import({ElasticJobRegistryCenterConfiguration.class, ElasticJobTracingConfiguration.class, ElasticJobSnapshotServiceConfiguration.class}) /** * job相关配置信息 */ @EnableConfigurationProperties(ElasticJobProperties.class) public class ElasticJobLiteAutoConfiguration { @Configuration(proxyBeanMethods = false) /** * ElasticJobBootstrapConfiguration.class 创建job beans 注入spring容器 * ScheduleJobBootstrapStartupRunner.class 执行类型为ScheduleJobBootstrap.class 的job开始运行 */ @Import({ElasticJobBootstrapConfiguration.class, ScheduleJobBootstrapStartupRunner.class}) protected static class ElasticJobConfiguration { } }
Elastic-job 是利用zookeeper 实现分布式job的功能，所以在自动装配的时候，需要有zookeeper注册中心的配置。
自动装配主要做了4件事事
1.配置zookeeper 客户端信息，启动连接zookeeper.
2.配置事件追踪数据库，用于保存job运行记录
3.解析所有job配置文件，将所有job的bean放置在spring 单例bean中
4.识别job类型，在zookeeper节点上处理job节点数据，运行定时任务job.
第一件事：配置zookeeper 客户端信息，启动连接zookeeper.
ZookeeperRegistryCenter.class
public void init() { log.debug("Elastic job: zookeeper registry center init, server lists is: {}.", zkConfig.getServerLists()); CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder() //设置zookeeper 服务器地址 .connectString(zkConfig.getServerLists()) //设置重试机制 .retryPolicy(new ExponentialBackoffRetry(zkConfig.getBaseSleepTimeMilliseconds(), zkConfig.getMaxRetries(), zkConfig.getMaxSleepTimeMilliseconds())) //设置命名空间，zookeeper节点名称 .namespace(zkConfig.getNamespace()); //设置session超时时间 if (0 != zkConfig.getSessionTimeoutMilliseconds()) { builder.sessionTimeoutMs(zkConfig.getSessionTimeoutMilliseconds()); } //设置连接超时时间 if (0 != zkConfig.getConnectionTimeoutMilliseconds()) { builder.connectionTimeoutMs(zkConfig.getConnectionTimeoutMilliseconds()); } if (!Strings.isNullOrEmpty(zkConfig.getDigest())) { builder.authorization("digest", zkConfig.getDigest().getBytes(StandardCharsets.UTF_8)) .aclProvider(new ACLProvider() { @Override public List<ACL> getDefaultAcl() { return ZooDefs.Ids.CREATOR_ALL_ACL; } @Override public List<ACL> getAclForPath(final String path) { return ZooDefs.Ids.CREATOR_ALL_ACL; } }); } client = builder.build(); //zookeeper 客户端开始启动 client.start(); try { //zookeeper 客户端一直连接 if (!client.blockUntilConnected(zkConfig.getMaxSleepTimeMilliseconds() * zkConfig.getMaxRetries(), TimeUnit.MILLISECONDS)) { client.close(); throw new KeeperException.OperationTimeoutException(); } //CHECKSTYLE:OFF } catch (final Exception ex) { //CHECKSTYLE:ON RegExceptionHandler.handleException(ex); } }
第二件事： 配置事件追踪数据库，用于保存job运行记录
ElasticJobTracingConfiguration.java
/** * Create a bean of tracing DataSource. * * @param tracingProperties tracing Properties * @return tracing DataSource */ @Bean("tracingDataSource") //spring中注入bean name 为tracingDataSource的job数据库连接信息 public DataSource tracingDataSource(final TracingProperties tracingProperties) { //获取elastic-job 数据库配置 DataSourceProperties dataSource = tracingProperties.getDataSource(); if (dataSource == null) { return null; } HikariDataSource tracingDataSource = new HikariDataSource(); tracingDataSource.setJdbcUrl(dataSource.getUrl()); BeanUtils.copyProperties(dataSource, tracingDataSource); return tracingDataSource; } /** * Create a bean of tracing configuration. * * @param dataSource required by constructor * @param tracingDataSource tracing ataSource * @return a bean of tracing configuration */ @Bean @ConditionalOnBean(DataSource.class) @ConditionalOnProperty(name = "elasticjob.tracing.type", havingValue = "RDB") public TracingConfiguration<DataSource> tracingConfiguration(final DataSource dataSource, @Nullable final DataSource tracingDataSource) { /** * dataSource 是业务数据库 * tracingDataSource 是job数据库 * 当配置elasticjob.tracing.type = RDB时，如果单独配置job数据库是，默认使用job数据库作为job运行轨迹的记录 * 但这边同时业务数据库和job追踪数据库同时注入是，mybatis-plus 结合@Table 使用的时候，很有可能找不到正确对应的数据源 */ DataSource ds = tracingDataSource; if (ds == null) { ds = dataSource; } return new TracingConfiguration<>("RDB", ds); }
通过elasticjob.tracing.type=RDB的配置开启事件追踪功能，这边job的事件追踪数据源可以和业务数据源配置不一样。
第三件事：解析所有job配置文件
ElasticJobBootstrapConfiguration.class
public void createJobBootstrapBeans() { //获取job配置 ElasticJobProperties elasticJobProperties = applicationContext.getBean(ElasticJobProperties.class); //获取单利注册对象 SingletonBeanRegistry singletonBeanRegistry = ((ConfigurableApplicationContext) applicationContext).getBeanFactory(); //获取注入zookeeper 客户端 CoordinatorRegistryCenter registryCenter = applicationContext.getBean(CoordinatorRegistryCenter.class); //获取job事件追踪 TracingConfiguration<?> tracingConfig = getTracingConfiguration(); //构造JobBootstraps constructJobBootstraps(elasticJobProperties, singletonBeanRegistry, registryCenter, tracingConfig); }
重要的是constructJobBootstraps 这个方法，来看下
private void constructJobBootstraps(final ElasticJobProperties elasticJobProperties, final SingletonBeanRegistry singletonBeanRegistry, final CoordinatorRegistryCenter registryCenter, final TracingConfiguration<?> tracingConfig) { //遍历配置的每一个job for (Map.Entry<String, ElasticJobConfigurationProperties> entry : elasticJobProperties.getJobs().entrySet()) { ElasticJobConfigurationProperties jobConfigurationProperties = entry.getValue(); //校验 job class 和 type 都为空抛异常 Preconditions.checkArgument(null != jobConfigurationProperties.getElasticJobClass() || !Strings.isNullOrEmpty(jobConfigurationProperties.getElasticJobType()), "Please specific [elasticJobClass] or [elasticJobType] under job configuration."); //校验 job class 和 type 都有 报相互排斥 Preconditions.checkArgument(null == jobConfigurationProperties.getElasticJobClass() || Strings.isNullOrEmpty(jobConfigurationProperties.getElasticJobType()), "[elasticJobClass] and [elasticJobType] are mutually exclusive."); if (null != jobConfigurationProperties.getElasticJobClass()) { //通过class 注入job registerClassedJob(entry.getKey(), entry.getValue().getJobBootstrapBeanName(), singletonBeanRegistry, registryCenter, tracingConfig, jobConfigurationProperties); } else if (!Strings.isNullOrEmpty(jobConfigurationProperties.getElasticJobType())) { //通过type 注入job registerTypedJob(entry.getKey(), entry.getValue().getJobBootstrapBeanName(), singletonBeanRegistry, registryCenter, tracingConfig, jobConfigurationProperties); } } }
Job 有两种类型的注入，第一种是是class,配置成job的全路径信息注入
再来看看registerClassedJob 方法里的内容
private void registerClassedJob(final String jobName, final String jobBootstrapBeanName, final SingletonBeanRegistry singletonBeanRegistry, final CoordinatorRegistryCenter registryCenter, final TracingConfiguration<?> tracingConfig, final ElasticJobConfigurationProperties jobConfigurationProperties) { //获取job配置 JobConfiguration jobConfig = jobConfigurationProperties.toJobConfiguration(jobName); //配置job事件追踪 jobExtraConfigurations(jobConfig, tracingConfig); //获取job类型 ElasticJob elasticJob = applicationContext.getBean(jobConfigurationProperties.getElasticJobClass()); //没有配置cron表达式 就初始化为OneOffJobBootstrap对象，一次性任务 if (Strings.isNullOrEmpty(jobConfig.getCron())) { Preconditions.checkArgument(!Strings.isNullOrEmpty(jobBootstrapBeanName), "The property [jobBootstrapBeanName] is required for One-off job."); singletonBeanRegistry.registerSingleton(jobBootstrapBeanName, new OneOffJobBootstrap(registryCenter, elasticJob, jobConfig)); } else { //有配置cron表达式 就初始化为ScheduleJobBootstrap对象，定时任务 //设置bean name String beanName = !Strings.isNullOrEmpty(jobBootstrapBeanName) ? jobBootstrapBeanName : jobConfig.getJobName() + "ScheduleJobBootstrap"; //注入ScheduleJobBootstrap对象为单利对象 singletonBeanRegistry.registerSingleton(beanName, new ScheduleJobBootstrap(registryCenter, elasticJob, jobConfig)); } }
Class 类型注入的job有两种类型
1.ScheduleJobBootstrap：定时任务类型的job。
2.OneOffJobBootstrap：一定次job类型。
看下定义的new ScheduleJobBootstrap 方法
public JobScheduler(final CoordinatorRegistryCenter regCenter, final ElasticJob elasticJob, final JobConfiguration jobConfig) { Preconditions.checkArgument(null != elasticJob, "Elastic job cannot be null."); this.regCenter = regCenter; //获取job监听器 Collection<ElasticJobListener> jobListeners = getElasticJobListeners(jobConfig); // 集成所有操作zookeeper 节点的services,job 监听器 setUpFacade = new SetUpFacade(regCenter, jobConfig.getJobName(), jobListeners); //获取当前job名称 String jobClassName = JobClassNameProviderFactory.getProvider().getJobClassName(elasticJob); //zookeeper节点 {namespace}/{jobclassname}/config 放置job配置信息 this.jobConfig = setUpFacade.setUpJobConfiguration(jobClassName, jobConfig); // 集成所有操作zookeeper 节点的services schedulerFacade = new SchedulerFacade(regCenter, jobConfig.getJobName()); jobFacade = new LiteJobFacade(regCenter, jobConfig.getJobName(), jobListeners, findTracingConfiguration().orElse(null)); //检验job配置 validateJobProperties(); //定义job执行器 jobExecutor = new ElasticJobExecutor(elasticJob, this.jobConfig, jobFacade); //监听器里注入GuaranteeService setGuaranteeServiceForElasticJobListeners(regCenter, jobListeners); //创建定时任务，开始执行 jobScheduleController = createJobScheduleController(); }
看下createJobScheduleController文章来源：https://www.toymoban.com/news/detail-426170.html
private JobScheduleController createJobScheduleController() { JobScheduleController result = new JobScheduleController(createScheduler(), createJobDetail(), getJobConfig().getJobName()); //注册job JobRegistry.getInstance().registerJob(getJobConfig().getJobName(), result); //注册器开始运行 registerStartUpInfo(); return result; }
看下registerStartUpInfo方法文章来源地址https://www.toymoban.com/news/detail-426170.html
public void registerStartUpInfo(final boolean enabled) { //开始所有的监听器 listenerManager.startAllListeners(); //选举leader /{namespace}/leader/election/instance 放置选举出来的服务器 leaderService.electLeader(); //{namespace}/{ipservers} 设置enable处理 serverService.persistOnline(enabled); //临时节点 /{namespave}/instances 放置运行服务实例信息 instanceService.persistOnline(); //开启一个异步服务 if (!reconcileService.isRunning()) { reconcileService.startAsync(); } }
这里实行的操作：
1.开启所有监听器处理
2.leader选举
3.持久化节点数据
4.开启异步服务
第四步：4.识别job类型，在zookeeper节点上处理job节点数据，运行定时任务job.
@Override public void run(final String... args) { log.info("Starting ElasticJob Bootstrap."); applicationContext.getBeansOfType(ScheduleJobBootstrap.class).values().forEach(ScheduleJobBootstrap::schedule); log.info("ElasticJob Bootstrap started."); }
获取到所有的定时任务job(ScheduleJobBootstrap类型)，执行schedule方法，底层实际使用quartz框架运行定时任务。
到了这里，关于elastic-job源码（1）- job自动装配的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
