XXL-JOB分布式任务调度 （从0-1项目实战）
目录
一.什么是分布式任务调度？
二.XXL-JOB介绍与部署：
1.什么是XXL-JOB？
2.部署XXL-JOB：
3.XXL-JOB的使用：
三.XXL-JOB实战例子解析：
1.高级配置：
2.分片方案：
3.业务流程：
4.抢占任务测试：
📢 本博客转载自【黑马学成在线】！
🚀 主题：XXL-JOB分布式调度框架
📖 内容涵盖：
✅ 从0到1部署 —— 手把手教你搭建XXL-JOB调度中心
✅ 源码级分析 —— 深入理解调度机制与执行原理
✅ 真实项目实战 —— 结合【学成在线】媒资业务场景，落地分布式任务调度💡 适合人群：
🔹 想学习分布式任务调度的开发者
🔹 需要解决定时任务高可用、分片问题的团队
🔹 对XXL-JOB底层实现感兴趣的技术控👇 下面阅读原文，开启分布式调度之旅吧~
一.什么是分布式任务调度？
通常任务调度的程序是集成在应用中的，比如：优惠卷服务中包括了定时发放优惠卷的的调度程序，结算服务中包括了定期生成报表的任务调度程序，由于采用分布式架构，一个服务往往会部署多个冗余实例来运行我们的业务，在这种分布式系统环境下运行任务调度，我们称之为分布式任务调度，如下图：
分布式调度要实现的目标：
不管是任务调度程序集成在应用程序中，还是单独构建的任务调度系统，如果采用分布式调度任务的方式就相当于将任务调度程序分布式构建，这样就可以具有分布式系统的特点，并且提高任务的调度处理能力：
1、并行任务调度
并行任务调度实现靠多线程，如果有大量任务需要调度，此时光靠多线程就会有瓶颈了，因为一台计算机CPU的处理能力是有限的。
如果将任务调度程序分布式部署，每个结点还可以部署为集群，这样就可以让多台计算机共同去完成任务调度，我们可以将任务分割为若干个分片，由不同的实例并行执行，来提高任务调度的处理效率。
2、高可用
若某一个实例宕机，不影响其他实例来执行任务。
3、弹性扩容
当集群中增加实例就可以提高并执行任务的处理效率。
4、任务管理与监测
对系统中存在的所有定时任务进行统一的管理及监测。让开发人员及运维人员能够时刻了解任务执行情况，从而做出快速的应急处理响应。
5、避免任务重复执行
当任务调度以集群方式部署，同一个任务调度可能会执行多次，比如在上面提到的电商系统中到点发优惠券的例子，就会发放多次优惠券，对公司造成很多损失，所以我们需要控制相同的任务在多个运行实例上只执行一次。
而XXL-JOB就可以很好的实现这些目标。
二.XXL-JOB介绍与部署：
1.什么是XXL-JOB？
XXL-JOB是一个轻量级分布式任务调度平台，其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线，开箱即用。
官网：官网地址
文档：官方文档
XXL-JOB主要有调度中心、执行器、任务：
调度中心：
负责管理调度信息，按照调度配置发出调度请求，自身不承担业务代码；
主要职责为执行器管理、任务管理、监控运维、日志管理等
任务执行器：
负责接收调度请求并执行任务逻辑；
只要职责是注册服务、任务执行服务（接收到任务后会放入线程池中的任务队列）、执行结果上报、日志服务等
任务：负责执行具体的业务处理。
调度中心与执行器之间的工作流程如下：
执行流程：
1.任务执行器根据配置的调度中心的地址，自动注册到调度中心
2.达到任务触发条件，调度中心下发任务
3.执行器基于线程池执行任务，并把执行结果放入内存队列中、把执行日志写入日志文件中
4.执行器消费内存队列中的执行结果，主动上报给调度中心
5.当用户在调度中心查看任务日志，调度中心请求任务执行器，任务执行器读取任务日志文件并返回日志详情
2.部署XXL-JOB：
首先下载XXL-JOB
GitHub：https://github.com/xuxueli/xxl-job
https://github.com/xuxueli/xxl-job
码云：xxl-job: 一个分布式任务调度平台，其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线，开箱即用。https://gitee.com/xuxueli0323/xxl-job
使用IDEA打开解压后的目录：
xxl-job-admin：调度中心
xxl-job-core：公共依赖
xxl-job-executor-samples：执行器Sample示例（选择合适的版本执行器，可直接使用）
：xxl-job-executor-sample-springboot：Springboot版本，通过Springboot管理执行器，推荐这种方式；
：xxl-job-executor-sample-frameless：无框架版本；
doc :文档资料，包含数据库脚本。
我们导入xxl-job数据库：
并修改账户密码后， 访问：http://192.168.101.65:8088/xxl-job-admin/
账号和密码：admin/123456
如果无法使用虚拟机运行xxl-job可以在本机idea运行xxl-job调度中心。
3.XXL-JOB的使用：
XXL-JOB一般分为调度中心与执行器，调度中心就是我们打开的网页：
而执行器是需要我们进行配置：
执行器负责与调度中心通信接收调度中心发起的任务调度请求。
首先进入调度中心的执行器管理：
点击新增，填写执行器信息，appname是前边在nacos中配置xxl信息时指定的执行器的应用名。
添加成功：
随后在service工程添加依赖：
<dependency><groupId>com.xuxueli</groupId><artifactId>xxl-job-core</artifactId><version>2.3.1</version>
</dependency>随后在配置中心配置相关文件：
xxl:job:admin: addresses: http://127.0.0.1:8088/xxl-job-adminexecutor:appname: media-process-service # 执行器名称（AppName）address: ip: port: 9999 # 给执行器暴露一个端口,来让调度中心调用logpath: /data/applogs/xxl-job/jobhandlerlogretentiondays: 30accessToken: default_token注意配置中的appname这是执行器的应用名，port是执行器启动的端口，如果本地启动多个执行器，注意端口不能重复。
随后将 xxl-job 示例工程下配置类拷贝到媒资管理的service工程下：
/*** @description 测试执行器* @version 1.0*/@Component@Slf4j
public class SampleJob {@XxlJob("testJob")public void testJob() throws Exception {log.info("开始执行.....");}
}到此完成媒资管理模块service工程配置xxl-job执行器，在xxl-job调度中心添加执行器，下边准备测试执行器与调度中心是否正常通信，因为接口工程依赖了service工程，所以启动媒资管理模块的接口工程。
启动后观察日志，出现下边的日志表示执行器在调度中心注册成功：
同时观察调度中心中的执行器界面：
在线机器地址处已显示1个执行器。
随后我们给执行器添加任务：
点击新增，填写任务信息：
调度类型：固定速度指按固定的间隔定时调度。
Cron，通过Cron表达式实现更丰富的定时调度策略。
Cron表达式是一个字符串，通过它可以定义调度策略，格式如下：
{秒数} {分钟} {小时} {日期} {月份} {星期} {年份(可为空)}
运行模式有BEAN和GLUE，bean模式较常用就是在项目工程中编写执行器的任务代码，GLUE是将任务代码编写在调度中心。
JobHandler即任务方法名，填写任务方法上边@XxlJob注解中的名称。
路由策略：当执行器集群部署时，调度中心向哪个执行器下发任务，这里选择第一个表示只向第一个执行器下发任务，路由策略的其它选项稍后在分片广播章节详细解释。
高级配置的其它配置项稍后下面实战详细解释。
添加成功，启动任务：
通过调度日志查看任务执行情况：
任务跑一段时间注意清理日志：
三.XXL-JOB实战例子解析：
1.高级配置：
掌握了xxl-job的基本使用，下边思考如何进行分布式任务处理呢？如下图，我们会启动多个执行器组成一个集群，去执行任务。
执行器在集群部署下调度中心有哪些路由策略呢？
查看xxl-job官方文档，阅读高级配置相关的内容：
高级配置：- 路由策略：当执行器集群部署时，提供丰富的路由策略，包括；FIRST（第一个）：固定选择第一个机器；LAST（最后一个）：固定选择最后一个机器；ROUND（轮询）：；RANDOM（随机）：随机选择在线的机器；CONSISTENT_HASH（一致性HASH）：每个任务按照Hash算法固定选择某一台机器，且所有任务均匀散列在不同机器上。LEAST_FREQUENTLY_USED（最不经常使用）：使用频率最低的机器优先被选举；LEAST_RECENTLY_USED（最近最久未使用）：最久未使用的机器优先被选举；FAILOVER（故障转移）：按照顺序依次进行心跳检测，第一个心跳检测成功的机器选定为目标执行器并发起调度；BUSYOVER（忙碌转移）：按照顺序依次进行空闲检测，第一个空闲检测成功的机器选定为目标执行器并发起调度；SHARDING_BROADCAST(分片广播)：广播触发对应集群中所有机器执行一次任务，同时系统自动传递分片参数；可根据分片参数开发分片任务；- 子任务：每个任务都拥有一个唯一的任务ID(任务ID可以从任务列表获取)，当本任务执行结束并且执行成功时，将会触发子任务ID所对应的任务的一次主动调度，通过子任务可以实现一个任务执行完成去执行另一个任务。- 调度过期策略：- 忽略：调度过期后，忽略过期的任务，从当前时间开始重新计算下次触发时间；- 立即执行一次：调度过期后，立即执行一次，并从当前时间开始重新计算下次触发时间；- 阻塞处理策略：调度过于密集执行器来不及处理时的处理策略；单机串行（默认）：调度请求进入单机执行器后，调度请求进入FIFO队列并以串行方式运行；丢弃后续调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，本次请求将会被丢弃并标记为失败；覆盖之前调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，将会终止运行中的调度任务并清空队列，然后运行本地调度任务；- 任务超时时间：支持自定义任务超时时间，任务运行超时将会主动中断任务；- 失败重试次数；支持自定义任务失败重试次数，当任务失败时将会按照预设的失败重试次数主动进行重试；下边要重点说的是分片广播策略，分片是指是调度中心以执行器为维度进行分片，将集群中的执行器标上序号：0，1，2，3...，广播是指每次调度会向集群中的所有执行器发送任务调度，请求中携带分片参数。
每个执行器收到调度请求同时接收分片参数。
xxl-job支持动态扩容执行器集群从而动态增加分片数量，当有任务量增加可以部署更多的执行器到集群中，调度中心会动态修改分片的数量。
作业分片适用哪些场景呢？
- 分片任务场景：10个执行器的集群来处理10w条数据，每台机器只需要处理1w条数据，耗时降低10倍；
- 广播任务场景：广播执行器同时运行shell脚本、广播集群节点进行缓存更新等。
所以，广播分片方式不仅可以充分发挥每个执行器的能力，并且根据分片参数可以控制任务是否执行，最终灵活控制了执行器集群分布式处理任务。
"分片广播" 和普通任务开发流程一致，不同之处在于可以获取分片参数进行分片业务处理。
BEAN、GLUE模式(Java)，可参考Sample示例执行器中的示例任务"ShardingJobHandler"：
/*** 2、分片广播任务*/
@XxlJob("shardingJobHandler")
public void shardingJobHandler() throws Exception {// 分片序号，从0开始int shardIndex = XxlJobHelper.getShardIndex();// 分片总数int shardTotal = XxlJobHelper.getShardTotal();log.info("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);log.info("开始执行第" + shardIndex + "批任务");// ....
}
添加成功：
下边启动两个执行器实例，观察每个实例的执行情况
首先在nacos中配置media-service的本地优先配置：
# 配置本地优先
spring:cloud:config:override-none: true将media-service启动两个实例
两个实例的在启动时注意端口不能冲突：
实例1 在VM options处添加：-Dserver.port=63051 -Dxxl.job.executor.port=9998
实例2 在VM options处添加：-Dserver.port=63050 -Dxxl.job.executor.port=9999
启动两个实例，观察任务调度中心，稍等片刻执行器有两个:
如果其中一个执行器挂掉，只剩下一个执行器在工作，稍等片刻调用中心发现少了一个执行器将动态调整总分片数为1。
到此作业分片任务调试完成，此时我们可以思考：
当一次分片广播到来，各执行器如何根据分片参数去分布式执行任务，保证执行器之间执行的任务不重复呢？
2.分片方案：
掌握了xxl-job的分片广播调度方式，下边思考如何分布式去执行学成在线平台中的视频处理任务。
任务添加成功后，对于要处理的任务会添加到待处理任务表中，现在启动多个执行器实例去查询这些待处理任务，此时如何保证多个执行器不会查询到重复的任务呢？
XXL-JOB并不直接提供数据处理的功能，它只会给执行器分配好分片序号，在向执行器任务调度的同时下发分片总数以及分片序号等参数，执行器收到这些参数根据自己的业务需求去利用这些参数。
下图表示了多个执行器获取视频处理任务的结构：
每个执行器收到广播任务有两个参数：分片总数、分片序号。每个执行从数据表取任务时可以让任务id 模上 分片总数，如果等于分片序号则执行此任务。
上边两个执行器实例那么分片总数为2，序号为0、1，从任务1开始，如下：
1 % 2 = 1 执行器2执行
2 % 2 = 0 执行器1执行
3 % 2 = 1 执行器2执行
以此类推.
通过作业分片方案保证了执行器之间查询到不重复的任务，如果一个执行器在处理一个视频还没有完成，此时调度中心又一次请求调度，为了不重复处理同一个视频该怎么办？
首先配置调度过期策略：
查看文档如下：
- 调度过期策略：调度中心错过调度时间的补偿处理策略，包括：忽略、立即补偿触发一次等；
- 忽略：调度过期后，忽略过期的任务，从当前时间开始重新计算下次触发时间；
- 立即执行一次：调度过期后，立即执行一次，并从当前时间开始重新计算下次触发时间；
- 阻塞处理策略：调度过于密集执行器来不及处理时的处理策略；
这里我们选择忽略，如果立即执行一次就可能重复执行相同的任务。
其次，再看阻塞处理策略，阻塞处理策略就是当前执行器正在执行任务还没有结束时调度中心进行任务调度，此时该如何处理。
查看文档如下：
单机串行（默认）：调度请求进入单机执行器后，调度请求进入FIFO队列并以串行方式运行；
丢弃后续调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，本次请求将会被丢弃并标记为失败；
覆盖之前调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，将会终止运行中的调度任务并清空队列，然后运行本地调度任务；
这里如果选择覆盖之前调度则可能重复执行任务，这里选择 丢弃后续调度或单机串行方式来避免任务重复执行。
只做这些配置可以保证任务不会重复执行吗？
做不到，还需要保证任务处理的幂等性，什么是任务的幂等性？任务的幂等性是指：对于数据的操作不论多少次，操作的结果始终是一致的。在本项目中要实现的是不论多少次任务调度同一个视频只执行一次成功的转码。
什么是幂等性？
它描述了一次和多次请求某一个资源对于资源本身应该具有同样的结果。
幂等性是为了解决重复提交问题，比如：恶意刷单，重复支付等。
解决幂等性常用的方案：
1）数据库约束，比如：唯一索引，主键。
2）乐观锁，常用于数据库，更新数据时根据乐观锁状态去更新。
3）唯一序列号，操作传递一个唯一序列号，操作时判断与该序列号相等则执行。
基于以上分析，在执行器接收调度请求去执行视频处理任务时要实现视频处理的幂等性，要有办法去判断该视频是否处理完成，如果正在处理中或处理完则不再处理。
这里我们在数据库视频处理表中添加处理状态字段，视频处理完成更新状态为完成，执行视频处理前判断状态是否完成，如果完成则不再处理。
3.业务流程：
下边梳理整个视频上传及处理的业务流程：
上传视频成功向视频处理待处理表添加记录。
视频处理的详细流程如下：
1、任务调度中心广播作业分片。
2、执行器收到广播作业分片，从数据库读取待处理任务，读取未处理及处理失败的任务。
3、执行器更新任务为处理中，根据任务内容从MinIO下载要处理的文件。
4、执行器启动多线程去处理任务。
5、任务处理完成，上传处理后的视频到MinIO。
6、将更新任务处理结果，如果视频处理完成除了更新任务处理结果以外还要将文件的访问地址更新至任务处理表及文件表中，最后将任务完成记录写入历史表。
查询待处理任务只处理未提交及处理失败的任务，任务处理失败后进行重试，最多重试3次。
任务处理成功将待处理记录移动到历史任务表。
下图是待处理任务表：
历史任务表与待处理任务表的结构相同。（读写分离）
上传视频成功向视频处理待处理表添加记录，暂时只添加对avi视频的处理记录。
根据MIME Type去判断是否是avi视频，下边列出部分MIME Type：
avi视频的MIME Type是video/x-msvideo
修改文件信息入库方法，如下：
@Transactional
public MediaFiles addMediaFilesToDb(Long companyId, String fileMd5, UploadFileParamsDto uploadFileParamsDto, String bucket, String objectName) {//从数据库查询文件MediaFiles mediaFiles = mediaFilesMapper.selectById(fileMd5);if (mediaFiles == null) {mediaFiles = new MediaFiles();//拷贝基本信息BeanUtils.copyProperties(uploadFileParamsDto, mediaFiles);mediaFiles.setId(fileMd5);mediaFiles.setFileId(fileMd5);mediaFiles.setCompanyId(companyId);//媒体类型mediaFiles.setUrl("/" + bucket + "/" + objectName);mediaFiles.setBucket(bucket);mediaFiles.setFilePath(objectName);mediaFiles.setCreateDate(LocalDateTime.now());mediaFiles.setAuditStatus("002003");mediaFiles.setStatus("1");//保存文件信息到文件表int insert = mediaFilesMapper.insert(mediaFiles);if (insert < 0) {log.error("保存文件信息到数据库失败,{}", mediaFiles.toString());XueChengPlusException.cast("保存文件信息失败");}//添加到待处理任务表addWaitingTask(mediaFiles);log.debug("保存文件信息到数据库成功,{}", mediaFiles.toString());}return mediaFiles;}/*** 添加待处理任务* @param mediaFiles 媒资文件信息*/
private void addWaitingTask(MediaFiles mediaFiles){//文件名称String filename = mediaFiles.getFilename();//文件扩展名String exension = filename.substring(filename.lastIndexOf("."));//文件mimeTypeString mimeType = getMimeType(exension);//如果是avi视频添加到视频待处理表if(mimeType.equals("video/x-msvideo")){MediaProcess mediaProcess = new MediaProcess();BeanUtils.copyProperties(mediaFiles,mediaProcess);mediaProcess.setStatus("1");//未处理mediaProcess.setFailCount(0);//失败次数默认为0mediaProcessMapper.insert(mediaProcess);}
}如何保证查询到的待处理视频记录不重复？
编写根据分片参数获取待处理任务的DAO方法，定义DAO接口如下：
@Mapper
public interface MediaProcessMapper extends BaseMapper<MediaProcess> {/*** @description 根据分片参数获取待处理任务，保证了执行器之间查询到不重复的任务* @param shardTotal 分片总数* @param shardIndex 分片序号* @param count 任务数* @return java.util.List<com.xuecheng.media.model.po.MediaProcess>* 多个执行器处理任务：id % 执行器数 = 执行器编号* status:1.未处理 3.处理失败* fail_count:三次重试机会* limit:每次只查2条数据*/@Select("select * from media_process t where t.id % #{shardTotal} = #{shardIndex} and (t.status = '1' or t.status = '3') and t.fail_count < 3 limit #{count}")List<MediaProcess> selectListByShardIndex(@Param("shardTotal") int shardTotal, @Param("shardIndex") int shardIndex, @Param("count") int count);
}
为了避免多线程去争抢同一个任务可以使用synchronized同步锁去解决，如下代码：
synchronized(锁对象){执行任务...
}而synchronized只能保证同一个虚拟机中多个线程去争抢锁。
如果是多个执行器分布式部署，并不能保证同一个视频只有一个执行器去处理。
现在要实现分布式环境下所有虚拟机中的线程去同步执行就需要让多个虚拟机去共用一个锁，虚拟机可以分布式部署，锁也可以分布式部署，如下图：
虚拟机都去抢占同一个锁，锁是一个单独的程序提供加锁、解锁服务。
该锁已不属于某个虚拟机，而是分布式部署，由多个虚拟机所共享，这种锁叫分布式锁。
实现分布式锁的方案有很多，常用的如下：
1、基于数据库实现分布锁
利用数据库主键唯一性的特点，或利用数据库唯一索引、行级锁的特点，多个线程同时去更新相同的记录，谁更新成功谁就抢到锁。
2、基于redis实现锁
redis提供了分布式锁的实现方案，比如：SETNX、set nx、redisson等。
拿SETNX举例说明，SETNX命令的工作过程是去set一个不存在的key，多个线程去设置同一个key只会有一个线程设置成功，设置成功的的线程拿到锁。
3、使用zookeeper实现
zookeeper是一个分布式协调服务，主要解决分布式程序之间的同步的问题。zookeeper的结构类似的文件目录，多线程向zookeeper创建一个子目录(节点)只会有一个创建成功，利用此特点可以实现分布式锁，谁创建该结点成功谁就获得锁。
下边基于数据库方式实现分布锁，开始执行任务将任务执行状态更新为4表示任务执行中。
下边的sql语句可以实现更新操作：
| update media_process m set m.status='4' where m.id=? |
如果是多个线程去执行该sql都将会执行成功，但需求是只能有一个线程抢到锁，所以此sql无法满足需求。
下边使用乐观锁的方式实现更新操作：
下边使用乐观锁的方式实现更新操作：
| update media_process m set m.status='4' where (m.status='1' or m.status='3') and m.fail_count<3 and m.id=? |
多个线程同时执行上边的sql只会有一个线程执行成功。
什么是乐观锁、悲观锁？
synchronized是一种悲观锁，在执行被synchronized包裹的代码时需要首先获取锁，没有拿到锁则无法执行，是总悲观的认为别的线程会去抢，所以要悲观锁。
乐观锁的思想是它不认为会有线程去争抢，尽管去执行，如果没有执行成功就再去重试。
@Mapper
public interface MediaProcessMapper extends BaseMapper<MediaProcess> {/*** 开启一个任务* @param id 任务id* @return 更新记录数* 4:处理中*/@Update("update media_process m set m.status='4' where (m.status='1' or m.status='3') and m.fail_count<3 and m.id=#{id}")int startTask(@Param("id") long id);
}
service方法：
/*** 开启一个任务* @param id 任务id* @return true开启任务成功，false开启任务失败*/
public boolean startTask(long id);// ServiceImpl实现如下
public boolean startTask(long id) {int result = mediaProcessMapper.startTask(id);return result<=0?false:true;
}任务处理完成需要更新任务处理结果，任务执行成功更新视频的URL、及任务处理结果，将待处理任务记录删除，同时向历史任务表添加记录。
package com.xuecheng.media.service.impl;import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.xuecheng.media.mapper.MediaFilesMapper;
import com.xuecheng.media.mapper.MediaProcessHistoryMapper;
import com.xuecheng.media.mapper.MediaProcessMapper;
import com.xuecheng.media.model.po.MediaFiles;
import com.xuecheng.media.model.po.MediaProcess;
import com.xuecheng.media.model.po.MediaProcessHistory;
import com.xuecheng.media.service.MediaFileProcessService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfigureOrder;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;import java.time.LocalDateTime;
import java.util.List;@Slf4j
@Service
public class MediaFileProcessServiceImpl implements MediaFileProcessService {@AutowiredMediaFilesMapper mediaFilesMapper;@AutowiredMediaProcessMapper mediaProcessMapper;@AutowiredMediaProcessHistoryMapper mediaProcessHistoryMapper;/*** @description 获取待处理任务* @param shardIndex 分片序号* @param shardTotal 分片总数* @param count 获取记录数* @return java.util.List<com.xuecheng.media.model.po.MediaProcess>*/@Overridepublic List<MediaProcess> getMediaProcessList(int shardIndex, int shardTotal, int count) {List<MediaProcess> mediaProcesses = mediaProcessMapper.selectListByShardIndex(shardTotal, shardIndex, count);return mediaProcesses;}/*** 开启一个任务* @param id 任务id* @return true开启任务成功，false开启任务失败*/@Overridepublic boolean startTask(long id) {int result = mediaProcessMapper.startTask(id);return result <= 0 ? false : true;}// 处理保存失败的结果@Transactional@Overridepublic void saveProcessFinishStatus(Long taskId, String status, String fileId, String url, String errorMsg) {// 查出任务，如果不存在则直接返回MediaProcess mediaProcess = mediaProcessMapper.selectById(taskId);if(mediaProcess == null){return ;}// 处理失败，更新任务处理结果LambdaQueryWrapper<MediaProcess> queryWrapperById = new LambdaQueryWrapper<MediaProcess>().eq(MediaProcess::getId, taskId);// 处理失败if(status.equals("3")){MediaProcess mediaProcess_u = new MediaProcess();mediaProcess_u.setStatus("3");mediaProcess_u.setErrormsg(errorMsg);mediaProcess_u.setFailCount(mediaProcess.getFailCount() + 1);mediaProcessMapper.update(mediaProcess_u,queryWrapperById);log.debug("更新任务处理状态为失败，任务信息:{}",mediaProcess_u);return ;}// 任务处理成功MediaFiles mediaFiles = mediaFilesMapper.selectById(fileId);if(mediaFiles != null){//更新媒资文件中的访问urlmediaFiles.setUrl(url);mediaFilesMapper.updateById(mediaFiles);}// 处理成功，更新url和状态mediaProcess.setUrl(url);mediaProcess.setStatus("2");mediaProcess.setFinishDate(LocalDateTime.now());mediaProcessMapper.updateById(mediaProcess);// 添加到历史记录MediaProcessHistory mediaProcessHistory = new MediaProcessHistory();BeanUtils.copyProperties(mediaProcess, mediaProcessHistory);mediaProcessHistoryMapper.insert(mediaProcessHistory);// 删除mediaProcessmediaProcessMapper.deleteById(mediaProcess.getId());}
}下面编写执行器代码：
首先根据CPU占有核心数来设置每次在数据库取出多少任务（size个），随后启动size个线程的线程池，而我们只是创建了size个线程，线程创建完成就代表videoJobHandler()方法结束：
ExecutorService threadPool = Executors.newFixedThreadPool(size);我们想要让线程待执行方法调度完成任务后在结束，所以需要加入计数器，以此确保所有视频处理逻辑执行完毕后再结束方法：
CountDownLatch countDownLatch = new CountDownLatch(size);以此就可以通过操作计数器控制：
countDownLatch.countDown(); // 计算器-1// 阻塞主线程，直到所有子线程通过countDown()通知完成，确保所有视频处理逻辑执行完毕后再退出任务。 // 等待,给一个充裕的超时时间,防止无限等待，到达超时时间30min还没有处理完成则结束任务 countDownLatch.await(30, TimeUnit.MINUTES);
countDownLatch.await(30, TimeUnit.MINUTES)设置的 30分钟超时 是指整个videoJobHandler方法的 最大执行时间。这30分钟是从
countDownLatch.await()调用开始计时，如果30分钟内所有子线程未完成（即计数器未减到0），主线程会继续执行，方法退出，未完成的任务会继续在后台线程池中运行，如果XXL-JOB管理界面也配置了任务超时时间（如20分钟），那么XXL-JOB会认为本次调度已超时。
下面是分析线程处理方法：
首先使用方法对每个查询出来的任务进行线程调度：
// 计数器,协调并发线程生命周期,确保所有视频处理逻辑执行完毕后再退出任务 CountDownLatch countDownLatch = new CountDownLatch(size); // 将处理任务加入线程池 mediaProcessList.forEach(mediaProcess -> {threadPool.execute(() -> {try{// 业务逻辑...} finally {countDownLatch.countDown(); // 计算器-1} }); // 阻塞主线程，直到所有子线程通过countDown()通知完成，确保所有视频处理逻辑执行完毕后再退出任务。// 等待,给一个充裕的超时时间,防止无限等待，到达超时时间30min还没有处理完成则结束任务countDownLatch.await(30, TimeUnit.MINUTES); }在任务里面就需要抢占任务，随后将要处理的文件下载到服务器上，处理结束的视频文件，随后在将处理完成的mp4上传到MinIO。
package com.xuecheng.media.service.jobhandler;import com.xuecheng.base.utils.Mp4VideoUtil;
import com.xuecheng.media.model.po.MediaProcess;
import com.xuecheng.media.service.MediaFileProcessService;
import com.xuecheng.media.service.MediaFileService;
import com.xuecheng.media.service.MediaFileTransactionalService;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.*;@Slf4j
@Component
public class VideoTask {@AutowiredMediaFileService mediaFileService;@Autowiredprivate MediaFileTransactionalService mediaFileTransactionalService;@AutowiredMediaFileProcessService mediaFileProcessService;@Value("${videoprocess.ffmpegpath}")String ffmpegpath;@XxlJob("videoJobHandler")public void videoJobHandler() throws Exception {// 分片参数int shardIndex = XxlJobHelper.getShardIndex();int shardTotal = XxlJobHelper.getShardTotal();List<MediaProcess> mediaProcessList = null;int size = 0;try {// 取出cpu核心数作为一次处理数据的条数int processors = Runtime.getRuntime().availableProcessors();// 一次处理视频数量不要超过cpu核心数mediaProcessList = mediaFileProcessService.getMediaProcessList(shardIndex, shardTotal, processors);size = mediaProcessList.size();log.debug("取出待处理视频任务{}条", size);if (size < 0) {return;}} catch (Exception e) {e.printStackTrace();return;}// 启动size个线程的线程池ExecutorService threadPool = Executors.newFixedThreadPool(size);// 计数器,协调并发线程生命周期,确保所有视频处理逻辑执行完毕后再退出任务// 因为下面代码只是创建了size个线程，线程创建完成就代表videoJobHandler()方法结束，而我们想让方法调度完成任务后在结束，所以需要加入计数器CountDownLatch countDownLatch = new CountDownLatch(size);// 将处理任务加入线程池mediaProcessList.forEach(mediaProcess -> {threadPool.execute(() -> {try {// 任务idLong taskId = mediaProcess.getId();// 抢占任务boolean b = mediaFileProcessService.startTask(taskId);if (!b) {return;}log.debug("开始执行任务:{}", mediaProcess);// 下边是处理逻辑// 桶String bucket = mediaProcess.getBucket();// 存储路径String filePath = mediaProcess.getFilePath();// 原始视频的md5值String fileId = mediaProcess.getFileId();// 原始文件名称String filename = mediaProcess.getFilename();// 将要处理的文件下载到服务器上File originalFile = mediaFileService.downloadFileFromMinIO(mediaProcess.getBucket(), mediaProcess.getFilePath());if (originalFile == null) {log.debug("下载待处理文件失败,originalFile:{}", mediaProcess.getBucket().concat(mediaProcess.getFilePath()));// 保存任务处理失败的结果mediaFileProcessService.saveProcessFinishStatus(mediaProcess.getId(), "3", fileId, null, "下载待处理文件失败");return;}// 处理结束的视频文件File mp4File = null;try {mp4File = File.createTempFile("mp4", ".mp4");} catch (IOException e) {log.error("创建mp4临时文件失败");mediaFileProcessService.saveProcessFinishStatus(mediaProcess.getId(), "3", fileId, null, "创建mp4临时文件失败");return;}// 视频处理结果String result = "";try {// 开始处理视频Mp4VideoUtil videoUtil = new Mp4VideoUtil(ffmpegpath, originalFile.getAbsolutePath(), mp4File.getName(), mp4File.getAbsolutePath());// 开始视频转换，成功将返回successresult = videoUtil.generateMp4();} catch (Exception e) {e.printStackTrace();log.error("处理视频文件:{},出错:{}", mediaProcess.getFilePath(), e.getMessage());}if (!result.equals("success")) {// 记录错误信息log.error("处理视频失败,视频地址:{},错误信息:{}", bucket + filePath, result);mediaFileProcessService.saveProcessFinishStatus(mediaProcess.getId(), "3", fileId, null, result);return;}// 将mp4上传至minio// mp4在minio的存储路径String objectName = getFilePath(fileId, ".mp4");// 访问urlString url = "/" + bucket + "/" + objectName;try {mediaFileService.addMediaFilesToMinIO(mp4File.getAbsolutePath(), "video/mp4", bucket, objectName);// 将url存储至数据，并更新状态为成功，并将待处理视频记录删除存入历史mediaFileProcessService.saveProcessFinishStatus(mediaProcess.getId(), "2", fileId, url, null);} catch (Exception e) {log.error("上传视频失败或入库失败,视频地址:{},错误信息:{}", bucket + objectName, e.getMessage());// 最终还是失败了mediaFileProcessService.saveProcessFinishStatus(mediaProcess.getId(), "3", fileId, null, "处理后视频上传或入库失败");}}finally {countDownLatch.countDown(); // 计算器-1}});});// 阻塞主线程，直到所有子线程通过countDown()通知完成，确保所有视频处理逻辑执行完毕后再退出任务。// 等待,给一个充裕的超时时间,防止无限等待，到达超时时间30min还没有处理完成则结束任务countDownLatch.await(30, TimeUnit.MINUTES);}private String getFilePath(String fileMd5,String fileExt){return fileMd5.substring(0,1) + "/" + fileMd5.substring(1,2) + "/" + fileMd5 + "/" +fileMd5 +fileExt;}}进入xxl-job调度中心添加执行器和视频处理任务
在xxl-job配置任务调度策略：
1）配置阻塞处理策略为：丢弃后续调度。
2）配置视频处理调度时间间隔不用根据视频处理时间去确定，可以配置的小一些，如：5分钟，即使到达调度时间如果视频没有处理完会丢弃调度请求。
配置完成开始测试视频处理：
1、首先上传至少4个视频，非mp4格式。
2、在xxl-job启动视频处理任务
3、观察媒资管理服务后台日志
4.抢占任务测试：
修改调度中心中视频处理任务的阻塞处理策略为“覆盖之间的调度”：
在抢占任务代码处打断点并选择支持多线程方式：
在抢占任务代码处的下边两行代码分别打上断点，避免观察时代码继续执行。启动任务。