Elastic-Job相关

文档参考视频：09_SpringBoot案例演示_哔哩哔哩_bilibili

一、Elastic-Job介绍

Elastic-Job 是一个轻量级、分布式的任务调度框架，旨在解决分布式环境下的定时任务调度问题。

1.1. Elastic-Job 的核心组件

Elastic-Job 是由多个核心组件构成的，每个组件负责不同的任务调度和分布式管理功能。以下是 Elastic-Job 的主要核心组件：

1. 作业（Job）

• 定义：作业是 Elastic-Job 中的核心概念，代表需要被调度执行的任务。Elastic-Job 提供了两种类型的作业：

  • SimpleJob：最简单的作业类型，适用于需要简单定时执行的任务。它只需实现 SimpleJob 接口，并实现 execute(ShardingContext shardingContext) 方法即可。
  • DataflowJob：适用于数据流处理场景，作业会分批处理数据，并且支持读取和写入数据。

• 作用：作业负责处理实际的业务逻辑，Elastic-Job 框架提供了调度机制来管理这些作业在集群中的分配和执行。

2. 作业配置（Job Configuration）

• 定义：作业配置是 Elastic-Job 中用来配置作业调度规则、分片策略等的配置对象。它包含了以下几个重要部分：

  • JobCoreConfiguration：作业的核心配置，包含作业名称、Cron 表达式、分片数、分片参数等。
  • SimpleJobConfiguration：用于简单作业的配置，指定作业类型（如 SimpleJob）和核心配置。
  • LiteJobConfiguration：用于 Lite 作业的配置，继承了 SimpleJobConfiguration，是最终的作业配置对象，包含了所有作业调度的设置。

• 作用：通过作业配置，你可以灵活地设置作业调度的各项参数，如执行频率、作业分片、分片参数、容错设置等。

3. 作业调度器（JobScheduler）

• 定义：作业调度器是负责执行作业的核心组件，它会定时触发作业的执行。Elastic-Job 中有不同类型的调度器，最常用的是 SpringJobScheduler，它将作业调度与 Spring 集成。
• 作用：作业调度器负责调度作业的执行，监听作业配置的变化，并按设定的 Cron 表达式或时间间隔执行作业。它也负责处理作业的生命周期，例如初始化、暂停、停止等操作。

4. 注册中心（Registry Center）

• 定义：Elastic-Job 通过注册中心来协调分布式作业的执行，确保作业在多个节点之间正确分配和执行。常用的注册中心是 Zookeeper，Elastic-Job 支持将 Zookeeper 作为作业的注册中心。

• 作用：注册中心用于存储作业的调度信息和状态，包括作业执行的分片信息、作业的执行状态、任务是否失败等。Elastic-Job 使用注册中心来保证作业调度的协调性，避免作业的重复执行和确保作业的高可用性。

• 注册中心的功能：

  • 协调作业的分片分配：Elastic-Job 会将作业的分片信息存储在注册中心，确保每个分片只会在一个节点上执行。
  • 作业状态管理：通过注册中心，Elastic-Job 可以实时获取作业的执行状态，并根据需要做出调整（例如重新分配分片）。
  • 容错与恢复：如果某个节点失效，注册中心会重新分配失败节点的作业分片。

5. 分片策略（Sharding Strategy）

• 定义：Elastic-Job 使用分片机制来将作业拆分为多个任务片段（分片），这些分片会被不同的节点（实例）执行。作业的分片可以是静态的（固定分片）或动态的（动态调整）。
• 作用：分片策略控制作业分片的分配，确保作业的任务在不同节点之间平衡负载，避免单个节点承受过多的压力。Elastic-Job 提供了多个分片策略（例如：按数据划分、按任务划分等）来灵活适应不同的业务场景。

6. 作业执行上下文（ShardingContext）

• 定义：ShardingContext 是作业执行时的上下文信息，它包含了当前作业的分片编号、作业名称、作业实例编号等信息。
• 作用：ShardingContext 提供给作业执行方法 execute(ShardingContext shardingContext) 使用，可以让作业根据自己的分片编号来执行相应的逻辑，确保作业在分布式环境下正确执行。

1.2. Elastic-Job 的工作原理

Elastic-Job 的工作原理大致如下：

1. 作业注册与调度：作业类实现了 SimpleJob 接口并定义了作业的执行逻辑，然后通过作业配置创建作业的调度配置。作业调度器（如 SpringJobScheduler）会读取这些配置，启动调度任务。

