SpringBoot整合Elastic-job实现

ゞ浴缸里的玫瑰 2021-12-04 02:45 483阅读 0赞

> ### SpringBoot整合Elastic-job实现 ###

** 【基本整合】：原理参考：**[Elastic-Job原理][Elastic-Job]

**（1）引用pom依赖：**

<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>

**（2） 定义Zookeeper连接**

package com.caox.elasticJob;
    
    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.boot.autoconfigure.condition.ConditionalOnExpression;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/7/16 19:07
     */
    @Configuration
    @ConditionalOnExpression("'${regCenter.serverList}'.length() > 0")
    public class JobRegistryCenterConfig {
        @Bean(initMethod = "init")
        public ZookeeperRegistryCenter regCenter(@Value("${regCenter.serverList}") final String serverList,
                                                 @Value("${regCenter.namespace}") final String namespace) {
            return new ZookeeperRegistryCenter(new ZookeeperConfiguration(serverList, namespace));
        }
    }

**（3）定义任务监听器，统计每次任务执行的时间 **

package com.dalaoyang.listen;
    
    import com.dangdang.ddframe.job.executor.ShardingContexts;
    import com.dangdang.ddframe.job.lite.api.listener.ElasticJobListener;
    import org.apache.commons.lang3.StringUtils;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/7/17 13:23
     */
    public class MyElasticJobListener implements ElasticJobListener {
        private static final Logger logger = LoggerFactory.getLogger(MyElasticJobListener.class);
    
        /**
         *  长日期格式
         */
        public static String TIME_FORMAT = "yyyy-MM-dd HH:mm:ss";
        private long beginTime = 0;
        @Override
        public void beforeJobExecuted(ShardingContexts shardingContexts) {
            beginTime = System.currentTimeMillis();
    
            logger.info("===>{} JOB BEGIN TIME: {} <===",shardingContexts.getJobName(), beginTime);
        }
    
        @Override
        public void afterJobExecuted(ShardingContexts shardingContexts) {
            long endTime = System.currentTimeMillis();
            logger.info("===>{} JOB END TIME: {},TOTAL CAST: {} <===",shardingContexts.getJobName(),
                    endTime, endTime - beginTime);
        }
    
        /**
         * 将长整型数字转换为日期格式的字符串
         *
         * @param time
         * @param format
         * @return
         */
        public static String convert2String(long time, String format) {
            if (time > 0l) {
                if (StringUtils.isBlank(format))
                    format = TIME_FORMAT;
    
                SimpleDateFormat sf = new SimpleDateFormat(format);
                Date date = new Date(time);
    
                return sf.format(date);
            }
            return "";
        }
    }

**（4）配置JobConfiuration，配置job随容器一起启动 **

package com.dalaoyang.config;
    
    import com.dalaoyang.job.MySimpleJob;
    import com.dalaoyang.listen.MyElasticJobListener;
    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.api.JobScheduler;
    import com.dangdang.ddframe.job.lite.api.listener.ElasticJobListener;
    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 org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class MyJobConfig {
    
        private final String cron = "0/5 * * * * ?";
        private final int shardingTotalCount = 3;
        private final String shardingItemParameters = "0=A,1=B,2=C";
        private final String jobParameters = "parameter";
    
        @Autowired
        private ZookeeperRegistryCenter regCenter;
    
        /**
         * 配置任务监听器
         * @return
         */
        @Bean
        public ElasticJobListener elasticJobListener() {
            return new MyElasticJobListener();
        }
    
        @Bean
        public SimpleJob stockJob() {
            return new MySimpleJob();
        }
    
        @Bean(initMethod = "init")
        public JobScheduler simpleJobScheduler(final SimpleJob simpleJob) {
            // 监听job启动和完毕时间
            MyElasticJobListener elasticJobListener = new MyElasticJobListener();
            return new SpringJobScheduler(simpleJob, regCenter, getLiteJobConfiguration(simpleJob.getClass(),
                    cron, shardingTotalCount, shardingItemParameters, jobParameters),elasticJobListener);
        }
    
        private LiteJobConfiguration getLiteJobConfiguration(final Class<? extends SimpleJob> jobClass,
                                                             final String cron,
                                                             final int shardingTotalCount,
                                                             final String shardingItemParameters,
                                                             final String jobParameters) {
            // 定义作业核心配置
            JobCoreConfiguration simpleCoreConfig = JobCoreConfiguration.newBuilder(jobClass.getName(), cron, shardingTotalCount).
                    shardingItemParameters(shardingItemParameters).jobParameter(jobParameters).build();
            // 定义SIMPLE类型配置
            SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(simpleCoreConfig, jobClass.getCanonicalName());
            // 定义Lite作业根配置
            LiteJobConfiguration simpleJobRootConfig = LiteJobConfiguration.newBuilder(simpleJobConfig).overwrite(true).build();
            return simpleJobRootConfig;
    
        }
    }

**（5）定义job **

