1 xxl-job

1.1 前言

1.1.1 xxl-job简介

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

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

官网的架构图：

image.png

1.1.2 任务调度

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

• 系统需要定时在每天0点进行数据备份。
• 系统需要在活动开始前几小时预热执行一些前置业务。
• 系统需要定时对 MQ 消息表的发送装填，对发送失败的 MQ 消息进行补偿重新发送。

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

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

配置说明：

• xxl.job.admin.addresses 用于指定调度中心的地址。
• xxl.job.accessToken 用于指定访问口令（也就是前面搭建 xxl-job 中步骤3指定的）。
• xxl.job.executor.appname 用于指定执行器的名称（需要与后续配置执行器的名称一致）。
• xxl.job.executor.port 用于指定执行器的端口（执行器实际上是一个内嵌的 Server，默认端口为9999，配置多个同一服务实例时需要指定不同的执行器端口，否则会端口冲突）。

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

执行器的配置属性：

• 自动注册：执行器自动进行执行器注册，调度中心通过底层注册表可以动态发现执行器机器地址。
• 手动录入：人工手动录入执行器的地址信息，多地址逗号分隔，供调度中心使用。
• AppName：每个执行器集群的唯一标示 AppName，执行器会周期性以 AppName 为对象进行自动注册。可通过该配置自动发现注册成功的执行器，供任务调度时使用。
• 名称：执行器的名称（可以使用中文更好地体现该执行器是用来干嘛的）。
• 注册方式：调度中心获取执行器地址的方式（一般为了方便可以选用自动注册即可）。
• 机器地址："注册方式"为"手动录入"时有效，支持人工维护执行器的地址信息。

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("本次测试任务调度成功");
    }
}

• 通过注解也可以指定 初始化方法和销毁方法，如果不填写可以直接写一个 自定义的JobHandler名称 用于后面在调度中心中配置任务时对应任务的 JobHandler 属性值。
• 可以通过 XxlJobHelper.log 来打印日志，通过调度中心可以查看执行日志的情况。
• 可以通过 XxlJobHelper.handleFail 或 XxlJobHelper.handleSuccess 手动设置任务调度的结果（不设置时默认结果为成功状态，除非任务执行时出现异常）。

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

image.png

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

高级配置:

• 路由策略：当执行器集群部署时，提供丰富的路由策略,包括：
  • FIRST(第一个)：固定选择第一个机器;
  • LAST(最后一个)：固定选择最后一个机器;
  • ROUND(轮询)：;
  • RANDOM(随机)：随机选择在线的机器;
  • CONSISTENT_HASH(一致性HASH)：每个任务按照Hash算法固定选择某一台机器，且所有任务均匀散列在不同机器上。
  • LEAST_FREQUENTLY_USED(最不经常使用)：使用频率最低的机器优先被选举;
  • LEAST_RECENTLY_USED(最近最久未使用)：最久未使用的机器优先被选举;
  • FAILOVER(故障转移)：按照顺序依次进行心跳检测，第一个心跳检测成功的机器选定为目标执行器并发起调度
  • BUSYOVER(忙碌转移)：按照顺序依次进行空闲检测，第一个空闲检测成功的机器选定为目标执行器并发起调度
  • SHARDING_BROADCAST(分片广播)：广播触发对应集群中所有机器执行一次任务，同时系统自动传递分片参数，可根据分片参数开发分片任务;
• 子任务：每个任务都拥有一个唯一的任务ID(任务ID可以从任务列表获取)，当本任务执行结束并且执行成功时，将会触发子任务ID所对应的任务的一次主动调度。
• 调度过期策略：
  • 忽略：调度过期后，忽略过期的任务，从当前时间开始重新计算下次触发时间;
  • 立即执行一次：调度过期后，立即执行一次，并从当前时间开始重新计算下次触发时间;
• 阻塞处理策略：调度过于密集执行器来不及处理时的处理策略;
  • 单机串行(默认)：调度请求进入单机执行器后，调度请求进入FIFO队列并以串行方式运行;
  • 丢弃后续调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，本次请求将会被丢弃并标记为失败;
  • 覆盖之前调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，将会终止运行中的调度任务并清空队列，然后运行本地调度任务;
• 任务超时时间：支持自定义任务超时时间，任务运行超时将会主动中断任务;
• 失败重试次数：支持自定义任务失败重试次数，当任务失败时将会按照预设的失败重试次数主动进行重试;

1.3 xxl-job 实战分片任务

1.3.1 实战背景

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

这样可以带来两个好处：

• 任务调度的方式，可以使得视频转码操作不会阻塞主线程，避免影响主要业务的吞吐量；
• 以集群服务分片接收任务的方式，可以将任务均分给每个机器使得任务调度可以并行执行，提高总任务处理时间以及降低单台机器 CPU 的开销；

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 < 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 < 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);
}

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

• 通过 分片广播 拿到的参数以取模的方式 获取当前执行器所属的任务记录集合
• 遍历集合，以 多线程的方式 并发地执行任务
• 每次执行任务前需要先通过 数据库乐观锁的方式 抢占当前任务，抢占到才能执行
• 执行任务过程分为 分布式文件系统下载需要转码的视频文件 -> 视频转码 -> 上传转码后的视频 -> 更新任务状态（处理成功）
• 使用JUC工具类 CountDownLatch 实现所有任务执行完后才退出方法
• 中间使用 xxl-job 的日志记录错误信息和执行结果

1.3.3.2 视频补偿机制

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

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

1.3.3.3 测试并查看日志

准备好的任务表记录：

image.png

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

image.png

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

image.png

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

image.png

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

image.png

查看数据库表的变化：

image.png