文章目录
- 一、什么是分布式调度
- 二、Elastic-Job 介绍
- 三、Elastic-Job 实战
- 3.1 环境搭建
- 3.1.1 本地部署
- 3.1.2 服务器部署
- 3.1.3 Zookeeper 管控台界面
- 3.2 入门案例
- 3.3 SpringBoot 集成 Elastic-Job
- 3.4 任务分片（★）
- 3.5 Dataflow 类型调度任务
一、什么是分布式调度
什么是任务调度？
我们可以思考一下下面业务场景的解决方案：
① 某电商平台需要每天上午10点，下午3点，晚上8点发放一批优惠券
② 某银行系统需要在信用卡到期还款日的前三天进行短信提醒
③ 某财务系统需要在每天凌晨0:10分结算前一天的财务数据，统计汇总
以上场景就是任务调度所需要解决的问题。
任务调度是为了自动完成特定任务，在约定的特定时刻去执行任务的过程
以上的场景可以使用定时任务注解 @Scheduled 贴在业务方法上，并在启动类上贴上 @EnableScheduling 注解，以实现任务调度。
@Scheduled(cron = "0/20 * * * * ? ")
public void doWork(){
//doSomething
}
什么是分布式调度？
感觉 Spring 给我们提供的这个注解可以完成任务调度的功能，为什么还需要分布式呢？主要有如下这几点原因：
- 单机处理极限：原本 1 分钟内需要处理 1 万个订单，但是现在需要 1 分钟内处理 10 万个订单；原来一个统计需要1小时，现在业务方需要10分钟就统计出来。你也许会说，你也可以多线程、单机多进程处理。的确，多线程并行处理可以提高单位时间的处理效率，但是单机能力毕竟有限(主要是CPU、内存和磁盘)，始终会有单机处理不过来的情况。
- 高可用：单机版的定式任务调度只能在一台机器上运行，如果程序或者系统出现异常就会导致功能不可用。虽然可以在单机程序实现的足够稳定，但始终有机会遇到非程序引起的故障，而这个对于一个系统的核心功能来说是不可接受的。
- 防止重复执行：在单机模式下，定时任务是没什么问题的。但当我们部署了多台服务，同时又每台服务又有定时任务时，若不进行合理的控制在同一时间，只有一个定时任务启动执行，这时，定时执行可能存在执复、混乱和错误的情况了。
这个时候就需要分布式的任务调度来实现了。
二、Elastic-Job 介绍
Elastic-Job 是一个分布式调度的解决方案，它由两个相互独立的子项目 Elastic-job-Lite 和 Elastic-Job-Cloud 组成，使用 Elastic-Job 可以快速实现分布式任务调度。
官方地址：https://shardingsphere.apache.org/elasticjob/
功能列表：
- 分布式调度协调：在分布式环境中，任务能够按照指定的调度策略执行，并且能够避免同一任务多实例重复执行。
- 丰富的调度策略：基于成熟的定时任务作业框架 Quartz cron 表达式执行定时任务。
- 弹性拓容缩容：当集群中增加一个实例，它应当能够被选举被执行任务；当集群减少一个实例时，他所执行的任务能被转移到别的示例中执行。
- 失效转移：某示例在任务执行失败后，会被转移到其他实例执行。
- 错过执行任务重触发：若因某种原因导致作业错过执行，自动记录错误执行的作业，并在下次作业完成后自动触发。
- 支持并行调度：支持任务分片，任务分片是指将一个任务分成多个小任务在多个实例同时执行。
- 作业分片一致性：当任务被分片后，保证同一分片在分布式环境中仅一个执行实例。
- 支持作业生命周期操作：可以动态对任务进行开启及停止操作，丰富的作业类型。
执行架构如下：
三、Elastic-Job 实战
3.1 环境搭建
zookeeper 可以理解为 elastic-job 的注册中心，分布式调度等功能由它实现，首先要下载资源。
csdn搜索资源： zookeeper-3.4.11.tar.gz
3.1.1 本地部署
将 zookeeper-3.4.11.tar.gz 解压，并将 conf 目录下 zoo_sample.cfg 拷贝一份命名成 zoo.cfg
其中 zookeeper 默认端口是 2181
切换到 bin 目录下，双击 zkServer.cmd，即可启动 zookeeper
3.1.2 服务器部署
step1：将 zookeeper-3.4.11.tar.gz上传到 /usr/local/software目录下
step2：解压文件到指定目录
tar -zxvf /usr/local/software/zookeeper-3.4.11.tar.gz -C /usr/local/
step3：拷贝配置文件
cp /usr/local/software/zookeeper-3.4.11/conf/zoo_sample.cfg /usr/local/software/zookeeper-3.4.11/conf/zoo.cfg
step4：启动
/usr/local/zookeeper-3.4.11/bin/zkServer.sh start
step5：检查进程是否开启，需要查看到QuorumPeerMain进程，如果存在则证明启动成功。
jps
zookeeper常用名称参考： linux下的zookeeper启动、停止 常用命令
注：如果启动显示 Starting zookeeper ... already running as process 7827. 但是 jps 中没有 QuorumPeerMain 进程。则需查看 zookeeper_server.pid 文件的位置并删除。
# 查看该文件位置
find / -name "zookeeper_server.pid"
# 跳转到该文件的位置，并删除
rm -rf zookeeper_server.pid
另外：服务器需要暂时关闭防火墙 systemctl stop firewalld，并可使用 firewall-cmd --state 查看防火墙状态。
具体可参考：Linux关闭防火墙命令
3.1.3 Zookeeper 管控台界面
搜索下载：zooInspector.zip
解压后进入 build 目录，运行 jar 包：java -jar zookeeper-dev-ZooInspector.jar
点击绿色按钮，输入连接的IP和端口号即可。
3.2 入门案例
版本要求：JDK 要求1.7 以上版本；Maven 要求 3.0.4 及以上版本；Zookeeper 要求 3.4.6 以上版本
1、引入 pom 依赖
<dependency>
<groupId>com.dangdang</groupId>
<artifactId>elastic-job-lite-core</artifactId>
<version>2.1.5</version>
</dependency>
2、调度任务类
public class MyElasticJob implements SimpleJob {
@Override
public void execute(ShardingContext shardingContext) {
System.out.println("执行任务：" + new Date());
}
}
3、zookeeper 的配置类
启动类定义 JobScheduler 对象，里面传入两个对象：定时任务配置对象、注册中心配置，并调用 init() 方法完成初始化。
定时任务配置对象中要设置：任务名称、cron表达式和分片数量，并设置 任务对象的全路径类名。
注册中心配置对象中要设置：注册中心的地址、项目名以及节点的超时时间。
public class JobDemo {
public static void main(String[] args) {
new JobScheduler(createRegistryCenter(), createJobConfiguration()).init();
}
/**
* Zookeeper注册中心配置
*
* @return 注册中心配置对象
*/
private static CoordinatorRegistryCenter createRegistryCenter() {
// ZookeeperConfiguration("zookeeper地址", "项目名")
ZookeeperConfiguration configuration = new ZookeeperConfiguration("localhost:2181", "elastic-job");
// 设置节点超时时间，即每隔一段时间查看当前节点是否下线
configuration.setConnectionTimeoutMilliseconds(100);
ZookeeperRegistryCenter center = new ZookeeperRegistryCenter(configuration);
center.init();
return center;
}
/**
* 定时任务配置
*
* @return 定时任务配置对象
*/
private static LiteJobConfiguration createJobConfiguration() {
// 定义作业核心配置 newBuilder("任务名称", "corn表达式", "分片数量")
JobCoreConfiguration simpleCoreConfig = JobCoreConfiguration.newBuilder("myElasticJob", "0/10 * * * * ?", 1).build();
// 定义simple类型配置，MyElasticJob.class.getCanonicalName() 是获取MyElasticJob的全限定类名（全路径类名）
SimpleJobConfiguration configuration = new SimpleJobConfiguration(simpleCoreConfig, MyElasticJob.class.getCanonicalName());
// 定义Lite作业根配置，并返回
// 设置overwrite(true)，允许覆盖cron表达式（默认不允许，会每5s执行一次）
return LiteJobConfiguration.newBuilder(configuration).overwrite(true).build();
}
}
注： .overwrite(true) 如果不设置，默认 5 秒执行一次。
4、运行main方法。当开启第二个实例的时候，第一个实例停止打印，当关闭第二个实例的时候，第一个实例又重新开始运行。
3.3 SpringBoot 集成 Elastic-Job
1、添加 pom 依赖
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.axy</groupId>
<artifactId>elastic-job-boot</artifactId>
<packaging>jar</packaging>
<version>1.0-SNAPSHOT</version>
<name>elastic-job-boot</name>
<url>http://maven.apache.org</url>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.3.RELEASE</version>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>com.dangdang</groupId>
<artifactId>elastic-job-lite-spring</artifactId>
<version>2.1.5</version>
</dependency>
...
</dependencies>
<build>
<finalName>elastic-job-boot</finalName>
</build>
</project>
2、因为配置中心的地址并不是固定的，所以我们应该把这个地址信息配置在配置文件中，所以在配置文件
application.yml 中添加配置如下：
zookeeper:
url: localhost:2181
groupName: elastic-job-boot
3、调度任务类交给 Spring 进行管理
@Component
public class MyElasticJob implements SimpleJob {
@Override
public void execute(ShardingContext shardingContext) {
System.out.println("定时调度：" + new Date());
}
}
4、zookeeper 的配置类
elastic-job 集成 SpringBoot 以后，需要创建的不是 JobScheduler 对象，而是 SpringJobScheduler 对象，并交给 Spring 管理。其次，初始化该对象还需要传入需要分布式调度的任务对象当参数。
@Configuration
public class TestJobConfig {
@Bean(initMethod = "init") // 创建bean后调用init方法
public SpringJobScheduler testScheduler(MyElasticJob job, CoordinatorRegistryCenter registryCenter) {
// 方法形参，自动会去spring的容器中寻找，首先会去看类型匹配，然后才会去看变量名，匹配方式和@Autowird一样
LiteJobConfiguration configuration = createJobConfiguration(job.getClass(), "0/4 * * * * ?", 1);
return new SpringJobScheduler(job, registryCenter, configuration);
}
@Bean
public CoordinatorRegistryCenter registryCenter(@Value("${zookeeper.url}") String url,
@Value("${zookeeper.groupName}") String groupName) {
// ZookeeperConfiguration("zookeeper地址", "项目名")
ZookeeperConfiguration configuration = new ZookeeperConfiguration(url, groupName);
// 设置节点超时时间
configuration.setConnectionTimeoutMilliseconds(100);
ZookeeperRegistryCenter center = new ZookeeperRegistryCenter(configuration);
center.init();
return center;
}
/**
* 定时任务配置
* 这个定时任务使用的场景比较灵活，因此不建议放在spring的容器当中
*
* @param clazz 定时任务的字节码
* @param cron cron表达式
* @param shardingCount 分片数量
* @return 定时任务配置对象
*/
private static LiteJobConfiguration createJobConfiguration(Class clazz, String cron, int shardingCount) {
// 定义作业核心配置 newBuilder("任务名称", "corn表达式", "分片数量")
JobCoreConfiguration.Builder jobBuilder = JobCoreConfiguration.newBuilder(
clazz.getSimpleName(), cron, shardingCount);
JobCoreConfiguration simpleCoreConfig = jobBuilder.build();
// 定义simple类型配置，MyElasticJob.class.getCanonicalName() 是获取MyElasticJob的全限定类名（全路径类名）
System.out.println("MyElasticJob.class.getCanonicalName(): " + MyElasticJob.class.getCanonicalName());
JobTypeConfiguration configuration;
configuration = new SimpleJobConfiguration(simpleCoreConfig, clazz.getCanonicalName());
// 定义Lite作业根配置，并返回
// 设置overwrite(true)，允许覆盖cron表达式（默认不允许）
return LiteJobConfiguration.newBuilder(configuration).overwrite(true).build();
}
}
5、启动项目
3.4 任务分片（★）
分片就是在就是在机器中分线程执行，分片的数量决定了最终分得线程的数量，将一个任务拆分为多个独立的任务项，然后由分布式的应用实例分别执行某一个或者几个分布项，下面以案例来逐步带入分片的概念。
现要处理下图表中数据的 backedUp 属性设置为 1。表一共有 20 条数据，根据 type 进行分类可以分成 text、image、radio 和 video 四类，那我们可以自定义任务分 4 片，分片索引分别为：0、1、2、3。
- 当只有一台机器的情况下，在机器 A 启动四个线程，分别处理四个分片的内容。
- 当有两台机器的情况下，机器 A 启动两个线程负责索引 0 1 的分片内容，机器 B 负责 2 3 的分片内容。
- 当有三台机器的情况下，机器 A 负责索引 0 1 的分片内容，机器 B 负责 2，机器 C 负责 3。
- 当有四台机器的情况下，机器 A 负责索引 0 的分片内容，机器 B 负责 1，机器 C 负责 2，机器 D 负责 3。
注：分片数建议等于机器个数的倍数。如：分片四个在两台机器上，那么就是每台机器分两个线程来执行任务。
如何实现上文的案例呢？这里我们忽略关于数据库层面的配置，主要的类与配置如下：
1、这里我们新建文件对象
@AllArgsConstructor
@NoArgsConstructor
@Data
public class FileCustom {
// 唯⼀标识
private Long id;
//⽂件名
private String name;
//⽂件类型
private String type;
//⽂件内容
private String content;
// 是否已备份
private Boolean backedUp = false;
}
2、这里我们先定义 zookeeper 的配置类
在初始化定时任务配置的时候，以字符串的形式传入分片参数，传入"0=text,1=image,2=radio,3=vedio"，设置分片个数为 4，并添加分片功能 shardingItemParameters(...)。
- 如果 分片个数 小于 分片参数，则取参数中前几个。如：分片取 2，则只会对
0=text,1=image进行处理 - 如果 分片个数 大于 分片参数，则多出的参数补 null。如：分片取 5，则参数字符串会变为
0=text,1=image,2=radio,3=vedio,4=null
@Configuration
public class TestJobConfig {
@Bean(initMethod = "init")
public SpringJobScheduler fileScheduler(FileCustomElasticJob job, CoordinatorRegistryCenter registryCenter) {
LiteJobConfiguration configuration = createJobConfiguration(job.getClass(), "0 0/1 * * * ?", 4, "0=text,1=image,2=radio,3=vedio");
return new SpringJobScheduler(job, registryCenter, configuration);
}
@Bean
public CoordinatorRegistryCenter registryCenter(@Value("${zookeeper.url}") String url, @Value("${zookeeper.groupName}") String groupName) {
// ZookeeperConfiguration("zookeeper地址", "项目名")
ZookeeperConfiguration configuration = new ZookeeperConfiguration(url, groupName);
// 设置节点超时时间
configuration.setConnectionTimeoutMilliseconds(100);
ZookeeperRegistryCenter center = new ZookeeperRegistryCenter(configuration);
center.init();
return center;
}
/**
* 定时任务配置
* 这个定时任务使用的场景比较灵活，因此不建议放在spring的容器当中
*
* @param clazz 定时任务的字节码
* @param cron cron表达式
* @param shardingCount 分片数量
* @param shardingParam 分片参数
* @return 定时任务配置对象
*/
private static LiteJobConfiguration createJobConfiguration(Class clazz, String cron, int shardingCount,
String shardingParam) {
// 定义作业核心配置 newBuilder("任务名称", "corn表达式", "分片数量")
JobCoreConfiguration.Builder jobBuilder = JobCoreConfiguration.newBuilder(
clazz.getSimpleName(), cron, shardingCount);
if (!StringUtils.isEmpty(shardingParam)) {
jobBuilder.shardingItemParameters(shardingParam); // 添加分片功能
}
JobCoreConfiguration simpleCoreConfig = jobBuilder.build();
// 定义simple类型配置
JobTypeConfiguration configuration;
configuration = new SimpleJobConfiguration(simpleCoreConfig, clazz.getCanonicalName());
// 定义Lite作业根配置，并返回
// 设置overwrite(true)，允许覆盖cron表达式（默认不允许）
return LiteJobConfiguration.newBuilder(configuration).overwrite(true).build();
}
}
3、文件对象的调度任务类
@Slf4j
@Component
public class FileCustomElasticJob implements SimpleJob {
@Autowired
private FileCustomMapper fileCustomMapper;
@Override
public void execute(ShardingContext shardingContext) {
long threadId = Thread.currentThread().getId();
System.out.printf("线程ID:{}，任务的名称:{}，任务参数:{}，分片个数L:{}，分片索引号:{}，分片参数:{}\n",
threadId,
shardingContext.getJobName(),
shardingContext.getJobParameter(),
shardingContext.getShardingTotalCount(),
shardingContext.getShardingItem(),
shardingContext.getShardingParameter());
doWorkByParameter(shardingContext.getShardingParameter());
}
/**
* 根据类型查询出所有的备份任务
*
* @param shardingParameter 线程对应处理的文件类型
*/
private void doWorkByParameter(String shardingParameter) {
List<FileCustom> fileCustoms = fileCustomMapper.selectByType(shardingParameter);
for (FileCustom fileCustom : fileCustoms) {
backUp(fileCustom);
}
}
/**
* 模拟备份操作
*
* @param fileCustom 备份对象
*/
private void backUp(FileCustom fileCustom) {
System.out.println("备份的方法名：" + fileCustom.getName() + "，备份的类型：" + fileCustom.getType());
System.out.println("==================================");
try {
TimeUnit.SECONDS.sleep(1); // 延时一秒
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
fileCustomMapper.changeState(fileCustom.getId(), 1); // 修改数据的 backedUp
}
}
4、这里我们模拟有两台机器，即两个实例。
从运行结果我们可以看出，机器 A 开启了两个线程来处理分片索引为 0 1 的分片内容，机器 B 开启了两个线程来处理分片索引为 2 3 的分片内容。
因此，通过对任务的合理分片化，可以达到任务并行处理的效果。分片的优点如下：
- 分片项与业务处理解耦：Elastic-ob并不直接提供数据处理的功能,框架只会将分片项分配至各个运行中的作业服务器，开发者需要自行处理分片项与真实数据的对应关系。
- 最大限度利用资源：将分片项设置大于服务器的数据，最好是大于服务器倍数的数量，作业将会合理利用分布式资源，动态的分配分片项。
3.5 Dataflow 类型调度任务
Dataflow 类型适用于要处理的数据量很大的情况，Dataflow 类型的定时任务需要实现 Datafowjob 接口，该接口提供 2 个方法供覆盖，分别用于抓取 fetchData 和处理 processData 数据，我们继续对例子进行改造。
Dataflow 类型用于处理数据流，他和 Simplejob 不同，它以数据流的方式执行，调用 fetchData 抓取数据，知道抓取不到数据才停止作业。
1、修改 zookeeper 配置类，增加数据类型判断和逻辑
@Configuration
public class TestJobConfig {
@Bean(initMethod = "init")
public SpringJobScheduler fileScheduler(FileDataFlowJob job, CoordinatorRegistryCenter registryCenter) {
LiteJobConfiguration configuration = createJobConfiguration(job.getClass(), "0 0/1 * * * ?", 4, "0=text,1=image,2=radio,3=vedio", true);
return new SpringJobScheduler(job, registryCenter, configuration);
}
@Bean
public CoordinatorRegistryCenter registryCenter(@Value("${zookeeper.url}") String url, @Value("${zookeeper.groupName}") String groupName) {
// ZookeeperConfiguration("zookeeper地址", "项目名")
ZookeeperConfiguration configuration = new ZookeeperConfiguration(url, groupName);
// 设置节点超时时间
configuration.setConnectionTimeoutMilliseconds(100);
ZookeeperRegistryCenter center = new ZookeeperRegistryCenter(configuration);
center.init();
return center;
}
/**
* 定时任务配置
* 这个定时任务使用的场景比较灵活，因此不建议放在spring的容器当中
*
* @param clazz 定时任务的字节码
* @param cron cron表达式
* @param shardingCount 分片数量
* @param shardingParam 分片参数
* @param isDataFlow 是否是DataFlow类型
* @return 定时任务配置对象
*/
private static LiteJobConfiguration createJobConfiguration(Class clazz, String cron, int shardingCount,
String shardingParam, boolean isDataFlow) {
// 定义作业核心配置 newBuilder("任务名称", "corn表达式", "分片数量")
JobCoreConfiguration.Builder jobBuilder = JobCoreConfiguration.newBuilder(
clazz.getSimpleName(), cron, shardingCount);
if (!StringUtils.isEmpty(shardingParam)) {
jobBuilder.shardingItemParameters(shardingParam); // 添加分片功能
}
JobCoreConfiguration simpleCoreConfig = jobBuilder.build();
// 定义simple类型配置
JobTypeConfiguration configuration;
if (isDataFlow) {
// true 代表流处理
configuration = new DataflowJobConfiguration(simpleCoreConfig, clazz.getCanonicalName(), true);
} else {
configuration = new SimpleJobConfiguration(simpleCoreConfig, clazz.getCanonicalName());
}
// 定义Lite作业根配置，并返回
// 设置overwrite(true)，允许覆盖cron表达式（默认不允许）
return LiteJobConfiguration.newBuilder(configuration).overwrite(true).build();
}
}
2、定义新的 DataFlow 任务调度对象
@Component
public class FileDataFlowJob implements DataflowJob<FileCustom> {
@Autowired
private FileCustomMapper fileCustomMapper;
/**
* 抓取数据
*
* @param shardingContext
* @return
*/
@Override
public List<FileCustom> fetchData(ShardingContext shardingContext) {
// 取决于数据能否抓取到数据，有数据会继续调用该方法
// 如果没数据就会停止，此次定时任务执行停止
// 直到下次任务调度接着抓取
System.out.println("开始抓取数据...");
// select * from t_file_custom where backedUp = 0 and type = #{type} limit #{count}
List<FileCustom> fileCustomList = fileCustomMapper.selectLimit(shardingContext.getShardingParameter(), 2); // 查找 backedUp=0 的前两条数据
return fileCustomList; // 如果为null，则直接返回；如果不为null，则调用下方方法处理数据
}
/**
* 处理数据
*
* @param shardingContext
* @param list
*/
@Override
public void processData(ShardingContext shardingContext, List<FileCustom> list) {
for (FileCustom fileCustom : list) {
backUp(fileCustom);
}
}
/**
* 模拟备份操作
*
* @param fileCustom 备份对象
*/
private void backUp(FileCustom fileCustom) {
System.out.println("备份的方法名：" + fileCustom.getName() + "，备份的类型：" + fileCustom.getType());
System.out.println("==========================");
try {
TimeUnit.SECONDS.sleep(1); // 延时一秒
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
fileCustomMapper.changeState(fileCustom.getId(), 1);
}
}
3、启动项目发现，每次抓取两条数据，重复执行，有数据会继续调用该方法，如果没数据就会停止，此次定时任务执行停止。直至下次任务调度接着抓取。
文章参考：Java微服务商城高并发秒杀项目实战|Spring Cloud Alibaba真实项目实战+商城双11秒杀+高并发+消息+支付+分布式事物Seata
![[晕事]今天做了件晕事44 wireshark 首选项IPv4：Reassemble Fragented IPv4 datagrams](https://i-blog.csdnimg.cn/direct/dacc8abc55934d598b791cf1d8e2882a.png)
![[综述笔记]Federated learning for medical image analysis: A survey](https://i-blog.csdnimg.cn/direct/73a745c13278481c8e05c19be4d22b85.png)