项目作者: purgeteam

项目描述 :
elastic-job springboot starter,Simplify springboot integration costs.
高级语言: Java
项目地址: git://github.com/purgeteam/elastic-job-spring-boot.git
创建时间: 2019-06-03T10:35:39Z
项目社区:https://github.com/purgeteam/elastic-job-spring-boot

开源协议:Apache License 2.0

下载


elastic-job-spring-boot

Maven Central
License
License
License
License

🔥🔥🔥相关文档请访问 PurgeTeam docs🔥🔥🔥

1 简介

Elastic-Job是一个分布式调度解决方案,由两个相互独立的子项目Elastic-Job-LiteElastic-Job-Cloud组成。Elastic-Job-Lite定位为轻量级无中心化解决方案,使用jar包的形式提供分布式任务的协调服务。
基于quartz定时任务框架为基础的,因此具备quartz的大部分功能
使用zookeeper做协调,调度中心,更加轻量级
支持任务的分片
支持弹性扩容,可以水平扩展, 当任务再次运行时,会检查当前的服务器数量,重新分片,分片结束之后才会继续执行任务
失效转移,容错处理,当一台调度服务器宕机或者跟zookeeper断开连接之后,会立即停止作业,然后再去寻找其他空闲的调度服务器,来运行剩余的任务
提供运维界面,可以管理作业和注册中心。

1.1 使用场景

由于项目为微服务,单模块可能在两个实例以上的数量,定时器就会出现多实例同时执行的情况。
一般定时器缺少管理界面,无法监控定时器是否执行成功。
市面上常见的解决方案为定时器加锁的操作,或者采用第3方分布式定时器。
分布式定时器有多种方案,比如阿里内部的ScheduledX,当当网的Elastic job,个人开源的xxl-job等。

1.2 功能列表

  • 分布式调度协调
  • 弹性扩容缩容
  • 失效转移
  • 错过执行作业重触发
  • 作业分片一致性,保证同一分片在分布式环境中仅一个执行实例
  • 自诊断并修复分布式不稳定造成的问题
  • 支持并行调度
  • 支持作业生命周期操作
  • 丰富的作业类型
  • Spring整合以及命名空间提供
  • 运维平台

1.3 概念

分片:任务的分布式执行,需要将一个任务拆分为多个独立的任务项,然后由分布式的服务器分别执行某一个或几个分片项。
例如:有一个遍历数据库某张表的作业,现有2台服务器。为了快速的执行作业,那么每台服务器应执行作业的50%。 为满足此需求,可将作业分成2片,每台服务器执行1片。作业遍历数据的逻辑应为:服务器A遍历ID以奇数结尾的数据;服务器B遍历ID以偶数结尾的数据。 如果分成10片,则作业遍历数据的逻辑应为:每片分到的分片项应为ID%10,而服务器A被分配到分片项0,1,2,3,4;服务器B被分配到分片项5,6,7,8,9,直接的结果就是服务器A遍历ID0-4结尾的数据;服务器B遍历ID5-9结尾的数据。

历史轨迹:Elastic-Job提供了事件追踪功能,可通过事件订阅的方式处理调度过程的重要事件,用于查询、统计和监控。

1.4 封装elasticjob

由于当当网Elastic job处于1年间未更新阶段,相关jar处于可以使用阶段功能不全。考虑到使用场景为多项目使用,将elastic-job-lite-spring简单封装便于使用。

2.使用说明:

2.1 添加依赖

ps:实际version版本请使用最新版

  1. <dependency>
  2. <groupId>com.purgeteam</groupId>
  3. <artifactId>elasticjob-spring-boot-starter</artifactId>
  4. <version>0.1.1.RELEASE</version>
  5. </dependency>

2.2 配置

ps: 需要mysql,zookeeper支持,请提前搭建好。

配置bootstrap.yml或者application.yml

加入以下配置:

  1. spring:
  2. elasticjob:
  3. datasource: # job需要的记录数据源
  4. url: jdbc:mysql://127.0.0.1:3306/batch_log?useUnicode=true&characterEncoding=utf-8&verifyServerCertificate=false&useSSL=false&requireSSL=false
  5. driver-class-name: com.mysql.cj.jdbc.Driver
  6. username: root
  7. password: Rtqw123OpnmER
  8. regCenter: # 注册中心
  9. serverList: 127.0.0.1:2181
  10. namespace: elasticJobDemo

2.3 定时器实现方法编写

创建定时器类(唯一不同的地方在于将@Scheduled改为实现SimpleJob接口即可)
定时器实现方法编写在execute方法里。

  1. @Slf4j
  2. @Component
  3. public class MySimpleJob implements SimpleJob {
  4. // @Scheduled(cron = "0 0/1 * * * ?")
  5. @Override
  6. public void execute(ShardingContext shardingContext) {
  7. log.info(String.format("Thread ID: %s, 作业分片总数: %s, " +
  8. "当前分片项: %s.当前参数: %s," +
  9. "作业名称: %s.作业自定义参数: %s",
  10. Thread.currentThread().getId(),
  11. shardingContext.getShardingTotalCount(),
  12. shardingContext.getShardingItem(),
  13. shardingContext.getShardingParameter(),
  14. shardingContext.getJobName(),
  15. shardingContext.getJobParameter()
  16. ));
  17. // 分片大致如下:根据配置的分片参数执行相应的逻辑
  18. switch (context.getShardingItem()) {
  19. case 0:
  20. // do something by sharding item 0
  21. break;
  22. case 1:
  23. // do something by sharding item 1
  24. break;
  25. case 2:
  26. // do something by sharding item 2
  27. break;
  28. // case n: ...
  29. }
  30. }
  31. }
  1. logThread ID: 66, 作业分片总数: 1, 当前分片项: 0.当前参数: Beijing,作业名称: PropertiesSimpleJob.作业自定义参数: test