2. 作业分片：作业会根据配置的分片数将任务划分为多个片段，每个分片负责处理一部分任务。每个作业实例（节点）会根据自己的分片编号执行相应的任务片段。

3. 注册中心协调：作业的调度和分片信息会存储在注册中心（如 Zookeeper）中，Elastic-Job 使用注册中心来保证作业分片的协调性，避免重复执行，并确保作业的高可用性和容错性。

4. 作业执行：作业调度器会按照配置的 Cron 表达式或时间间隔触发作业的执行，并通过 ShardingContext 将分片信息传递给作业执行方法。每个作业实例根据分片信息执行任务。

5. 容错与失败重试：如果作业执行失败，Elastic-Job 会根据配置进行失败重试。并且，如果某个节点故障，Zookeeper 会重新分配该节点的分片，确保作业继续执行。

二、Spring boot集成Elastic-Job

下面是一个基本的 Spring Boot 集成 Elastic-Job 的完整实现步骤。

1.安装并启动zookeeper

下载地址：Index of /apache/zookeeper

Zookeeper 安装包解压后，里面会有一个 conf 文件夹，在该文件夹中有一个示例配置文件 zoo_sample.cfg。需要复制并重命名为 zoo.cfg，然后进行修改。

1. 进入 Zookeeper 解压后的目录，找到 conf 文件夹。

2. 复制 zoo_sample.cfg 文件，并将其重命名为 zoo.cfg。

3. 打开 zoo.cfg 配置文件，进行必要的配置修改。

# Zookeeper 数据目录，保存事务日志和快照
dataDir=C:/zookeeper/data# Zookeeper 的端口号（默认是 2181）
clientPort=2181

4. 启动zookeeper

进入 Zookeeper 安装目录中的 bin 文件夹。然后双击启动 zkServer.cmd 文件。

5. 测试 Zookeeper

Zookeeper 启动后，使用  zkCli.cmd 命令行工具连接到 Zookeeper，测试是否能够正常连接。进入 bin 文件夹，然后双击运行 zkCli.cmd 文件。输入以下内容进行测试：

[zk: localhost:2181(CONNECTED) 0] ls /
[zookeeper]
[zk: localhost:2181(CONNECTED) 1] create /my_node "Hello Zookeeper"
Created /my_node
[zk: localhost:2181(CONNECTED) 2] get /my_node
Hello Zookeeper

2.添加依赖

这里用的是2.1.5的版本

	<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><!--elastic-job--><dependency><groupId>com.dangdang</groupId><artifactId>elastic-job-lite-spring</artifactId><version>2.1.5</version></dependency></dependencies>

3.配置Zookeeper

application.yml文件中配置

spring:application:name: elastic-job-demo
elasticjob:zookeeper-url: localhost:2181namespace: elastic-job-demo
server:port: 8081

4.创建作业类

Elastic-Job 通过实现 SimpleJob 接口来定义一个作业。我们需要在作业类中实现 execute 方法，定义作业的具体业务逻辑。这里打印当前日期。

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;import java.util.Date;@Component
public class MyJob implements SimpleJob {private static final Logger log = LoggerFactory.getLogger(MyJob.class);@Overridepublic void execute(ShardingContext shardingContext) {log.info("=============={}============",new Date());}
}

5.创建和配置 Zookeeper 注册中心

Elastic-Job 依赖 Zookeeper 来存储作业调度的元数据（例如作业状态、分片信息等），并且使用 Zookeeper 来协调和管理作业的执行。

ZookeeperConfig 类的配置主要目的是确保 Elastic-Job 能够在分布式环境下正确执行作业，利用 Zookeeper 来进行作业的协调、分片管理和故障恢复。通过将 CoordinatorRegistryCenter 配置为 Spring Bean，Elastic-Job 可以在 Spring Boot 应用启动时自动连接到 Zookeeper，实现分布式作业的调度和管理。

/***  创建和配置 Zookeeper 注册中心，并将其注入到 Spring 容器中。*/
@Configuration
public class ZookeeperConfig {@Bean(initMethod = "init")public CoordinatorRegistryCenter createZookeeperCenter(@Value("${elasticjob.zookeeper-url}") String zookeeperUrl,@Value("${elasticjob.namespace}") String groupName) {//zk的配置 url 和命名空间名字ZookeeperConfiguration zookeeperConfiguration = new ZookeeperConfiguration(zookeeperUrl, groupName);//设置zk超时时间zookeeperConfiguration.setSessionTimeoutMilliseconds(10000);//创建注册中心CoordinatorRegistryCenter zookeeperRegistryCenter = new ZookeeperRegistryCenter(zookeeperConfiguration);return  zookeeperRegistryCenter;}
}

6.配置和初始化 Elastic-Job 的调度器

