xxl-job之框架讲解和使用

2024-09-26  本文已影响0人  上善若泪

1 xxl-job

1.1 前言

1.1.1 xxl-job简介

XXL-JOB 是一个分布式任务调度平台，其核心设计目标是开发迅速、学习简单、轻量级、易扩展。

设计思想：将调度行为抽象形成 调度中心 平台，平台本身不承担业务逻辑，而是负责发起 调度请求 后，由 执行器 接收调度请求并执行 任务，这里的 任务 抽象为 分散的 JobHandler。通过这种方式即可实现 调度 与 任务 相互解耦，从而提高系统整体的稳定性和拓展性。

官网的架构图：


image.png

1.1.2 任务调度

在开发项目时可能遇到过类似的场景问题：

这些场景问题都可以通过 任务调度 来解决，任务调度指的是系统在约定的指定时间自动去执行指定的任务的过程。

1.1.3 分布式任务调度平台

在分布式下，每个服务都可以搭建为集群，这样的好处是可以将任务切片分给每一个服务从而实现并行执行，提高任务调度的处理效率。那么为什么 分布式系统 不能使用 单体系统 的任务调度实现方式呢。

在集群服务下，如果还是使用每台机器按照单体系统的任务调度实现方式实现的话，会出现下面这四个问题：

通过上面的问题可以了解到分布式系统下需要一个满足高可用、容错管理、负载均衡等功能的任务调度平台来实现任务调度。分布式系统下，也有许多可以实现任务调度的第三方的分布式任务调度系统，如 xxl-job、Quartz、elastic-job 等等常用的分布式任务调度系统。

1.2 使用 xxl-job

作为开源软件的 xxl-job，可以在 github 或 gitee 上查看和下载 xxl-job 的源码。
gitee地址：https://gitee.com/xuxueli0323/xxl-job
github 地址：https://github.com/xuxueli/xxl-job

1.2.1 dokcer 安装 xxl-job

1.2.1.1 拉取镜像

docker 下拉取 xxl-job 的镜像（这里使用 2.3.1 版本）
docker pull xuxueli/xxl-job-admin:2.3.1

创建映射容器的文件目录
mkdir -p -m 777 /mydata/xxl-job/data/applogs

1.2.1.2 创建配置文件

在 /mydata/xxl-job 的目录下创建 application.properties 文件
application.properties 的具体路径如图：

image.png

这里需要注意数据库位置的填写：


image.png

如果还需要更改端口的可以更改这里：


image.png

这里还需要注意告警邮箱和访问口令（后续Spring Boot配置用到）：


image.png

将 tables_xxl-job.sql 文件导入上面步骤3指定的数据库（自己填写的那个数据库）
sql获取的路径图：


image.png

1.2.1.3 执行 docker 命令

注意这里的 -p 8088:8088 是因为更改了前面 application.porperties 文件的端口号为 8088，所以这里执行的 docker 命令为 -p 8088:8088，如果没有更改的这里一定要改为 -p 8080:8080。

docker run  -p 8088:8088 \
-d --name=xxl-job-admin --restart=always \
-v /mydata/xxl-job/application.properties:/application.properties \
-v /mydata/xxl-job/data/applogs:/data/applogs \
-e PARAMS='--spring.config.location=/application.properties' xuxueli/xxl-job-admin:2.3.1

执行后通过 docker ps 查看是否成功运行，如果失败可以通过 docker logs xxl-job-admin 查看具体错误日志。

1.2.1.4 登录查看

通过 http://127.0.0.1:8088/xxl-job-admin/ 访问（这里ip和端口是自己的）
账号：admin 密码：123456

1.2.2 Spring Boot 项目集成 xxl-job

xxl-job 由 调度中心 和 执行器 组成，上面已经完成了在 docker 上部署调度中心了，接下来介绍怎么配置部署执行器项目。
点击此处了解 xxl-job之API的方式接入

1.2.2.1 pom依赖与配置文件

在 Spring Boot 项目中导入 maven 依赖

<dependency>
    <groupId>com.xuxueli</groupId>
    <artifactId>xxl-job-core</artifactId>
    <version>2.3.1</version>
</dependency>

这里需要注意版本号与 xxl-job 版本需要一致，这里我配置的都是 2.3.1 版本。

在 Spring Boot 项目中配置 application.yml 文件

xxl:
  job:
    admin:
      addresses: http://192.168.101.25:8088/xxl-job-admin
    executor:
      appname: media-process-service
      address:
      ip:
      port: 9999
      logpath: /data/applogs/xxl-job/jobhandler
      logretentiondays: 30
    accessToken: default_token

配置说明：

1.2.2.2 编写配置类

@Slf4j
@Configuration
public class XxlJobConfig {
    @Value("${xxl.job.admin.addresses}")
    private String adminAddresses;

    @Value("${xxl.job.accessToken}")
    private String accessToken;

    @Value("${xxl.job.executor.appname}")
    private String appname;

    @Value("${xxl.job.executor.address}")
    private String address;

