分布式任务elasticjob整合springboot本地多服务启动（附源码）-526互联

Elastic-Job是一个分布式调度的解决方案，由当当网开源，它由两个相互独立的子项目Elastic-Job-Lite和ElasticJob-Cloud组成，使用Elastic-Job可以快速实现分布式任务调度。

Elastic-Job的github地址：https://github.com/elasticjob

功能列表：分布式调度协调在分布式环境中，任务能够按指定的调度策略执行，并且能够避免同一任务多实例重复执行。丰富的调度策略：基于成熟的定时任务作业框架Quartz cron表达式执行定时任务。弹性扩容缩容北京市昌平区建材城西路金燕龙办公楼一层电话：400-618-9090 当集群中增加某一个实例，它应当也能够被选举并执行任务；当集群减少一个实例时，它所执行的任务能被转移到别的实例来执行。失效转移某实例在任务执行失败后，会被转移到其他实例执行。错过执行作业重触发若因某种原因导致作业错过执行，自动记录错过执行的作业，并在上次作业完成后自动触发。支持并行调度支持任务分片，任务分片是指将一个任务分为多个小任务项在多个实例同时执行。作业分片一致性当任务被分片后，保证同一分片在分布式环境中仅一个执行实例。支持作业生命周期操作可以动态对任务进行开启及停止操作。丰富的作业类型支持Simple、DataFlow、Script三种作业类型，后续会有详细介绍。 Spring整合以及命名空间支持对Spring支持良好的整合方式，支持spring自定义命名空间，支持占位符。运维平台提供运维界面，可以管理作业和注册中心。

本次是用elasticjob本地简单测试多服务器启动定时任务，这里需要安装Zookeeper，这里就不多聊了

1、编写Zookeeper配置类

package elastic.job.demo.autoconfig;

import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @ClassName : ZookeeperAutoConfig
 * @Description :自动配置类
 */
@Configuration
@ConditionalOnProperty("elasticjob.zookeeper.server-list")
@EnableConfigurationProperties(ZookeeperProperties.class)
public class ZookeeperAutoConfig {

    private final ZookeeperProperties zookeeperProperties;

    public ZookeeperAutoConfig(ZookeeperProperties zookeeperProperties) {
        this.zookeeperProperties = zookeeperProperties;
    }

    @Bean(initMethod = "init")
    public CoordinatorRegistryCenter zkCenter(){
        String serverList = zookeeperProperties.getServerlist();
        String namespace = zookeeperProperties.getNamespace();

        ZookeeperConfiguration zookeeperConfiguration = new ZookeeperConfiguration(serverList, namespace);
        ZookeeperRegistryCenter zookeeperRegistryCenter = new ZookeeperRegistryCenter(zookeeperConfiguration);
        return zookeeperRegistryCenter;
    }
}

package elastic.job.demo.autoconfig;

import lombok.Getter;
import lombok.Setter;
import org.springframework.boot.context.properties.ConfigurationProperties;

/**
 * @ClassName : ZookeeperProperties
 * @Description :属性配置类
 */
@Getter
@Setter
@ConfigurationProperties(prefix = "elasticjob.zookeeper")
public class ZookeeperProperties {
    //zookeeper地址列表
    private String serverlist;
    //zookeeper命名空间
    private String namespace;

}

2、写完配置类后，我们这里写一个注解，用注解的方式进行实现elasticjob的任务，方便后续的开发和简洁性

package elastic.job.demo.autoconfig;

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

@Target(ElementType.TYPE)//表示使用在哪：这里是类，
@Retention(RetentionPolicy.RUNTIME)//表示运行时进行启动
public @interface ElasticSimpleJob {
    String jobName() default "";
    String cron() default "";
    int shardingTotalCount() default 1;
    boolean overwrite() default  false;
}

3、用springboot的自动装载功能，进行注解的扫描和自动注册

package elastic.job.demo.autoconfig;

import com.dangdang.ddframe.job.api.ElasticJob;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.JobRootConfiguration;
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.base.CoordinatorRegistryCenter;
import elastic.job.demo.job.MySimpleJob;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Map;

/**
 * @ClassName : SimpleJobAutoConfig
 * @Description :
 */
@Configuration
//@Service
@ConditionalOnBean(CoordinatorRegistryCenter.class)
@AutoConfigureAfter(ZookeeperAutoConfig.class)
public class SimpleJobAutoConfig {

    @Autowired
    private CoordinatorRegistryCenter coordinatorRegistryCenter;

    @Autowired
    private ApplicationContext applicationContext;

    //自动注册
    @PostConstruct
    public void initSimpleJob(){
        //获取spring的上下文
        Map<String, Object> beansWithAnnotation = applicationContext.getBeansWithAnnotation(ElasticSimpleJob.class);
        for(Map.Entry<String,Object> entry: beansWithAnnotation.entrySet()){
            Object instance = entry.getValue();
            Class<?>[] interfaces = instance.getClass().getInterfaces();
            for (Class<?> superInterface : interfaces) {
                if(superInterface == SimpleJob.class){
                    ElasticSimpleJob annotation = instance.getClass().getAnnotation(ElasticSimpleJob.class);
                    String jobName = annotation.jobName();
                    String cron = annotation.cron();
                    int shardingTotalCount = annotation.shardingTotalCount();
                    boolean overwrite =annotation.overwrite();
                    //注册定时任务
                    //job 核心配置
                    JobCoreConfiguration buildJcc = JobCoreConfiguration
                            .newBuilder(jobName, cron, shardingTotalCount)
                            .build();
                    //job类型配置
                    SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(
                            buildJcc, instance.getClass().getCanonicalName()
                    );
                    // job配置(LiteJobConfiguration)
                    LiteJobConfiguration buildLiteJobConfiguration = LiteJobConfiguration
                            .newBuilder(simpleJobConfiguration)
                            .overwrite(overwrite)
                            .build();

                    //启动
                    new SpringJobScheduler((ElasticJob) instance,coordinatorRegistryCenter,buildLiteJobConfiguration).init();

                }
            }
        }
    }
}

在springboot添加spring.factories用于扫描并自动注册

4、我们这里编写2个任务 MySimpleJob 和 MySimpleJob2，

MySimpleJob，5秒一次，总分片数为4，MySimpleJob2，5秒一次，总分片数为1

package elastic.job.demo.job;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import elastic.job.demo.autoconfig.ElasticSimpleJob;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * @ClassName : MySimpleJob
 * @Description : 我的定时任务
 */
@ElasticSimpleJob(jobName = "mySimpleJob",
        cron = "0/5 * * * * ?",
        shardingTotalCount = 4,
        overwrite = true)
@Component
public class MySimpleJob implements SimpleJob {

    @Value("${server.port}")
    private int port;

    @Override
    public void execute(ShardingContext shardingContext) {
        System.out.println("端口：" + port  + ",我是任务1分片项："+shardingContext.getShardingItem()+",总分片数是："+
                shardingContext.getShardingTotalCount());
    }
}

package elastic.job.demo.job;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import elastic.job.demo.autoconfig.ElasticSimpleJob;
import org.springframework.stereotype.Component;

/**
 * @author zhoucc
 * @date 2023-06-26 16:38
 */
@ElasticSimpleJob(jobName = "mySimpleJob2",
        cron = "0/5 * * * * ?",
        shardingTotalCount = 1,
        overwrite = true)
@Component
public class MySimpleJob2 implements SimpleJob {

    @Override
    public void execute(ShardingContext shardingContext) {
        System.out.println("我是任务2分片项："+shardingContext.getShardingItem()+",总分片数是："+
                shardingContext.getShardingTotalCount());
    }
}