springboot整合分布式任务调度 elastic-job

elastic-job官网地址：https://shardingsphere.apache.org/elasticjob/index_zh.html

程序在运行时，动态添加定时任务，这种场景应用非常广泛。下面我们使用elastic-job实现动态添加定时任务。
注：在使用elastic-job前，需先安装zookeeper。

实现步骤：

• 1.配置初始化的zookeeper地址
• 2.配置的定时任务命名空间(不一定会使用)
• 3.注册初始化数据
• 4.监听器->任务执行前后监听（可有可无）
• 5.动态添加定时任务实现
• 6.自定义任务处理过程-实现SimpleJob

1、配置zookeeper地址和任务命名空间

在yaml文件中配置：

#动态定时任务
zkserver: 192.168.3.10:2181 #zookeeper地址
zknamespace: zknamesp  # 任务命名空间

2、初始化zookeeper注册中心

@Configuration
public class ElasticJobConfig {

    //配置文件中的zookeeper的ip和端口
    @Value(value = "${zkserver}")
    private String serverlists;
    //指定一个命名空间
    @Value("${zknamespace}")
    private String namespace;

    
    @Bean
    public ZookeeperConfiguration zkConfig() {
        return new ZookeeperConfiguration(serverlists, namespace);
    }

    
    @Bean(initMethod = "init")
    public ZookeeperRegistryCenter regCenter(ZookeeperConfiguration config) {
        return new ZookeeperRegistryCenter(config);
    }

    
    @Bean
    public ElasticJobListener elasticJobListener() {
        //初始化要给定超时多少秒重连
        return new ElasticJobListener(100L,100L);
    }
}

3、创建监听器

public class ElasticJobListener extends AbstractDistributeOnceElasticJobListener {

    
    public ElasticJobListener(long startedTimeoutMilliseconds, long completedTimeoutMilliseconds) {
        super(startedTimeoutMilliseconds, completedTimeoutMilliseconds);
    }

    
    @Override
    public void doBeforeJobExecutedAtLastStarted(ShardingContexts shardingContexts) {
        System.out.println("========doBeforeJobExecutedAtLastStarted========"+ TimeUtil.date2FormatHHmmss(new Date()));
    }

    
    @Override
    public void doAfterJobExecutedAtLastCompleted(ShardingContexts shardingContexts) {
        System.out.println("=======doAfterJobExecutedAtLastCompleted============="+ TimeUtil.date2FormatHHmmss(new Date()));
    }
}

4、动态添加定时任务

我们创建一个动态配置任务的类，任何逻辑代码需要创建定时任务，可以直接调用该类的指定方法即可。

@Component
public class ElasticJobHandler {

    @Resource
    private ZookeeperRegistryCenter registryCenter;

    @Resource
    private ElasticJobListener elasticJobListener;

    
    private static LiteJobConfiguration.Builder simpleJobConfigBuilder(String jobName,                                                                Class jobClass,                                                                 int shardingTotalCount,                                                                     String cron,                                                                    String id) {
        //创建任务构建对象
        LiteJobConfiguration.Builder builder = LiteJobConfiguration.newBuilder(new SimpleJobConfiguration(
                JobCoreConfiguration.
                        //任务命名空间名字、任务执行周期表达式、分片个数
                        newBuilder(jobName, cron, shardingTotalCount).
                        //自定义参数
                        jobParameter(id).
                        build(),
                jobClass.getCanonicalName()));
        //本地配置是否可覆盖注册中心配置
        builder.overwrite(true);
        return builder;
    }

    
    public void addPublishJob(String cron,String id,String jobName,SimpleJob instance) {
        LiteJobConfiguration jobConfig = simpleJobConfigBuilder(
                jobName,
                instance.getClass(),
                1,
                cron,
                id).overwrite(true).build();
        //DynamicTask为具体的任务执行逻辑类
        new SpringJobScheduler(instance, registryCenter, jobConfig, elasticJobListener).init();
    }

    
    public static final String CRON_DATE_FORMAT = "ss mm HH dd MM ? yyyy";

    
    public static String getCron(final Date date) {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(CRON_DATE_FORMAT);
        return simpleDateFormat.format(date);
    }
}

5、实现自定义任务，需要实现SimpleJob接口

public class DynamicTask implements SimpleJob {

    @Override
    public void execute(ShardingContext shardingContext) {
        //传递的参数
        String id = shardingContext.getJobParameter();
        try {
            //具体任务逻辑
            System.out.println("执行你的逻辑代码！param:"+id);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

6、测试

@RestController
@RequestMapping(value = "/test")
public class TestController {

    @Autowired
    ElasticJobHandler elasticJobHandler;

    
    @GetMapping
    public Result add(Long times,String jobname,String param){
        //在当前指定时间内延迟times毫秒执行任务
        Date date = new Date(System.currentTimeMillis()+times);
        //需要传递给定时任务的参数
        String cron = ElasticJobHandler.getCron(date);

        //执行任务
        elasticJobHandler.addPublishJob(cron,param,jobname,new DynamicTask());
        return new Result(true, StatusCode.OK,"添加任务成功！");
    }
}

完成！！！