    @Value("${xxl.job.executor.ip}")
    private String ip;

    @Value("${xxl.job.executor.port}")
    private int port;

    @Value("${xxl.job.executor.logpath}")
    private String logPath;

    @Value("${xxl.job.executor.logretentiondays}")
    private int logRetentionDays;

    @Bean
    public XxlJobSpringExecutor xxlJobExecutor() {
        log.info(">>>>>>>>>>> xxl-job config init.");
        XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
        xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
        xxlJobSpringExecutor.setAppname(appname);
        xxlJobSpringExecutor.setAddress(address);
        xxlJobSpringExecutor.setIp(ip);
        xxlJobSpringExecutor.setPort(port);
        xxlJobSpringExecutor.setAccessToken(accessToken);
        xxlJobSpringExecutor.setLogPath(logPath);
        xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);
        return xxlJobSpringExecutor;
    }
}

1.2.2.3 调度中心中新增执行器

image.png

执行器的配置属性：

1.2.2.4 配置自定义任务

配置自定义任务有许多种模式，如 Bean模式（基于方法）、Bean模式（基于类）、GLUE模式等等。这里介绍通过 Bean模式（基于方法） 是如何自定义任务的。

Bean模式（基于方法）也就是每个任务对应一个方法，通过添加 @XxLJob(value="自定义JobHandler名称", init = "JobHandler初始化方法", destroy = "JobHandler销毁方法") 注解即可完成定义。

/**
 * 任务处理类
 */
@Component
public class TestJob {
    /**
     * 测试任务
     */
    @XxlJob("testHandler")
    public void testHandler() {
        XxlJobHelper.log("start.... ");
        XxlJobHelper.handleSuccess("本次测试任务调度成功");
    }
}

1.2.2.5 调度中心中新增任务及参数说明

image.png

这里主要注意 Cron 表达式的时间配置以及 JobHandler 的值需要与自定义任务方法的注解上的 value 属性值一致即可。

高级配置:

1.3 xxl-job 实战分片任务

1.3.1 实战背景

假如项目需要对上传到分布式文件系统 minio 中的视频文件进行统一格式的视频转码操作，由于视频转码操作会带了很大的时间消耗以及 CPU 的开销，所以考虑集群服务下使用 xxl-job 的方式以任务调度的方式定时处理视频转码操作。
点击了解 SpringBoot整合Minio

这样可以带来两个好处：

1.3.2 xxl-job

1.3.2.1 执行流程图

image.png

1.3.2.2 怎么将任务均分给每台服务器

由于任务执行时间过长，需要搭建集群服务来做到并行任务调度，从而减小 CPU 的开销，那么怎么均分任务呢？

利用 xxl-job 在集群部署时，配置路由策略中选择 分片广播 的方式，可以使一次任务调度会广播触发集群中所有的执行器执行一次任务，并且可以向系统传递分片参数。

image.png

利用这一特性可以根据 当前执行器的分片序号和分片总数 来获取对应的任务记录。

先来看看 Bean 模式下怎么获取分片序号和分片总数：

// 分片序号（当前执行器序号）
int shardIndex = XxlJobHelper.getShardIndex();
// 分片总数（执行器总数）
int shardTotal = XxlJobHelper.getShardTotal();

有了这两个属性，当执行器扫描数据库获取记录时，可以根据 取模 的方式获取属于当前执行器的任务，可以这样编写 sql 获取任务记录：

select * from media_process m
where m.id % #{shareTotal} = #{shareIndex}  
  and (m.status = '1' or m.status = '3')
  and m.fail_count &lt; 3
limit #{count}

扫描任务表，根据任务 id 对分片总数 取模 来实现对所有分片的均分任务，通过判断是否是当前分片序号，并且当前任务状态为 1（未处理）或 3（处理失败）并且当前任务失败次数小于3次时可以取得当前任务。每次扫描只取出 count 个任务数（批量处理）。

image.png

因此通过 xxl-job 的分片广播 ＋ 取模 的方式即可实现对集群服务均分任务的操作。

1.3.2.3 怎么确保任务不会被重复消费

由于视频转码本身处理时间就会比较长，所以更不允许服务重复执行，虽然上面通过分片广播＋取模的方式提高了任务不会被重复执行的机率，但是依旧存在如下情况：
如下图，有三台集群机器和六个任务，刚开始分配好了每台机器两个任务，执行器0正准备执行任务3时，刚好执行器2宕机了，此时执行器1刚好执行一次任务，因为分片总数减小，导致执行器1重新分配到需要执行的任务正好也是任务3，那么此时就会出现执行器0和执行器1都在执行任务3的情况。

image.png

那么这种情况就需要实现幂等性了，幂等性有很多种实现方法

这里使用乐观锁的方式实现幂等性，具体 sql 如下：

update media_process m
set m.status = '2'
where (m.status = '1' or m.status = '3')
  and m.fail_count &lt; 3
  and m.id = #{id}