        在 Spring Boot 集成 Elastic-Job 的过程中，除了配置 ZookeeperConfig 来管理作业的调度和分布式协调之外，还需要配置 ElasticJobConfig 来创建和初始化作业调度器 (SpringJobScheduler) 并配置作业的具体执行规则。

   ElasticJobConfig 主要是将作业的调度配置和作业任务与注册中心（Zookeeper）进行集成，从而实现具体的作业调度和执行管理。

@Configuration
public class ElasticJobConfig {@Autowiredprivate MyJob myJob;@Autowiredprivate CoordinatorRegistryCenter registryCenter;/*** 创建作业的核心配置* 这个方法的目的是 创建一个作业配置对象，并根据传入的参数（如作业类、Cron 表达式、分片数量等）生成配置，* 最终返回一个 LiteJobConfiguration 对象。该对象用于配置 Elastic-Job 中的作业调度，涉及到作业的名称、执行周期（Cron 表达式）、分片策略等。** @param jobClass               作业类* @param cron                   Cron 表达式* @param shardingTotalCount     总分片数* @param shardingItemParameters 分片项参数* @return LiteJobConfiguration 作业的配置对象*/private static LiteJobConfiguration createJobConfiguration(final Class<? extends SimpleJob> jobClass,final String cron,final int shardingTotalCount,final String shardingItemParameters) {// 定义作业核心配置 任务名称 cron表达式 分片数量JobCoreConfiguration.Builder jobCoreConfigurationBuilder =JobCoreConfiguration.newBuilder(jobClass.getSimpleName(), cron, shardingTotalCount);// 如果分片项参数不为空，设置它if (!StringUtils.isEmpty(shardingItemParameters)) {jobCoreConfigurationBuilder.shardingItemParameters(shardingItemParameters);}// 定义 SIMPLE 类型作业配置SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(jobCoreConfigurationBuilder.build(),jobClass.getCanonicalName() // 作业类的完整类名);// 定义 Lite 作业根配置LiteJobConfiguration liteJobRootConfig = LiteJobConfiguration.newBuilder(simpleJobConfig).overwrite(true)  // 覆盖已有作业.build();return liteJobRootConfig;}//初始化并配置一个 Elastic-Job 的作业调度器（SpringJobScheduler）@Bean(initMethod = "init")public SpringJobScheduler initSimpleElasticJob() {// 创建 SpringJobScheduler 实例 传入作业类、注册中心、LiteJobConfiguration 对SpringJobScheduler springJobScheduler = new SpringJobScheduler(myJob, registryCenter, createJobConfiguration(myJob.getClass(),"0/3 * * * * ?", 1, null));return springJobScheduler;}}

7.创建启动类

@SpringBootApplication
public class ElasticJobDemoApplication {public static void main(String[] args) {SpringApplication.run(ElasticJobDemoApplication.class, args);}}

8.测试

启动之后，可以在控制台看到日志：

 这里我们可以在不同端口启动多个实例：

注意添加VM参数，指定不同的端口号，然后同时启动。可以看到在没有分片的情况下，任务只会在一台实例上运行。

Elastic-Job 使用 Zookeeper 作为注册中心来协调不同节点的作业执行。当在不同端口上启动多个实例时，Zookeeper 会确保作业的分片不会重复执行。例如，如果一个节点失败或宕机，Zookeeper 会重新分配该节点的分片到其他健康的节点，确保作业继续执行，不会出现丢失或重复执行的情况。

