elastic-job动态任务配置

蔚落 2022-09-11 04:16 353阅读 0赞

## 前言 ##

在之前的springboot整合elastic-job篇中，我们了解到，elastic-job是一个不错的任务调度框架，本篇将进一步说明，如何使用elastic-job实现常用的动态任务的执行场景

## 环境准备 ##

zookeeper安装与启动(本地可以使用windows版，比较快捷)

## 简易工程搭建 ##

**1、添加基础pom依赖**

<parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.2.1.RELEASE</version>
            <relativePath/> 
        </parent>
    
        <properties>
            <mybatis-plus-boot-starter.version>3.3.0</mybatis-plus-boot-starter.version>
            <mybatis-plus-generator.version>3.3.0</mybatis-plus-generator.version>
            <swagger.version>2.9.2</swagger.version>
            <swagger-bootstrap-ui.version>1.9.6</swagger-bootstrap-ui.version>
        </properties>
    
        <dependencies>
    
            
            <dependency>
                <groupId>io.springfox</groupId>
                <artifactId>springfox-swagger2</artifactId>
                <version>${swagger.version}</version>
            </dependency>
            <dependency>
                <groupId>io.springfox</groupId>
                <artifactId>springfox-swagger-ui</artifactId>
                <version>${swagger.version}</version>
            </dependency>
            <dependency>
                <groupId>com.github.xiaoymin</groupId>
                <artifactId>swagger-bootstrap-ui</artifactId>
                <version>${swagger-bootstrap-ui.version}</version>
            </dependency>
    
            
            
    
            <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>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
                <version>2.2.1.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
    
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>8.0.11</version>
            </dependency>
    
            
            <dependency>
                <groupId>com.baomidou</groupId>
                <artifactId>mybatis-plus-boot-starter</artifactId>
                <version>${mybatis-plus-boot-starter.version}</version>
            </dependency>
            <dependency>
                <groupId>com.baomidou</groupId>
                <artifactId>mybatis-plus-generator</artifactId>
                <version>${mybatis-plus-generator.version}</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.18.0</version>
            </dependency>
    
        </dependencies>

**2、配置文件配置zk信息**

elastic-job的任务调度，在运行的时候依赖zk，因此需要根据自身情况配置zk，可以理解为任务需要结合zk的节点进行使用，如果不需要数据库连接可以忽略

server.port=8081
    
    # zookeeper集群
    elaticjob.zookeeper.server-lists=127.0.0.1:2181
    elaticjob.zookeeper.namespace=updatetask
    
    #动态配置
    zkserver= 127.0.0.1:2181
    zknamespace=zknamespce
    
    #动态配置2
    elasticjob.serverlists=127.0.0.1:2181
    elasticjob.namespace=boot-job
    
    spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
    spring.datasource.url=jdbc:mysql://IP:3306/dbname?autoReconnect=true&useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&useSSL=false
    spring.datasource.username=root
    spring.datasource.password=root
    
    mybatis-plus.mapper-locations=classpath*:mapper/*.xml
    mybatis-plus.global-config.db-column-underline=true
    mybatis-plus.global-config.db-config.id-type=uuid
    mybatis-plus.global-config.db-config.field-strategy=not_null
    mybatis-plus.global-config.refresh=true
    mybatis-plus.configuration.map-underscore-to-camel-case=true
    mybatis-plus.configuration.log-impl=org.apache.ibatis.logging.stdout.StdOutImpl

**3、配置ElasticJob相关**

包括zk注册中心，elastic-job监听器等

import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
    import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
    import lombok.Data;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    @Data
    public class ElasticJobConfig {
    
        @Value("${elasticjob.serverlists}")
        private String serverLists;
    
        @Value("${elasticjob.namespace}")
        private String namespace;
    
        @Bean
        public ZookeeperConfiguration zConfig(){
            return new ZookeeperConfiguration(serverLists, namespace);
        }
    
        @Bean(initMethod = "init")
        public ZookeeperRegistryCenter registryCenter(ZookeeperConfiguration config){
            return new ZookeeperRegistryCenter(config);
        }
    
        @Bean
        public ElasticJobListener elasticJobListener(){
            return new ElasticJobListener(1000L, 10000L);
        }
    }

**4、动态任务处理配置类**

主要包括动态任务的增删改

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.listener.ElasticJobListener;
    import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
    import com.dangdang.ddframe.job.lite.internal.schedule.JobRegistry;
    import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
    import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    
    @Component
    public class ElasticJobHandler {
    
        @Resource
        private ZookeeperRegistryCenter registryCenter;
    
        @Resource
        private ElasticJobListener elasticJobListener;
    
        private static LiteJobConfiguration.Builder simpleJobConfigBuilder(String jobName,
                                                                           Class<? extends SimpleJob> jobClass,
                                                                           int shardTotalCount,
                                                                           String cron,
                                                                           String id) {
            return LiteJobConfiguration.newBuilder(new SimpleJobConfiguration(
                    JobCoreConfiguration.newBuilder(jobName,cron,shardTotalCount).jobParameter(id).build(),jobClass.getCanonicalName()
            ));
        }
    
        /**
         * 添加定时任务
         */
        public void addJob(String jobName, SimpleJob jobInstance, String cron, Integer shardTotalCount, String id) {
            LiteJobConfiguration jobConfig = simpleJobConfigBuilder(jobName, jobInstance.getClass(), shardTotalCount, cron, id)
                    .overwrite(true).build();
            new SpringJobScheduler(jobInstance,registryCenter,jobConfig,elasticJobListener).init();
        }
    
        /**
         * 修改定时任务
         * @param jobName
         * @param cron
         */
        public void updateJob(String jobName, String cron) {
            JobRegistry.getInstance().getJobScheduleController(jobName).rescheduleJob(cron);
        }
    
        /**
         * 移除定时任务
         * @param jobName
         */
        public void removeJob(String jobName){
            JobRegistry.getInstance().getJobScheduleController(jobName).shutdown();
        }
    }