这里只需要依靠任务的状态即可实现（未处理1；处理中2；处理失败3；处理成功4），可以看到这里类似于 CAS 的方式通过比较和设置的方式只有在状态为未处理或处理失败时才能设置为处理中。这样在并发场景下，即使多个执行器同时处理该任务，也只有一个任务可以设置成功进入处理任务阶段。

为了真正达到幂等性，还需要设置一下 xxl-job 的调度过期策略和阻塞处理策略来保证真正的幂等性。分别设置为 忽略（调度过期后，忽略过期的任务，从当前时间开始重新计算下次触发时间） 和 丢弃后续调度（调度请求进入单机执行器后，发现执行器存在运行的调度任务，本次请求将会被丢弃并标记为失败）。

image.png

1.3.3 其他操作

1.3.3.1 分片视频转码处理

/**
 * 视频转码处理任务
 */
@XxlJob("videoTranscodingHandler")
public void videoTranscodingHandler() throws InterruptedException {
    int shardTotal = XxlJobHelper.getShardTotal();//总分片数
    int shardIndex = XxlJobHelper.getShardIndex();//当前分片数
    // 1. 分片获取当前执行器需要执行的所有任务
    List<MediaProcess> mediaProcessList = mediaProcessService.getMediaProcessList(shardIndex, shardTotal, count);
    // 通过JUC工具类阻塞直到所有任务执行完
    CountDownLatch countDownLatch = new CountDownLatch(mediaProcessList.size());
    // 遍历所有任务
    mediaProcessList.forEach(mediaProcess -> {
        // 以多线程的方式执行所有任务
        executor.execute(() -> {
            try {
                // 2. 尝试抢占任务（通过乐观锁实现）
                boolean res = mediaProcessService.startTask(id);
                if (!res) {
                    XxlJobHelper.log("任务抢占失败，任务id{}", id);
                    return;
                }
                
                // 3. 从minio中下载视频到本地
                File file = mediaFileService.downloadFileFromMinIO(bucket, objectName);
                // 下载失败
                if (file == null) {
                    XxlJobHelper.log("下载视频出错,任务id:{},bucket:{},objectName:{}", id, bucket, objectName);
                    // 出现异常重置任务状态为处理失败等待下一次处理
                    mediaProcessService.saveProcessFinishStatus(id, Constants.MediaProcessCode.FAIL.getValue(), fileId, null, "下载视频到本地失败");
                    return;
                }
                
                // 4. 视频转码
                String result = videoUtil.generateMp4();
                if (!result.equals("success")) {
                    XxlJobHelper.log("视频转码失败,原因:{},bucket:{},objectName:{},", result, bucket, objectName);
                    // 出现异常重置任务状态为处理失败等待下一次处理
                    mediaProcessService.saveProcessFinishStatus(id, Constants.MediaProcessCode.FAIL.getValue(), fileId, null, "视频转码失败");
                    return;
                }
                
                // 5. 上传转码后的文件
                boolean b1 = mediaFileService.addMediaFilesToMinIO(new_File.getAbsolutePath(), "video/mp4", bucket, objectNameMp4);
                if (!b1) {
                    XxlJobHelper.log("上传 mp4 到 minio 失败,任务id:{}", id);
                    // 出现异常重置任务状态为处理失败等待下一次处理
                    mediaProcessService.saveProcessFinishStatus(id, Constants.MediaProcessCode.FAIL.getValue(), fileId, null, "上传 mp4 文件到 minio 失败");
                    return;
                }

                // 6. 更新任务状态为成功
                mediaProcessService.saveProcessFinishStatus(id, Constants.MediaProcessCode.SUCCESS.getValue(), fileId, url, "创建临时文件异常");
                
            } finally {
                countDownLatch.countDown();
            }
        });
    });
    // 阻塞直到所有方法执行完成（30min后不再等待）
    countDownLatch.await(30, TimeUnit.MINUTES);
}

核心任务 - 分片获取任务后执行视频转码任务，步骤如下：

1.3.3.2 视频补偿机制

由于使用乐观锁会将任务状态更新为处理中，如果此时执行任务的执行器（服务）宕机了，会导致该任务记录一直存在，因为乐观锁的原因别的执行器也无法获取，这个时候同样需要使用任务调度的方式，定期扫描任务表，判断任务是否处于处理中状态并且任务创建时间远大于30分钟，则说明任务超时了，则是使用任务调度的方式重新更新任务的状态为未处理，等待下一次视频转码任务的调度处理。此外视频补偿机制任务调度还需要检查是否存在任务最大次数已经大于3次的，如果存在则交付给人工处理。

主要实现两个功能：① 处理任务超时情况下的任务，做出补偿；② 处理失败次数大于3次的任务，做出补偿；

1.3.3.3 测试并查看日志

准备好的任务表记录：


image.png

启动三台媒资服务器，并开启任务：


image.png

可以单独查看每个任务的日志：


image.png

通过日志中的执行日志查看具体日志信息：


image.png

可以看到直接为了测试改错的路径导致下载视频出错：


image.png

查看数据库表的变化：


image.png
上一篇 下一篇

猜你喜欢

热点阅读