package com.dalaoyang.job;
    
    import com.dangdang.ddframe.job.api.ShardingContext;
    import com.dangdang.ddframe.job.api.simple.SimpleJob;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    public class MySimpleJob implements SimpleJob {
        Logger logger = LoggerFactory.getLogger(MySimpleJob.class);
    
        @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()
            ));
    
        }
    }

**（6）配置文件**

spring.application.name=springboot2_elasticjob
    
    regCenter.serverList=localhost:2181
    regCenter.namespace=springboot2_elasticjob
    
    spring.datasource.jdbc.url=jdbc:mysql://localhost:3306/batch_log?useUnicode=true&amp&characterEncoding=UTF-8
    spring.datasource.jdbc.username=root
    spring.datasource.jdbc.password=123456
    #spring.datasource.jdbc.driverClass=com.mysql.jdbc.Driver

**【采用注解方式改进】： **

** （1）配置任务Zookeeper注册中心、任务监听器、事件追踪持久化DB等**

package com.dalaoyang.config;
    
    
    import com.dalaoyang.listen.MyElasticJobListener;
    import com.dangdang.ddframe.job.event.JobEventConfiguration;
    import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration;
    import com.dangdang.ddframe.job.lite.api.listener.ElasticJobListener;
    import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
    import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import javax.sql.DataSource;
    
    @Configuration
    @ConditionalOnExpression("'${regCenter.serverList}'.length() > 0")
    
    public class JobRegistryCenterConfig {
    
        @Autowired
        private DataSource dataSource;
    
        @Bean(initMethod = "init")
        public ZookeeperRegistryCenter regCenter(@Value("${regCenter.serverList}") final String serverList,
                                                 @Value("${regCenter.namespace}") final String namespace) {
            return new ZookeeperRegistryCenter(new ZookeeperConfiguration(serverList, namespace));
        }
    
        /**
         * 配置任务监听器
         * @return
         */
        @Bean
        public ElasticJobListener elasticJobListener() {
            return new MyElasticJobListener();
        }
    
        /**
         * 将作业运行的痕迹进行持久化到DB
         */
        @Bean
        public JobEventConfiguration jobEventConfiguration(){
            return new JobEventRdbConfiguration(dataSource);
        }
    
    }

**（2）定义数据源：**

package com.dalaoyang.annotation;
    
    import com.alibaba.druid.pool.DruidDataSource;
    import lombok.*;
    import org.springframework.boot.context.properties.ConfigurationProperties;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.Primary;
    
    import javax.sql.DataSource;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/7/18 17:43
     */
    @Getter
    @Setter
    @NoArgsConstructor
    @AllArgsConstructor
    @ToString
    @Configuration
    @ConfigurationProperties(prefix = "spring.datasource.jdbc")
    public class DataSourceProperties {
    
        private String url;
        private String username;
        private String password;
    
        @Bean
        @Primary
        public DataSource getDataSource() {
            DruidDataSource dataSource = new DruidDataSource();
            dataSource.setUrl(url);
            dataSource.setUsername(username);
            dataSource.setPassword(password);
            return dataSource;
        }
    }

**（3）.添加ElasticScheduler注解 **

package com.dalaoyang.annotation;
    
    import org.springframework.stereotype.Component;
    
    import java.lang.annotation.ElementType;
    import java.lang.annotation.Retention;
    import java.lang.annotation.RetentionPolicy;
    import java.lang.annotation.Target;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/7/17 15:03
     */
    @Component
    @Target({ ElementType.TYPE })
    @Retention(RetentionPolicy.RUNTIME)
    public @interface ElasticScheduler {
        /**
         * 任务名称
         * @return
         */
        String name();
    
        /**
         * cron表达式，用于控制作业触发时间
         * @return
         */
        String cron() default "";
    
        /**
         * 分片参数
         * @return
         */
        String shardingItemParameters() default "";
    
        /**
         * 总分片数
         * @return
         */
        int shardingTotalCount();
    
        /**
         * 任务描述信息
         * @return
         */
        String description() default "";
    
        /**
         * 任务参数
         */
        String jobParameters() default "";
    }

**（4）抽象添加job方法：**

package com.dalaoyang.annotation;
    
    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.event.JobEventConfiguration;
    import com.dangdang.ddframe.job.lite.api.listener.ElasticJobListener;
    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 org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/7/17 15:01
     * 抽象添加job方法
     */
    @Component
    public class ElasticJobHandler {
        @Autowired
        private ZookeeperRegistryCenter regCenter;
        /**
         * 监听job启动和完毕时间
         */
        @Resource
        private ElasticJobListener elasticJobListener;
    
        /**
         * 事件追踪
         */
        @Resource
        private JobEventConfiguration jobEventConfiguration;
    
    
        /**
         * @Description 任務配置類
         */
        private LiteJobConfiguration getLiteJobConfiguration(final Class<? extends SimpleJob> jobClass,
                                                             final String cron,
                                                             final int shardingTotalCount,
                                                             final String shardingItemParameters,
                                                             final String jobParameters) {
            return LiteJobConfiguration.newBuilder(new SimpleJobConfiguration(
                    JobCoreConfiguration.newBuilder(jobClass.getName(), cron, shardingTotalCount)
                            .shardingItemParameters(shardingItemParameters).jobParameter(jobParameters).build()
                    , jobClass.getCanonicalName())
            ).overwrite(true).build();
        }
    
        /**
         * 抽象添加job方法
         * @param simpleJob
         * @param cron
         * @param shardingTotalCount
         * @param shardingItemParameters
         * @throws IllegalAccessException
         * @throws InstantiationException
         */
        public void addJob(final SimpleJob simpleJob,
                           final String cron,
                           final Integer shardingTotalCount,
                           final String shardingItemParameters,
                           final String jobParameters)
                throws IllegalAccessException, InstantiationException {
    
            LiteJobConfiguration jobConfig =
                    getLiteJobConfiguration(simpleJob.getClass(), cron, shardingTotalCount, shardingItemParameters,jobParameters);
    
            new SpringJobScheduler(simpleJob, regCenter, jobConfig, jobEventConfiguration, elasticJobListener).init();
        }
    }