 三、模拟备份数据

这里模拟备份文件，以backedUp 为标识，1代表备份，0代表未备份。

1.建表

CREATE TABLE file_info(id INT PRIMARY KEY,name VARCHAR(50),content VARCHAR(255),type VARCHAR(50),backedUp TINYINT
);INSERT INTO file_info(id, name, content, type, backedUp) VALUES
(1, '文件1', '内容1', 'text', 1),
(2, '文件2', '内容2', 'text', 1),
(3, '文件3', '内容3', 'text', 1),
(4, '文件4', '内容4', 'image', 1),
(5, '文件5', '内容5', 'image', 1),
(6, '文件6', '内容6', 'radio', 1),
(7, '文件7', '内容7', 'radio', 1),
(8, '文件8', '内容8', 'vedio', 1),
(9, '文件9', '内容9', 'vedio', 1),
(10, '文件10', '内容10', 'vedio', 1),
(11, '文件11', '内容11', 'text', 1),
(12, '文件12', '内容12', 'image', 1),
(13, '文件13', '内容13', 'radio', 1),
(14, '文件14', '内容14', 'vedio', 1),
(15, '文件15', '内容15', 'image', 1),
(16, '文件16', '内容16', 'text', 1),
(17, '文件17', '内容17', 'radio', 1),
(18, '文件18', '内容18', 'vedio', 1),
(19, '文件19', '内容19', 'radio', 1),
(20, '文件20', '内容20', 'vedio', 1);

2.添加依赖

		<dependency><groupId>com.alibaba</groupId><artifactId>druid</artifactId><version>1.1.18</version></dependency><dependency><groupId>org.mybatis.spring.boot</groupId><artifactId>mybatis-spring-boot-starter</artifactId><version>2.0.0</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.33</version></dependency>

3.添加配置

spring:application:name: elastic-job-demodatasource:url: jdbc:mysql://localhost:3306/test?useSSL=false&serverTimezone=UTC&characterEncoding=utf8username: rootpassword: 123456driver-class-name: com.mysql.cj.jdbc.Driver
elasticjob:zookeeper-url: localhost:2181namespace: elastic-job-demo
server:port: 8081

4.添加实体类

public class FileInfo {private int id;private String name;private String content;private String type;private boolean backedUp;// 无参构造函数public FileInfo() {}// getter 和 setter 方法public int getId() {return id;}public void setId(int id) {this.id = id;}public String getName() {return name;}public void setName(String name) {this.name = name;}public String getContent() {return content;}public void setContent(String content) {this.content = content;}public String getType() {return type;}public void setType(String type) {this.type = type;}public boolean isBackedUp() {return backedUp;}public void setBackedUp(boolean backedUp) {this.backedUp = backedUp;}@Overridepublic String toString() {return "FileInfo{" +"id=" + id +", name='" + name + '\'' +", content='" + content + '\'' +", type='" + type + '\'' +", backedUp=" + backedUp +'}';}
}

5.添加mapper

@Mapper
public interface FileInfoMapper {@Select("select id, name, content, type, backedUp from file_info where backedUp=0")public abstract List<FileInfo> selectAll();@Update("update file_info set backedUp = #{state} where id = #{id}")public abstract int changeState(@Param("id") int id, @Param("state") int state);
}

6.job对象

@Component
public class FileBackupElasticJob implements SimpleJob {@Autowiredprivate FileInfoMapper fileInfoMapper;@Overridepublic void execute(ShardingContext shardingContext) {doWork();}private void doWork() {List<FileInfo> fileList = fileInfoMapper.selectAll();System.out.println("需要备份文件个数: " + fileList.size());for (FileInfo FileInfo : fileList) {backUpFile(FileInfo);}}private void backUpFile(FileInfo fileInfo) {try {// 模拟备份动作Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("执行文件备份 ===> " + fileInfo);fileInfoMapper.changeState(fileInfo.getId(), 1);}
}

7.配置调度器

    @Bean(initMethod = "init")public SpringJobScheduler initSimpleElasticJob(FileBackupElasticJob fileBackupElasticJob) {// 创建 SpringJobScheduler 实例 传入作业类、注册中心、LiteJobConfiguration 对SpringJobScheduler springJobScheduler = new SpringJobScheduler(fileBackupElasticJob, registryCenter, createJobConfiguration(fileBackupElasticJob.getClass(),"0 0/1 * * * ?", 1, null));return springJobScheduler;}

启动类：

@SpringBootApplication
@MapperScan("com.example.elasticjobdemo.mapper")
public class ElasticJobDemoApplication {public static void main(String[] args) {SpringApplication.run(ElasticJobDemoApplication.class, args);}}

 

8.测试

9.作业分片

定义分片规则：4分片，规则："0=text,1=image,2=radio,3=vedio"

新增根据类型查询数据方法：

执行的doWork方法把ShardingParameter传进去，也就是把0=text 的text传进去。

根据传入的类型进行处理，备份对应类型的文件。

测试，可以看到不同的端口，备份的文件类型是不一样的