SpringBoot整合elastic-job实现分布式任务调度

迷南。 2022-10-11 15:54 275阅读 0赞

### SpringBoot整合elastic-job ###

*  maven依赖添加
 *  application.properties文件配置
 *  zookeeper 注册中心
 *  Abstract
 *  具体的job
 *  elasticJob 管理后台

ElasticJob 是一个非常不错的分布式定时任务调度开源组件。使用它实现定时任务简单又方便，网上有很多种做法，今天记录一下我在项目中的一种简单的使用方式。  
官网地址： [http://elasticjob.io/][http_elasticjob.io]

# maven依赖添加 #

<dependency>
        <groupId>com.dangdang</groupId>
        <artifactId>elastic-job-lite-core</artifactId>
        <version>2.1.5</version>
    </dependency>
    <dependency>
        <groupId>com.dangdang</groupId>
        <artifactId>elastic-job-lite-spring</artifactId>
        <version>2.1.5</version>
    </dependency>
    
    <dependency>
        <groupId>com.dangdang</groupId>
        <artifactId>elastic-job-lite-lifecycle</artifactId>
        <version>2.1.5</version>
    </dependency>

核心依赖就两个包，如果你想对任务更好的控制可以把第三个依赖也加上。

# application.properties文件配置 #

#新加
    elastic.job.zk.address=service01.zk.test8.oneplus.cn:2181,service02.zk.test8.oneplus.cn:2181,service03.zk.test8.oneplus.cn:2181
    elastic.job.zk.namespace=routerstorage_job
    elastic.job.zk.routerstorage.sleep.mill=10000
    elastic.job.zk.max.sleep.mill=30000
    elastic.job.zk.max.retries=5
    elastic.job.routerstorage.sharding.total=4
    elastic.job.routerstorage.sharding.overwrite=false
    elastic.job.routerstorage.consistency.cron=0 0/2 * * * ?
    elastic.shardingItemParameters =
    
    elastic.job.routerstorage.sharding.backup.total=2
    elastic.job.routerstorage.sharding.backup.overwrite=false
    elastic.job.routerstorage.consistency.backup.cron=0 0/5 * * * ?
    elastic.backup.shardingItemParameters =
    elastic.backup.jobParameter =

# zookeeper 注册中心 #

package com.oneplus.routerstorage.facade.config;
    
    import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
    import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /** * @author evanYang * @version 1.0 * @date 2020/12/2 17:25 */
    @Configuration
    public class ElasticZkConfig { 
    
    
        @Bean(initMethod = "init")
        public ZookeeperRegistryCenter regCenter(@Value("${elastic.job.zk.address}") final String serverList,
                                                 @Value("${elastic.job.zk.namespace}") final String namespace,
                                                 @Value("${elastic.job.zk.routerstorage.sleep.mill}") Integer baseSleepTimeMilliseconds ,
                                                 @Value("${elastic.job.zk.max.sleep.mill}") Integer maxSleepTimeMilliseconds,
                                                 @Value("${elastic.job.zk.max.retries}")    Integer maxRetries   ) { 
            ZookeeperConfiguration zookeeperConfiguration = new ZookeeperConfiguration(serverList, namespace);
            zookeeperConfiguration.setMaxRetries(maxRetries);
            zookeeperConfiguration.setMaxSleepTimeMilliseconds(maxSleepTimeMilliseconds);
            zookeeperConfiguration.setBaseSleepTimeMilliseconds(baseSleepTimeMilliseconds);
            return new ZookeeperRegistryCenter(zookeeperConfiguration);
        }
    
    }

# Abstract #

为了更方便的创建任务类，我们定义了一个抽象类AbstractJob，抽象类包含了这个任务需要的主要参数，如任务执行时间的表达式、任务分片总数、任务分片参数、任务参数等。AbstraJob实现SimpleJob接口，代码如下：

package com.oneplus.routerstorage.facade.job;
    
    import com.dangdang.ddframe.job.api.ShardingContext;
    import com.dangdang.ddframe.job.api.simple.SimpleJob;
    import com.dangdang.ddframe.job.config.JobCoreConfiguration;
    import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
    import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
    import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
    import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
    import lombok.Data;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /** * @author evanYang * @version 1.0 * @date 2021/7/1 17:23 */
    @Data
    public abstract class AbstractJob implements SimpleJob { 
        protected static final Logger logger = LoggerFactory.getLogger("job");
    
        /** 任务执行时间表达式 */
        private String cron;
        /** 任务分片总数 */
        private Integer shardingTotalCount;
        /** 任务分片参数 */
        private String shardingItemParameters;
        /** 任务参数 */
        private String jobParameter;
        /** 是否已经启动 */
        private boolean started = false;
    
        public AbstractJob() { 
            super();
        }
    
        /** * 构造函数 * * @param cron * @param shardingTotalCount * @param shardingItemParameters * @param jobParameter */
        public AbstractJob(String cron, Integer shardingTotalCount,
                           String shardingItemParameters, String jobParameter) { 
            super();
            this.cron = cron;
            this.shardingTotalCount = shardingTotalCount;
            this.shardingItemParameters = shardingItemParameters;
            this.jobParameter = jobParameter;
        }
    
        /** * 启动任务,调用此方法，将启动任务调度 * @param zookeeperRegistryCenter 任务注册中心 */
        protected synchronized void start(ZookeeperRegistryCenter zookeeperRegistryCenter) { 
            // 保证一个任务只执行一次start
            if (this.started){ 
                return;
            }
            this.started = true;
            SpringJobScheduler s = new SpringJobScheduler(this,
                    zookeeperRegistryCenter, getLiteJobConfiguration());
            s.init();
        }
    
        @Override
        public void execute(ShardingContext shardingContext) { 
            logger.info(String.format("Thread ID:%s,任务总片数:%s," + "当前分片项:%s,当前参数:%s,"
                            + "当前任务名称:%s,当前任务参数:%s", Thread.currentThread().getId(),
                    shardingContext.getShardingTotalCount(),
                    shardingContext.getShardingItem(),
                    shardingContext.getShardingParameter(),
                    shardingContext.getJobName(), shardingContext.getJobParameter()));
            executeJob(shardingContext.getShardingTotalCount(),
                    shardingContext.getShardingItem(),
                    shardingContext.getShardingParameter(),
                    shardingContext.getJobParameter());
        }
    
        /** * 执行任务 * * @param shardingTotalCount * 任务分片数 * @param shardingItem * 当前分配序号 * @param itemParameter * 当前分配任务参数 */
        public abstract void executeJob(Integer shardingTotalCount,
                                        Integer shardingItem, String itemParameter, String jobParameter);
    
        /** * 获取任务配置 * @return */
        protected LiteJobConfiguration getLiteJobConfiguration() { 
            return LiteJobConfiguration
                    .newBuilder(
                            new SimpleJobConfiguration(JobCoreConfiguration
                                    .newBuilder(this.getClass().getName(),
                                            this.cron, this.shardingTotalCount)
                                    .shardingItemParameters(
                                            this.shardingItemParameters)
                                    .jobParameter(this.jobParameter).build(), this
                                    .getClass().getCanonicalName()))
                    .overwrite(true).build();
        }
    
    
    }

# 具体的job #

抽象类中的excute会打印当前任务的一些线程信息及任务分片信息，子类只需要继承抽象类，实现executeJob方法即可。

由于我们是基于springboot，为了让应用启动后，任务就自动注册，我们只需要在任务类上加入@Component，在任务类bean初始化完成后，执行一个init方法即可开启任务，这里用一个CodisConsistencyBackUpJob为例，如下

package com.oneplus.routerstorage.facade.job;
    
    import com.dangdang.ddframe.job.api.ShardingContext;
    import com.dangdang.ddframe.job.api.simple.SimpleJob;
    import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
    import com.oneplus.routerstorage.core.service.CodisConsistencyService;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.PostConstruct;
    
    /** * @author evanYang * @version 1.0 * @date 2020/12/2 17:15 */
    @Component
    public class CodisConsistencyBackUpJob extends AbstractJob { 
        Logger logger = LoggerFactory.getLogger(CodisConsistencyBackUpJob.class);
    
        @Autowired
        private ZookeeperRegistryCenter zookeeperRegistryCenter;
    
        public CodisConsistencyBackUpJob(
                @Value("${elastic.job.routerstorage.consistency.backup.cron}") String cron,
                @Value("${elastic.job.routerstorage.sharding.backup.total}") Integer shardingTotalCount,
                @Value("${elastic.backup.shardingItemParameters}") String shardingItemParameters,
                @Value("${elastic.backup.jobParameter}") String jobParameter) { 
            super(cron, shardingTotalCount, shardingItemParameters, jobParameter);
        }
    
        //@Autowired
        //private CodisConsistencyService codisConsistencyService;
        //@Override
        //public void execute(ShardingContext shardingContext) { 
        // logger.info(String.format("Thread ID: %s, 作业分片总数: %s, " + "当前分片项: %s", Thread.currentThread().getId(),
        // shardingContext.getShardingTotalCount(), shardingContext.getShardingItem()
        // ));
        // codisConsistencyService.codisConsistency(shardingContext);
        //}
    
        @PostConstruct
        public void init() { 
            start(zookeeperRegistryCenter);
        }
    
        @Override
        public void executeJob(Integer shardingTotalCount, Integer shardingItem,
                               String itemParameter, String jobParameter) { 
            logger.info("这是一个测试任务");
        }
    }

# elasticJob 管理后台 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zMDQwOTkyNw_size_16_color_FFFFFF_t_70]

[http_elasticjob.io]: http://elasticjob.io/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zMDQwOTkyNw_size_16_color_FFFFFF_t_70]: /images/20221005/e4378cc3864744818c227c985d1d5b28.png