**（5） 定义扫描方法：**

package com.dalaoyang.annotation;
    
    import com.dangdang.ddframe.job.api.simple.SimpleJob;
    import org.springframework.beans.BeansException;
    import org.springframework.beans.factory.InitializingBean;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.ApplicationContextAware;
    import org.springframework.core.annotation.AnnotationUtils;
    import org.springframework.stereotype.Component;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/7/17 17:13
     * 定义扫描方法
     */
    @Component
    public class ElasticSchedulerAspect implements ApplicationContextAware, InitializingBean {
        private ApplicationContext applicationContext;
        @Autowired
        private ElasticJobHandler elasticJobHandler;
        @Override
        public void afterPropertiesSet() throws Exception {
            registrJob(applicationContext);
        }
    
        /**
         * 解析context信息，开始注册
         * @param applicationContext
         */
        private void registrJob(ApplicationContext applicationContext) {
            String[] beanNamesForAnnotation = applicationContext.getBeanNamesForAnnotation(ElasticScheduler.class);
            for (String beanName : beanNamesForAnnotation) {
                Class<?> handlerType = applicationContext.getType(beanName);
                Object bean = applicationContext.getBean(beanName);
                ElasticScheduler annotation = AnnotationUtils.findAnnotation(handlerType, ElasticScheduler.class);
                addJobToContext(annotation,bean);
            }
        }
    
        /**
         * 将任务添加到容器中
         * @param elasticScheduler
         * @param bean
         */
        private void addJobToContext(ElasticScheduler elasticScheduler, Object bean) {
            String cron = elasticScheduler.cron();
            String name = elasticScheduler.name();
            String description = elasticScheduler.description();
            String shardingItemParameters = elasticScheduler.shardingItemParameters();
            Integer shardingTotalCount = elasticScheduler.shardingTotalCount();
            String jobParamters = elasticScheduler.jobParameters();
            try {
                elasticJobHandler.addJob((SimpleJob) bean,cron,shardingTotalCount,shardingItemParameters,jobParamters);
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            } catch (InstantiationException e) {
                e.printStackTrace();
            }
        }
    
        @Override
        public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
            this.applicationContext=applicationContext;
        }
    }

**（6）使用注解定义job：**

package com.dalaoyang.annotation;
    
    import com.dangdang.ddframe.job.api.ShardingContext;
    import com.dangdang.ddframe.job.api.simple.SimpleJob;
    import org.springframework.stereotype.Component;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/7/17 17:18
     */
    @Component
    @ElasticScheduler(cron = "0/5 * * * * ?",shardingTotalCount = 4,name = "测试注解",
            shardingItemParameters = "0=0,1=0,2=1,3=1",jobParameters = "parameter")
    public class StockSimpleJob implements SimpleJob {
        @Override
        public void execute(ShardingContext shardingContext) {
            System.out.println(String.format("------Thread ID: %s, 任務總片數: %s, " +
                            "當前分片項: %s.當前參數: %s," +
                            "當前任務名稱: %s.當前任務參數: %s"
                    ,
                    Thread.currentThread().getId(),
                    shardingContext.getShardingTotalCount(),
                    shardingContext.getShardingItem(),
                    shardingContext.getShardingParameter(),
                    shardingContext.getJobName(),
                    shardingContext.getJobParameter()
    
            ));
        }
    }

**（7）需要再添加pom依赖：**

<dependency>
    	<groupId>com.alibaba</groupId>
    	<artifactId>druid</artifactId>
    	<version>1.1.12</version>
    </dependency>
    <dependency>
    	<groupId>org.projectlombok</groupId>
    	<artifactId>lombok</artifactId>
    	<version>RELEASE</version>
    </dependency>
    <dependency>
    	<groupId>mysql</groupId>
    	<artifactId>mysql-connector-java</artifactId>
    	<version>5.1.34</version>
    </dependency>

**（8）运行结果： **

![20190719163241639.png][]

[Elastic-Job]: https://blog.csdn.net/qq924862077/article/details/82956790
[20190719163241639.png]: /images/20211203/880380984e394ad6a054c9b1f1293b65.png