分布式任务调度 ElasticJob demo

代码地址：GitHub - augtsy/springboot-elasticJob-demo: ElasticJob 分布式任务调度ElasticJob 分布式任务调度. Contribute to augtsy/springboot-elasticJob-demo development by creating an account on GitHub.https://github.com/augtsy/springboot-elasticJob-demo 一、常见开源产品

Quartz、XXL-Job、ElasticJob等

Quartz：该框架应用最为广泛，其完全基于 Java 实现，Quartz 对单个任务的控制基本做到了极致，以其强大功能和应用灵活性，成为开源任务调度领域的权威及同类开源产品如 Antares 的基石；
XXL-JOB：一个轻量级分布式任务调度平台，其核心设计目标是开发迅速、学习简单、轻量级、易扩展。XXL-JOB 支持分片，支持简单任务依赖，支持子任务依赖，不支持跨平台的。
Elastic-Job：支持任务分片（作业分片一致性），没有任务编排，不支持跨平台；

二、业务场景

公司任务中心，提供客户导入导出等定时任务处理。

三、代码实现

elastic-job依赖zookeeper，首先要安装好zookeeper。可以docker 启动镜像等。

properties配置

info.app.name=elastic-job-demo
zookeeper.servers=172.19.8.76:2181

引入依赖


    com.dangdang
    elastic-job-lite-core
    2.1.5



    com.dangdang
    elastic-job-lite-spring
    2.1.5

注册中心配置类

@Configuration
public class ElasticRegCenterConfig {

    
    @Bean(initMethod = "init")
    public ZookeeperRegistryCenter zookeeperRegistryCenter(
            @Value("${zookeeper.servers}") final String serverList,
            @Value("${info.app.name}") final String namespace) {
        return new ZookeeperRegistryCenter(new ZookeeperConfiguration(serverList, namespace));
    }

}

注解类

@Inherited
@Documented
@Retention(value = RetentionPolicy.RUNTIME)
@Target(value = ElementType.TYPE)
public @interface ElasticJobConfig {


    String cron() default "";

    String value() default "";

    
    int shardingTotalCount() default 1;

    
    String shardingItemParameters() default "0=0";
}

SimpleJob配置类

@Slf4j
@Configuration
public class SimpleJobConfig {

    @Autowired
    private ApplicationContext applicationContext;
    @Autowired
    private ZookeeperRegistryCenter zookeeperRegistryCenter;

    @PostConstruct
    public void startSimpleJob() {
        applicationContext.getBeansWithAnnotation(ElasticJobConfig.class).forEach((className, obj) -> {
            ElasticJobConfig config = obj.getClass().getAnnotation(ElasticJobConfig.class);
            // 表达式
            String cron = StringUtils.defaultIfBlank(config.cron(), config.value());
            // 分片总数
            int shardingTotalCount = config.shardingTotalCount();
            // 分片项参数
            String shardingItemParameters = config.shardingItemParameters();
            JobListener elasticJobListener = new JobListener();
            SimpleJob simpleJob = (SimpleJob) obj;
            new SpringJobScheduler(simpleJob, zookeeperRegistryCenter,
                    getLiteJobConfiguration(simpleJob.getClass(), cron, shardingTotalCount, shardingItemParameters),
                    elasticJobListener).init();
        });
    }

    
    private LiteJobConfiguration getLiteJobConfiguration(Class jobName, String cron, int shardingTotalCount, String shardingItemParameters) {
        return LiteJobConfiguration.newBuilder(
                new SimpleJobConfiguration(
                        JobCoreConfiguration.newBuilder(jobName.getName(), cron, shardingTotalCount)
                                .shardingItemParameters(shardingItemParameters)
                                .build(),
                        jobName.getCanonicalName()))
                .monitorExecution(true)
                .overwrite(false).build();
    }
}

定时任务实现

@Slf4j
@Component
@ElasticJobConfig(cron = "*/4 * * * * ?", shardingTotalCount = 5, shardingItemParameters = "0=0,1=1,2=2,3=3,4=4")
public class SimpleJobDemo implements SimpleJob {

    @Override
    public void execute(ShardingContext shardingContext) {
        //获取分片总数
        int shardingTotalCount = shardingContext.getShardingTotalCount();
        //获取分片项
        int shardingItem = shardingContext.getShardingItem();
        //获取分片项参数
        String shardingParameter = shardingContext.getShardingParameter();

        log.info("分片总数:{} 分片项:{} 分片项参数:{}", shardingTotalCount, shardingItem, shardingParameter);
        // 业务逻辑，ElasticJob 不提供数据处理的功能，框架只会将分片项分配至各个运行中的作业服务器，开发者需要自行处理
    }
}

四、测试

只启动一个实例，结果所有分片都被分配到这一个实例。

启动其他四台实例，结果5个分片被均匀分片5个节点上。

五、注册中心节点查看

后记：才疏学浅，不吝赐教。

分布式任务调度 ElasticJob demo

Java相关栏目本月热门文章