2.4 配置定时器

2.4.1 创建Configuration类

ZookeeperRegistryCenterJobEventConfiguration注入。
创建JobScheduler @Bean(initMethod = "init")
mySimpleJobScheduler方法里先通过ElasticJobUtils#getLiteJobConfiguration获取LiteJobConfiguration对象。
创建SpringJobScheduler对象返回即可。

  1. @Configuration
  2. public class MyJobConfig {
  3. // job 名称
  4. private static final String JOB_NAME = "MySimpleJob";
  5. // 定时器cron参数
  6. private static final String CRON = "0 0/1 * * * ?";
  7. // 定时器分片
  8. private static final int SHARDING_TOTAL_COUNT = 1;
  9. // 分片参数
  10. private static final String SHARDING_ITEM_PARAMETERS = "0=Beijing,1=Shanghai,2=Guangzhou";
  11. // 自定义参数
  12. private static final String JOB_PARAMETERS = "parameter";
  13. @Resource
  14. private ZookeeperRegistryCenter regCenter;
  15. @Resource
  16. private JobEventConfiguration jobEventConfiguration;
  17. @Bean(initMethod = "init")
  18. public JobScheduler mySimpleJobScheduler(final MySimpleJob mySimpleJob) {
  19. LiteJobConfiguration liteJobConfiguration = ElasticJobUtils
  20. .getLiteJobConfiguration(mySimpleJob.getClass(), JOB_NAME, CRON,
  21. SHARDING_TOTAL_COUNT, SHARDING_ITEM_PARAMETERS, JOB_PARAMETERS);
  22. // 参数:1.定时器实例,2.注册中心类,3.LiteJobConfiguration,
  23. // 3.历史轨迹(不需要可以省略)
  24. return new SpringJobScheduler(mySimpleJob, regCenter, liteJobConfiguration, jobEventConfiguration);
  25. }
  26. }

ElasticJobUtils#getLiteJobConfiguration参数简介:

  1. /**
  2. * 获取 {@link LiteJobConfiguration} 对象
  3. *
  4. * @param jobClass 定时器实现类
  5. * @param jobName 定时器名称
  6. * @param cron 定时参数
  7. * @param shardingTotalCount 作业分片总数
  8. * @param shardingItemParameters 当前参数 可以为null
  9. * @param jobParameters 作业自定义参数 可以为null
  10. * @return {@link LiteJobConfiguration}
  11. */
  12. public static LiteJobConfiguration getLiteJobConfiguration(
  13. final Class<? extends SimpleJob> jobClass,
  14. final String jobName,
  15. final String cron,
  16. final int shardingTotalCount,
  17. final String shardingItemParameters,
  18. final String jobParameters) {
  19. ...
  20. return ...;
  21. }

2.4.2 简化Configuration类

当然也可以用下面的@Configuration实现简化,配置bootstrap.yml或者application.yml

  1. spring:
  2. elasticjob:
  3. scheduled:
  4. jobConfigMap: // 为map集合
  5. PropertiesSimpleJob: // 定时器key名称
  6. jobName: PropertiesSimpleJob // job名称
  7. cron: 0 0/1 * * * ? // cron表达式
  8. shardingTotalCount: 2 // 分片数量
  9. shardingItemParameters: 0=123,1=332 // 分片参数
  10. jobParameters: test // 自定义参数

注入SpringJobSchedulerFactory,在propertiesSimpleJobScheduler方法里调用gerSpringJobScheduler方法即可。

  1. @Configuration
  2. public class PropertiesSimpleJobConfig {
  3. @Resource
  4. private SpringJobSchedulerFactory springJobSchedulerFactory;
  5. @Bean(initMethod = "init")
  6. public JobScheduler propertiesSimpleJobScheduler(final PropertiesSimpleJob job) {
  7. // 参数:1.定时器实例,2.配置名称,3.是否开启历史轨迹
  8. return springJobSchedulerFactory.getSpringJobScheduler(job,"PropertiesSimpleJob", true);
  9. }
  10. }

2.4.3 注解方式配置(推荐方式)

ps:这个注解包含了上述方式,简化定时器注入。

继承SimpleJob实现方法execute

AnnotationSimpleJob类上加入注解@ElasticJobScheduler即可。
下面为完整注解。

  1. @Slf4j
  2. @ElasticJobScheduler(
  3. name = "AnnotationSimpleJob", // 定时器名称
  4. cron = "0/8 * * * * ?", // 定时器表达式
  5. shardingTotalCount = 1, // 作业分片总数 默认为1
  6. shardingItemParameters = "0=Beijing,1=Shanghai,2=Guangzhou", // 分片序列号和参数用等号分隔 不需要参数可以不加
  7. jobParameters = "123", // 作业自定义参数 不需要参数可以不加
  8. isEvent = true // 是否开启数据记录 默认为true
  9. )
  10. public class AnnotationSimpleJob implements SimpleJob {
  11. @Override
  12. public void execute(ShardingContext shardingContext) {
  13. log.info(String.format("Thread ID: %s, 作业分片总数: %s, " +
  14. "当前分片项: %s.当前参数: %s," +
  15. "作业名称: %s.作业自定义参数: %s",
  16. Thread.currentThread().getId(),
  17. shardingContext.getShardingTotalCount(),
  18. shardingContext.getShardingItem(),
  19. shardingContext.getShardingParameter(),
  20. shardingContext.getJobName(),
  21. shardingContext.getJobParameter()
  22. ));
  23. }
  24. }