elastic-job源码（1）- job自动装配

版本：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

private JobScheduleController createJobScheduleController() {
    JobScheduleController result = new JobScheduleController(createScheduler(), createJobDetail(), getJobConfig().getJobName());
    //注册job
    JobRegistry.getInstance().registerJob(getJobConfig().getJobName(), result);
    //注册器开始运行
    registerStartUpInfo();
    return result;
}

看下registerStartUpInfo方法

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自动装配 - wjdxw

elastic-job源码（1）- job自动装配

Recommend

BitNile扩大与Core Scientific合作，5月起运作1万台比特币矿机

本田与台积电达成战略合作，以确保半导体稳定采购

ChatGPT now allows disabling chat history, declining training, and exporting dat...

watchOS 10 release date

AMD发布关于Ryzen 7000X3D烧坏问题的声明：正在积极调查，电压需在安全范围

Wealthiest People in Hong Kong (April 26, 2023)

Google I/O 2023

How Midsize Companies Can Access Capital in Turbulent Times

亚马逊美国站的销售税以及免税州介绍-跨境头条-AMZ123亚马逊导航-跨境电商出海门户

深圳智能装备厂商“铭创智能”完成B轮融资

About Joyk