分布式调度方案：Elastic-Job

文章目录

一、什么是分布式调度
二、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 {@Overridepublic 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:2181groupName: elastic-job-boot

3、调度任务类交给 Spring 进行管理

@Component
public class MyElasticJob implements SimpleJob {@Overridepublic 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);}@Beanpublic 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);}@Beanpublic 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 {@Autowiredprivate FileCustomMapper fileCustomMapper;@Overridepublic 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);}@Beanpublic 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> {@Autowiredprivate FileCustomMapper fileCustomMapper;/*** 抓取数据** @param shardingContext* @return*/@Overridepublic 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*/@Overridepublic 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);}
}