记一次 xxl-job 实战

前言

本篇文章主要记录项目中遇到的 xxl-job 的实战，希望能通过这篇文章告诉读者们什么是 xxl-job 以及怎么使用 xxl-job 并分享一个实战案例。

那么下面先说明什么是 xxl-job 以及为什么要使用它。

xxl-job 是什么？

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

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

为了更好理解，这里放一张官网的架构图：

任务调度是什么？

在开发项目时大家是否也遇到过类似的场景问题：

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

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

单体系统 中有许多实现 任务调度 的方式，如多线程方式、Timer 类、Spring Tasks 等等。这里比较常用的是 Spring Tasks（通过 @EnableScheduling + @Scheduled 的注解可以自定义定时任务，有兴趣的可以去了解一下）

为什么需要分布式任务调度平台？

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

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

1. 怎么做到对任务的控制（如何避免任务重复执行）。
2. 如果某台机器宕机了，会不会存在任务丢失。
3. 如果要增加服务实例，怎么做到弹性扩容。
4. 如何做到对任务调度的执行情况统一监测。

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

本篇文章将介绍 xxl-job 分布式任务调度系统。

如何使用 xxl-job

作为开源软件的 xxl-job，可以在 github 或 gitee 上查看和下载 xxl-job 的源码。

下面将介绍我使用 xxl-job 的流程（如果有操作不当的，可以查看官方的中文文档 分布式任务调度平台XXL-JOB (xuxueli.com)）

dokcer 下安装 xxl-job

1. docker 下拉取 xxl-job 的镜像（这里使用 2.3.1 版本）

   docker pull xuxueli/xxl-job-admin:2.3.1
   

2. 创建映射容器的文件目录

   mkdir -p -m 777 /mydata/xxl-job/data/applogs
   

3. 在 /mydata/xxl-job 的目录下创建 application.properties 文件

   由于 application.properties 的代码过长，这里就不展示了，需要的可以去 gitee 上获取，具体路径如图：

   

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

   

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

   

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

   

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

   同样由于文件代码过长，这里展示 gitee 上获取的路径图：

   

5. 执行 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 查看具体错误日志。

6. 通过 http://192.168.101.25:8088/xxl-job-admin/ 访问（这里ip和端口是自己的）

   

   账号：admin 密码：123456

到这里就算是完成了 xxl-job 在 docker 的搭建。

Spring Boot 项目集成 xxl-job

xxl-job 由 调度中心 和 执行器 组成，上面已经完成了在 docker 上部署调度中心了，接下来介绍怎么配置部署执行器项目。

1. 在 Spring Boot 项目中导入 maven 依赖

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

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

2. 在 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，配置多个同一服务实例时需要指定不同的执行器端口，否则会端口冲突）。
   • 其他属性只需要照着配置即可（想要了解属性的具体含义可以查看中文文档中的2.4配置部署执行器项目章节）。

3. 编写配置类

   /**
    * XXL-JOB配置类
    *
    * @author 单程车票
    */
   @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;
       }
   }
   

4. 调度中心中新增执行器

   

   执行器的配置属性：

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

5. 配置自定义任务

   配置自定义任务有许多种模式，如 Bean模式（基于方法）、Bean模式（基于类）、GLUE模式等等。这里介绍通过 Bean模式（基于方法） 是如何自定义任务的（对于其余的模式可以参考官方文档）。

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

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

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

6. 调度中心中新增任务

   

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

   关于高级配置这里放一张中文文档的详细说明（也可以直接去看文档）：

   

   需要搭建集群或过期策略等高级玩法时可以进行配置。

到这里就完成了 SpringBoot 集成 xxl-job 实现分布式任务调度的全过程了，接下来会通过一个实战案例来具体看看 xxl-job 的用处。

xxl-job 实战

下面通过一个最近自己在跟着做的学习项目中使用到 xxl-job 的场景案例来具体了解一下如何利用 xxl-job 来实现任务调度。

实战背景

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

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

xxl-job 执行流程图

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

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

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

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

先来看看 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 个任务数（批量处理）。

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

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

由于视频转码本身处理时间就会比较长，所以更不允许服务重复执行，虽然上面通过分片广播＋取模的方式提高了任务不会被重复执行的机率，但是依旧存在如下情况：

如下图，有三台集群机器和六个任务，刚开始分配好了每台机器两个任务，执行器0正准备执行任务3时，刚好执行器2宕机了，此时执行器1刚好执行一次任务，因为分片总数减小，导致执行器1重新分配到需要执行的任务正好也是任务3，那么此时就会出现执行器0和执行器1都在执行任务3的情况。

那么这种情况就需要实现幂等性了，幂等性有很多种实现方法，有兴趣了解的可以参考：开发中常遇到的接口幂等性问题及实现 - 编程 ()

这里使用乐观锁的方式实现幂等性，具体 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 的调度过期策略和阻塞处理策略来保证真正的幂等性。分别设置为 忽略（调度过期后，忽略过期的任务，从当前时间开始重新计算下次触发时间） 和 丢弃后续调度（调度请求进入单机执行器后，发现执行器存在运行的调度任务，本次请求将会被丢弃并标记为失败）。

编写完成该功能所需的所有任务

1. 分片视频转码处理

代码（这里的代码只展示部分核心步骤代码）：

/**
 * 视频转码处理任务
 */
@XxlJob("videoTranscodingHandler")
public void videoTranscodingHandler() throws InterruptedException {
    // 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 的日志记录错误信息和执行结果

2. 清理任务表中转码成功的任务的记录并将其插入任务历史表

由于任务表处理完任务后只是更新任务状态，这样随着任务增多会导致检索起来时间消耗过大，所以使用任务调度的方式定期扫描任务表，将任务状态为处理成功的任务删除并重新插入任务历史表中留存（由于代码过于简单，这里就不做展示了）。

主要实现两个功能：① 清理任务表中已成功处理的任务； ② 将处理成功的任务记录插入历史表中；

3. 视频补偿机制

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

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

测试并查看日志

准备好的任务表记录：

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

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

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

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

查看数据库表的变化：

到这里可以看到核心的视频转码任务执行成功，并且逻辑正确，能够起到分布式任务调度的作用。

这就是本次 xxl-job 实战的全部内容了，写这篇文章主要是为了记录一下项目中是如何使用 xxl-job 的，并且提供一种分片广播均分任务的思路以及幂等性问题如何处理，具体使用 xxl-job 还需根据自己项目的需求。遇到问题可以参考官网：分布式任务调度平台XXL-JOB (xuxueli.com)