**5、接口实现任务的增删改**

import com.congge.task.config.ElasticJobHandler;
    import com.congge.task.config.MyElasticJob;
    import org.apache.commons.lang3.StringUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    
    @Controller
    @RequestMapping("/task")
    public class ElasticController {
    
        @Autowired
        private ElasticJobHandler jobHandler;
    
        @GetMapping("/add")
        public Object addJob(String jobName,String cronExpress){
            if(StringUtils.isEmpty(jobName)){
                jobName = "test-job";
            }
            if(StringUtils.isEmpty(cronExpress)){
                cronExpress = "1/5 * * * * ?";
            }
            jobHandler.addJob(jobName, new MyElasticJob(), cronExpress, 1, "1");
            return "add success";
        }
    
        @GetMapping("/update")
        public Object updateJob(String jobName,String cronExpress){
            if(StringUtils.isEmpty(jobName)){
                jobName = "test-job";
            }
            if(StringUtils.isEmpty(cronExpress)){
                cronExpress = "1/5 * * * * ?";
            }
            jobHandler.updateJob(jobName,cronExpress);
            return "update success";
        }
    
        @GetMapping("/remove")
        public Object removeJob(String jobName){
            jobHandler.removeJob(jobName);
            return "remove success";
        }
    
    }

下面来测试一下接口是否好用，我们先来添加一个任务  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP56CB5Yac5Y-U5Y-U_size_20_color_FFFFFF_t_70_g_se_x_16]  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP56CB5Yac5Y-U5Y-U_size_15_color_FFFFFF_t_70_g_se_x_16]  
可以看到，后台已经开始执行任务，由于我们没有给执行cron，默认值每隔5秒执行一次

当然，我们也可以连接elastic-job的控制台，看到我们后台正在运行的job，可以通过管控台对job进行处理，比如我们停止这个job  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP56CB5Yac5Y-U5Y-U_size_20_color_FFFFFF_t_70_g_se_x_16 1]  
当点击关闭之后，任务将不再执行  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP56CB5Yac5Y-U5Y-U_size_20_color_FFFFFF_t_70_g_se_x_16 2]  
![在这里插入图片描述][78d85bb2add84622aed4788a32fdd5a6.png]

## 与实际的业务整合 ##

经常有这样的场景，我们需要对数据库的某个表的数据进行状态值修改，或者定时备份某个表的数据，或者执行一些特殊任务等，都可以使用elastic-job来完成

具体来说，只需要在实现了SimpleJob的类中的execute方法里面去做就可以了，例如有这样一个场景，每隔5秒需要对user表中状态为1的数据进行修改  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP56CB5Yac5Y-U5Y-U_size_20_color_FFFFFF_t_70_g_se_x_16 3]

那么就可以直接改造MyElasticJob这个类

public class MyElasticJob implements SimpleJob {
    
        UserService userService = SpringContextUtil.getBean("userService");
    
        @Override
        public void execute(ShardingContext shardingContext) {
            String jobName = shardingContext.getJobName();
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("HH:mm:ss");
            System.out.println("任务:" + jobName +",执行" + "时间:" + simpleDateFormat.format(new Date()));
            //执行具体的任务
            userService.changeStatus();
        }
    }

如果有更多的业务要处理，可以类似的操作，下面启动工程，添加一个任务来执行下，看看是否能打到预期的目的，

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP56CB5Yac5Y-U5Y-U_size_20_color_FFFFFF_t_70_g_se_x_16 4]  
![在这里插入图片描述][bdc47d4a6c0b476abc7e8d77e3695e05.png]  
可以看到，执行之后，status的值变成了-1

**以上是本文的全部内容，实际应用中，还需要结合较多的因素进行改造，希望对您有帮助哦**

有需要源码的同学可前往下载：https://download.csdn.net/download/zhangcongyi420/22501493

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP56CB5Yac5Y-U5Y-U_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/839535827dff424cb4f8c26a8740b690.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP56CB5Yac5Y-U5Y-U_size_15_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/56193ae7822b49dea022d7c775678930.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP56CB5Yac5Y-U5Y-U_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220828/d12a0004f9d34f08a1a3b01d40da5c81.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP56CB5Yac5Y-U5Y-U_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220828/38dccfa809a64fa5ab2d5e74ef4f51d6.png
[78d85bb2add84622aed4788a32fdd5a6.png]: /images/20220828/27e42bece39f43b7905e94c7a9c08355.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP56CB5Yac5Y-U5Y-U_size_20_color_FFFFFF_t_70_g_se_x_16 3]: /images/20220828/902435cba838497db7cbd91a665c6988.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bCP56CB5Yac5Y-U5Y-U_size_20_color_FFFFFF_t_70_g_se_x_16 4]: /images/20220828/98cd22de92ac4fef93e428d70c720890.png
[bdc47d4a6c0b476abc7e8d77e3695e05.png]: /images/20220828/dac76f9268c5402991c82db27d0c1aff.png