【网课平台】Day5.XXL-Job实现分布式任务调度

ゝ一世哀愁。 2023-10-08 23:50 16阅读 0赞

#### 文章目录 ####

*  需求: 视频处理
 *  1、知识背景
 *   *  1.1 文件格式与编码格式
     *  1.2 FFmpeg
     *  1.3 视频处理工具类
 *  2、分布式任务处理
 *   *  2.1 分布式任务调度
     *  2.2 XXL-JOB
     *  2.3 搭建XXL-Job
     *  2.4 分片广播
 *  3、技术方案
 *  4、查询待处理任务
 *   *  4.1 需求分析
     *  4.2 表设计
     *  4.3 添加待处理任务
     *  4.4 查询待处理任务
 *  5、任务执行
 *   *  5.1 分布式锁
     *  5.2 开启任务
     *  5.3 更新任务状态
     *  5.4 视频处理

## 需求: 视频处理 ##

视频上传成功需要对视频的格式进行转码处理，比如：avi转成mp

## 1、知识背景 ##

### 1.1 文件格式与编码格式 ###

视频文件的内容主要包括视频和音频，其文件格式是按照一 定的编码格式去编码，并且按照该文件所规定的封装格式将视频、音频、字幕等信息封装在一起，播放器会根据它们的封装格式去提取出编码，然后由播放器解码，最终播放音视频。

*  .mp4、.avi、.rmvb等 这些视频文件的不同扩展名, 称为文件格式
 *  编码格式: 通过音视频的压缩技术，将视频格式转换成另一种视频格式，通过视频编码实现流媒体的传输。比如：一个.avi的视频文件原来的编码是a，通过编码后编码格式变为b，音频原来为c，通过编码后变为d

音视频编码格式种类繁多, 目前最常用的编码标准是视频H.264，音频AAC。

### 1.2 FFmpeg ###

下载地址:
    https://www.ffmpeg.org/download.html#build-windows

下载后解压ffmpeg.zip, 在.exe文件的目录下打开cmd窗口, 运行 ffmpeg -v  
![在这里插入图片描述][c7ed6c9919364e9a88daa2ac095e95c7.png]  
以上即安装成功

//对1.avi文件进行转码:
    //先把avi转mp4
    D:\soft\ffmpeg\ffmpeg.exe -i 1.avi 1.mp4

//转mp3
    ffmpeg -i 1.avi 1.mp3
    //转gif
    ffmpeg -i 1.avi 1.gif

### 1.3 视频处理工具类 ###

要通过ffmpeg对视频转码，得用Java程序去启动ffmpeg，使用`java.lang.ProcessBuilder`去完成

//这里以启动QQ为例
    ProcessBuilder builder = new ProcessBuilder();
    builder.command("C:\\Program Files (x86)\\Tencent\\QQ\\Bin\\QQScLauncher.exe");
    //将标准输入流和错误输入流合并，通过标准输入流程读取信息
    builder.redirectErrorStream(true);
    Process p = builder.start();

对视频的转码, 直接使用工具类:

Java
    public static void main(String[] args) throws IOException {
        
        //ffmpeg的路径
        String ffmpeg_path = "D:\\soft\\ffmpeg\\ffmpeg.exe";//ffmpeg的安装位置
        //源avi视频的路径
        String video_path = "D:\\develop\\bigfile_test\\nacos01.avi";
        //转换后mp4文件的名称
        String mp4_name = "nacos01.mp4";
        //转换后mp4文件的路径
        String mp4_path = "D:\\develop\\bigfile_test\\nacos01.mp4";
        //创建工具类对象
        Mp4VideoUtil videoUtil = new Mp4VideoUtil(ffmpeg_path,video_path,mp4_name,mp4_path);
        //开始视频转换，成功将返回success
        //videoUtil工具类的generateMp4方法封装了对ffmpeg的启动以及转码所用到的语句
        String s = videoUtil.generateMp4();
        System.out.println(s);
    }

generateMp4方法源代码如下:

public String generateMp4(){
        
            //清除已生成的mp4
    //        clear_mp4(mp4folder_path+mp4_name);
            clear_mp4(mp4folder_path);
            /*
            ffmpeg.exe -i  lucene.avi -c:v libx264 -s 1280x720 -pix_fmt yuv420p -b:a 63k -b:v 753k -r 18 .\lucene.mp4
             */
            List<String> commend = new ArrayList<String>();
            //commend.add("D:\\Program Files\\ffmpeg-20180227-fa0c9d6-win64-static\\bin\\ffmpeg.exe");
            commend.add(ffmpeg_path);
            commend.add("-i");
    //        commend.add("D:\\BaiduNetdiskDownload\\test1.avi");
            commend.add(video_path);
            commend.add("-c:v");
            commend.add("libx264");
            commend.add("-y");//覆盖输出文件
            commend.add("-s");
            commend.add("1280x720");
            commend.add("-pix_fmt");
            commend.add("yuv420p");
            commend.add("-b:a");
            commend.add("63k");
            commend.add("-b:v");
            commend.add("753k");
            commend.add("-r");
            commend.add("18");
    //        commend.add(mp4folder_path  + mp4_name );
            commend.add(mp4folder_path  );
            String outstring = null;
            try {
        
                ProcessBuilder builder = new ProcessBuilder();
                builder.command(commend);
                //将标准输入流和错误输入流合并，通过标准输入流程读取信息
                builder.redirectErrorStream(true);
                Process p = builder.start();
                outstring = waitFor(p);
     
            } catch (Exception ex) {
        
     
                ex.printStackTrace();
     
            }
    //        Boolean check_video_time = this.check_video_time(video_path, mp4folder_path + mp4_name);
            Boolean check_video_time = this.check_video_time(video_path, mp4folder_path);
            if(!check_video_time){
        
                return outstring;
            }else{
        
                return "success";
            }
        }

## 2、分布式任务处理 ##

### 2.1 分布式任务调度 ###

想高效的去处理一批任务，有以下两种思路：

*  多线程：充分利用单机资源
 *  `分布式`加多线程：`充分利用多台计算机`，每台计算机使用多线程处理

关于任务调度，思考以下场景：

*  每隔24小时执行数据备份任务
 *  12306网站会根据车次不同，设置几个时间点分批次放票
 *  某财务系统需要在每天上午10点前结算前一天的账单数据，统计汇总
 *  商品成功发货后，需要向客户发送短信提醒

以上场景的实现思路有：

> 思路一：线程休眠

public static void main(String[] args) {
            
        //任务执行间隔时间
        final long timeInterval = 1000;
        Runnable runnable = new Runnable() {
        
            public void run() {
        
                while (true) {
        
                    //TODO：something
                    try {
        
                        Thread.sleep(timeInterval);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
            }
        };
        Thread thread = new Thread(runnable);
        thread.start();
    }

> 思路2：Timer

Jdk也提供了相关支持，如Timer、ScheduledExecutor

public static void main(String[] args){
          
        Timer timer = new Timer();  
        timer.schedule(new TimerTask(){
        
            @Override  
            public void run() {
          
               //TODO：something
            }  
        }, 1000, 2000);  //1秒后开始调度，每2秒执行一次
    }

每个Timer对应一个线程，因此可以同时启动多个Timer并行执行多个任务，同一个Timer中的任务是串行执行

> 思路3：ScheduledExecutor

基于线程池设计的 ScheduledExecutor，每一个被调度的任务都会由线程池中一个线程去执行，因此任务是并发执行的，相互之间不会受到干扰

public static void main(String [] agrs){
        
        ScheduledExecutorService service = Executors.newScheduledThreadPool(10);
        service.scheduleAtFixedRate(
                new Runnable() {
        
                    @Override
                    public void run() {
        
                        //TODO：something
                        System.out.println("todo something");
                    }
                }, 1,
                2, TimeUnit.SECONDS);
    }

以上三种能完成简单的任务，如给定开始时间与重复间隔，重复执行某任务，但比较复杂的实现不了，如：设置每月第一天凌晨1点执行任务、复杂调度任务的管理、任务间传递数据等等

> 任务的调度，即指系统为了完成特定业务，基于给定时间点，给定时间间隔或者给定执行次数，去自动执行任务

![在这里插入图片描述][cf1017cd9cad4ccaa0faceeabca951f8.png]  
而分布式任务调度，即多台计算机共同去完成任务调度。从而可以：

*  突破单机多线程的瓶颈，并行任务调度
 *  高可用
 *  弹性扩容

此外，需要注意任务管理与监测，以及防止任务被重复执行。

### 2.2 XXL-JOB ###

XXL-JOB是一个分布式任务调度平台，主要有调度中心和执行器。官网：[https://www.xuxueli.com/xxl-job/][https_www.xuxueli.com_xxl-job]  
![在这里插入图片描述][8f9dc76752bd4e719f6f01fc829461ca.png]

调度中心和执行器之间的工作流程：

![在这里插入图片描述][868bebf9d8f5463393ad6f74cb56a9de.png]

*  `任务执行器根据配置的调度中心的地址，自动注册到调度中心`
 *  `当达到任务的触发条件时，调度中心下发任务给任务执行器`
 *  `执行器基于线程池执行任务，并把执行结果放入内存队列中、把执行日志写入日志文件中`
 *  执行器消费内存队列中的执行结果，主动上报给调度中心
 *  当用户在调度中心查看任务日志，调度中心请求任务执行器，任务执行器读取任务日志文件并返回日志详情

### 2.3 搭建XXL-Job ###

> **`调度中心部分`**

*  下载源码解压并用IDEA打开项目

GitHub：https://github.com/xuxueli/xxl-job
    码云：https://gitee.com/xuxueli0323/xxl-job
    eg: https://github.com/xuxueli/xxl-job/releases/tag/2.3.1

*  项目代码结构介绍  
    ![在这里插入图片描述][84136ffe30394319bc6dfd5e091f3b09.png]

xxl-job-admin：调度中心
    xxl-job-core：公共依赖
    xxl-job-executor-samples：执行器Sample示例（选择合适的版本执行器，可直接使用）
        ：xxl-job-executor-sample-springboot：Springboot版本，通过Springboot管理执行器，推荐这种方式；
        ：xxl-job-executor-sample-frameless：无框架版本；
    doc :文档资料，包含数据库脚本

*  执行doc文件中的SQL脚本，创建数据库表  
    ![在这里插入图片描述][4a66500c5ded45d3a8c89cf9df331fbe.png]
 *  启动调度中心

虚拟机执行：sh /data/soft/restart.sh自动启动xxl-job调度中心
    访问：http://localhost:8088/xxl-job-admin/
    账号和密码：admin/123456
    
    也可直接在IDEA中启动xxl-job-admin模块

![在这里插入图片描述][1cf74e6be59e40cc9c0e6e2dc462fcb6.png]

> **`执行器部分`**

*  进入调度中心，添加执行器  
    ![在这里插入图片描述][110a598f51f843ac9e9f3641f029c31c.png]
 *  填写执行器信息，appname是前边在nacos中配置xxl信息时指定的执行器的应用名  
    ![在这里插入图片描述][ba178f0ede5f493fb187bc290ba644b1.png]
 *  在需要跑定时任务的模块加入依赖，在项目的父工程已约定了版本2.3.1

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

*  在nacos下对应服务的xxx-service-dev.yaml下配置xxl-job

xxl:
      job:
        admin: 
          # 调度中心地址
          addresses: http://localhost:8088/xxl-job-admin
        executor:
          # 创建执行器时的appname
          appname: media-process-service
          address: 
          ip: 
          # port是执行器启动的端口，如果本地启动多个执行器注意端口不能重复
          port: 9999
          logpath: /data/applogs/xxl-job/jobhandler
          logretentiondays: 30
        accessToken: default_token

*  将xxl-job配置类拷贝到相应的工程目录下  
    ![在这里插入图片描述][0467114e86aa499cb79d06ead0a2e0a3.png]

> **`执行任务`**

*  在xxxservice包下新建jobhandler存放任务类

/**
     * @description 测试执行器
     */
     @Component
     @Slf4j
    public class SampleJob {
        
    
     /**
      * 1、简单任务示例（Bean模式）
      */
     @XxlJob("testJob")
     public void testJob() throws Exception {
        
      //这里写调度任务的实现代码
      log.info("开始执行.....");
    
     }
    
    }

*  在调度中心选择执行器后，点击添加任务，进入任务管理  
    ![在这里插入图片描述][bf751711ecb5433393092a9419cabe94.jpeg]
 *  填写任务信息  
    ![在这里插入图片描述][53a948271e0946979eaf071b02c82c8e.png]

**调度类型：**

- 固定速度：指按固定的间隔定时调度。
    - Cron：通过Cron表达式实现更丰富的定时调度策略。

cron表达式的格式为：`{秒数} {分钟} {小时} {日期} {月份} {星期} {年份(可为空)}`

举例：
    
    30 10 1 * * ?  每天1点10分30秒触发
    0/30 * * * * ? 每30秒触发一次
    * 0/10 * * * ? 每10分钟触发一次

![在这里插入图片描述][7526cbb653ac4178910c24dd661450c4.png]

**运行模式**：

- BEAN：bean模式在项目工程中编写执行器的任务代码
    - GLUE：GLUE是将任务代码编写在调度中心

**JobHandler即任务方法名，填写任务方法上边@XxlJob注解中的名称**

*  信息填写完成后点击启动  
    ![在这里插入图片描述][df7370cb7cc841da84821c5fa81c3c0e.png]
 *  最后，可停止任务、手动执行一次任务、清理日志  
    ![在这里插入图片描述][975da0eda949450da714286fd0142700.png]  
    ![在这里插入图片描述][debb0b1ca985452c84127635b8a9d250.png]

### 2.4 分片广播 ###

启动多个执行器，组成集群去执行任务，此时调度中心如何去调度这些执行器？  
![在这里插入图片描述][95584a9d985240c38379ec4a645163aa.png]

当执行器集群部署时，xxl-job可选的路由策略包括：

FIRST（第一个）: 固定选择第一个机器；
    LAST（最后一个）: 固定选择最后一个机器；
    ROUND（轮询）: 轮流分配
    RANDOM（随机）: 随机选择在线的机器；
    CONSISTENT_HASH（一致性HASH）: 每个任务按照Hash算法固定选择某一台机器，且所有任务均匀散列在不同机器上。
    LEAST_FREQUENTLY_USED（最不经常使用）: 使用频率最低的机器优先被选举；
    LEAST_RECENTLY_USED（最近最久未使用）: 最久未使用的机器优先被选举；
    FAILOVER（故障转移）: 按照顺序依次进行心跳检测，第一个心跳检测成功的机器选定为目标执行器并发起调度；
    BUSYOVER（忙碌转移）: 按照顺序依次进行空闲检测，第一个空闲检测成功的机器选定为目标执行器并发起调度；
    SHARDING_BROADCAST(分片广播): 广播触发对应集群中所有机器执行一次任务，同时系统自动传递分片参数；可根据分片参数开发分片任务；

当过了任务的执行时间，而又收到了请求时，调度过期策略有：

- 忽略: 调度过期后，忽略过期的任务，从当前时间开始重新计算下次触发时间；
    - 立即执行一次: 调度过期后，立即执行一次，并从当前时间开始重新计算下次触发时间；

调度过于密集执行器来不及处理时的处理策略–阻塞处理策略有：

单机串行（默认）: 调度请求进入单机执行器后，调度请求进入FIFO队列并以串行方式运行；
    丢弃后续调度: 调度请求进入单机执行器后，发现执行器存在运行的调度任务，本次请求将会被丢弃并标记为失败；
    覆盖之前调度: 调度请求进入单机执行器后，发现执行器存在运行的调度任务，将会终止运行中的调度任务并清空队列，然后运行本地调度任务；

而分片广播策略，分片，即将集群中的执行器标上序号，广播是指每次调度会向集群中的所有执行器发送任务调度，并且**请求中携带分片参数**（序号，总的执行器数量）  
![在这里插入图片描述][452eaaf703ca457591a81316e17dba1c.png]

*  定义分片的任务方法：

@XxlJob("shardingJobHandler")
     public void shardingJobHandler() throws Exception {
        
    
      // 分片参数
      int shardIndex = XxlJobHelper.getShardIndex();
      int shardTotal = XxlJobHelper.getShardTotal();
    
    log.info("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);

*  在调度中心添加任务  
    ![在这里插入图片描述][75b09b6c0e7c415894612e21eb7bdd02.png]  
    ![在这里插入图片描述][4b1366b3e0bc4d3ca8ebe381ff2279ce.png]
 *  启动任务，查看控制台日志  
    ![在这里插入图片描述][ac29658e0d8b42e6b77d5712492f0bba.png]

接下来启动两个执行器实例：

*  将媒资服务media-service启动两个实例：

实例1 在VM options处添加：-Dserver.port=63051 -Dxxl.job.executor.port=9998
    实例2 在VM options处添加：-Dserver.port=63050 -Dxxl.job.executor.port=9999

![在这里插入图片描述][bf49663a60074d3db8f8901a934b5476.png]

*  注意在nacos中开启本地配置优先

#配置本地优先
    spring:
     cloud:
      config:
        override-none: true

*  看到调度中心有两个执行器  
    ![在这里插入图片描述][ed31f4bf021e439cb9fd0a9b3c1a5679.png]  
    ![在这里插入图片描述][65a21930ffbb42eead6da752f3a9c11f.png]

## 3、技术方案 ##

XXL-JOB只会给执行器分配好分片序号，在向执行器任务调度的同时下发分片总数以及分片序号等参数。将要处理的任务会添加到待处理任务表中，多个执行器如何保证拿到的任务不重复？

![在这里插入图片描述][38d64c5614d24c8db1c01f6cd6894bd1.png]  
每个执行器收到广播任务有两个参数：分片总数、分片序号。让取出来的任务id模上分片总数，结果等于几，就让几号执行器处理：

1  %  2 = 1    执行器2执行
    2  %  2 =  0    执行器1执行
    3  %  2 =  1     执行器2执行
    .....

配置过期调度策略：

![在这里插入图片描述][6e5d4ac6192d4c24aaf9ce02680b1346.png]  
此时，任务还是有可能被重复执行，因此需要保证任务的[幂等性][Link 1]。视频处理的业务流程图如下：

![在这里插入图片描述][fd6b485b7ac34e2e9167c0708a01a0d3.png]

画出时序图：

![在这里插入图片描述][d0c767c0de174a6caf5020bb3e29ba08.png]

## 4、查询待处理任务 ##

### 4.1 需求分析 ###

查询待处理任务只处理未提交及处理失败的任务，任务处理失败后进行重试，最多重试3次。

### 4.2 表设计 ###

待处理任务表：

![在这里插入图片描述][741f6cb808044bd199a7b059f8ffdd8f.png]  
`考虑到任务表要常做查询，而处理完的数据如果一直堆积，数据量会猛增，因此，引入历史记录表，将处理成功的记录从任务表移动到历史任务表，两张表结构相同`。

### 4.3 添加待处理任务 ###

判断如果是avi视频，需要入库任务表：

/**
     * 添加待处理任务
     * @param mediaFiles 媒资文件信息
     */
    private void addWaitingTask(MediaFiles mediaFiles){
        
        //文件名称
        String filename = mediaFiles.getFilename();
        //文件扩展名
        String exension = filename.substring(filename.lastIndexOf("."));
        //文件mimeType
        String mimeType = getMimeType(exension);
        
        //如果是avi视频添加到视频待处理表
        if(mimeType.equals("video/x-msvideo")){
        
            MediaProcess mediaProcess = new MediaProcess();
            BeanUtils.copyProperties(mediaFiles,mediaProcess);
            mediaProcess.setStatus("1");//未处理
            mediaProcess.setFailCount(0);//失败次数默认为0
            mediaProcessMapper.insert(mediaProcess);
        }
    }

修改文件信息入库的方法，在方法内部调用上面的添加待处理任务的方法

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

### 4.4 查询待处理任务 ###

定义Mapper层接口：

public interface MediaProcessMapper extends BaseMapper<MediaProcess> {
        
        /**
         * @description 根据分片参数获取待处理任务
         * @param shardTotal  分片总数
         * @param shardindex  分片序号
         * @param count 任务数
        */
        @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);
    }

定义Service层接口：

public interface MediaFileProcessService {
        
    
        /**
         * @description 获取待处理任务
         * @param shardIndex 分片序号
         * @param shardTotal 分片总数
         * @param count 获取记录数
        */
        public List<MediaProcess> getMediaProcessList(int shardIndex,int shardTotal,int count);
    
    
    }

实现类：

@Slf4j
    @Service
    public class MediaFileProcessServiceImpl implements MediaFileProcessService {
        
    
     @Autowired
     MediaFilesMapper mediaFilesMapper;
    
     @Autowired
     MediaProcessMapper mediaProcessMapper;
    
    
     @Override
     public List<MediaProcess> getMediaProcessList(int shardIndex, int shardTotal, int count) {
        
      List<MediaProcess> mediaProcesses = mediaProcessMapper.selectListByShardIndex(shardTotal, shardIndex, count);
       return mediaProcesses;
     }
    
    
    }

## 5、任务执行 ##

### 5.1 分布式锁 ###

synchronized(锁对象){
        
       执行任务...
    }

synchronized只能保证同一个虚拟机中多个线程去争抢锁。如果是多个执行器分布式部署，多个虚拟机共用一个锁，该锁已不属于某个虚拟机，而是分布式部署，由多个虚拟机所共享，这种锁叫分布式锁。

![在这里插入图片描述][05c236a9eba6450bb708b8b336aab5fa.png]  
实现分布式锁可以：

*  基于数据库
 *  基于Redis
 *  基于zookeeper

### 5.2 开启任务 ###

基于数据库实现，基本SQL是：

update media_process m set m.status='4' where  m.id=?

引入乐观锁：

update media_process m 
    set m.status='4' 
    where (m.status='1' or m.status='3') and m.fail_count<3 and m.id=?

`这里的status字段和乐观锁常用的version一样的效果，不用再额外加version`。

定义Mapper层：

public interface MediaProcessMapper extends BaseMapper<MediaProcess> {
        
    
        /**
         * 开启一个任务
         * @param id 任务id
         * @return 更新记录数
         */
        @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);
    
    //实现如下
    public boolean startTask(long id) {
        
        int result = mediaProcessMapper.startTask(id);
        return result<=0?false:true;
    }

### 5.3 更新任务状态 ###

任务处理完成需要更新任务处理结果，逻辑是：

*  任务执行成功更新视频的URL、及任务处理结果
 *  将待处理任务记录删除，同时向历史任务表添加记录

Service层接口定义：（`传参先写你能确定的，后续写实现类，缺什么传参再回来加`）

/**
     * @description 保存任务结果
     * @param taskId  任务id
     * @param status 任务状态
     * @param fileId  文件id
     * @param url url
     * @param errorMsg 错误信息
     * @return void
     */
    void saveProcessFinishStatus(Long taskId,String status,String fileId,String url,String errorMsg);

实现类：

@Slf4j
    @Service
    public class MediaFileProcessServiceImpl implements MediaFileProcessService {
        
    
     @Autowired
     MediaFilesMapper mediaFilesMapper;
    
     @Autowired
     MediaProcessMapper mediaProcessMapper;
    
     @Autowired
     MediaProcessHistoryMapper mediaProcessHistoryMapper;
    
    
    
    @Transactional
    @Override
    public 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){
        
            //更新媒资文件中的访问url
            mediaFiles.setUrl(url);
            mediaFilesMapper.updateById(mediaFiles);
        }
        //处理成功，更新url和状态
        mediaProcess.setUrl(url);
        mediaProcess.setStatus("2");
        mediaProcess.setFinishDate(LocalDateTime.now());
        mediaProcessMapper.updateById(mediaProcess);
    
        //添加到历史记录
        //这里直接new同结构表的对象后copyProperties
        MediaProcessHistory mediaProcessHistory = new MediaProcessHistory();
        BeanUtils.copyProperties(mediaProcess, mediaProcessHistory);
        mediaProcessHistoryMapper.insert(mediaProcessHistory);
        //删除mediaProcess
        mediaProcessMapper.deleteById(mediaProcess.getId());
    
    }
    
     @Override
     public List<MediaProcess> getMediaProcessList(int shardIndex, int shardTotal, int count) {
        
      List<MediaProcess> mediaProcesses = mediaProcessMapper.selectListByShardIndex(shardTotal, shardIndex, count);
       return mediaProcesses;
     }
    
    
    }

### 5.4 视频处理 ###

视频采用并发处理，每个视频使用一个线程去处理，每次处理的视频数量不要超过cpu核心数。

所有视频处理完成结束本次执行，为防止代码异常出现无限期等待则添加超时设置，到达超时时间还没有处理完成仍结束任务。

@Slf4j
    @Component
    public class VideoTask {
        
    
        @Autowired
        MediaFileService mediaFileService;
        @Autowired
        MediaFileProcessService 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);
        //计数器
        CountDownLatch countDownLatch = new CountDownLatch(size);
        //将处理任务加入线程池
        mediaProcessList.forEach(mediaProcess -> {
        
            threadPool.execute(() -> {
        
                try {
        
                    //任务id
                    Long 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());
                        //开始视频转换，成功将返回success
                        result = 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");
                    //访问url
                    String 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();
                }
            });
        });
        //等待,给一个充裕的超时时间,防止无限等待，到达超时时间还没有处理完成则结束任务
        countDownLatch.await(30, TimeUnit.MINUTES);
        }
    
        private String getFilePath(String fileMd5,String fileExt){
        
            return   fileMd5.substring(0,1) + "/" + fileMd5.substring(1,2) + "/" + fileMd5 + "/" +fileMd5 +fileExt;
        }
    
    }

`写实现类，逻辑复杂时先用注释写出你每一步要干啥，再用代码翻译`。

[c7ed6c9919364e9a88daa2ac095e95c7.png]: https://img-blog.csdnimg.cn/c7ed6c9919364e9a88daa2ac095e95c7.png
[cf1017cd9cad4ccaa0faceeabca951f8.png]: https://img-blog.csdnimg.cn/cf1017cd9cad4ccaa0faceeabca951f8.png
[https_www.xuxueli.com_xxl-job]: https://www.xuxueli.com/xxl-job/
[8f9dc76752bd4e719f6f01fc829461ca.png]: https://img-blog.csdnimg.cn/8f9dc76752bd4e719f6f01fc829461ca.png
[868bebf9d8f5463393ad6f74cb56a9de.png]: https://img-blog.csdnimg.cn/868bebf9d8f5463393ad6f74cb56a9de.png
[84136ffe30394319bc6dfd5e091f3b09.png]: https://img-blog.csdnimg.cn/84136ffe30394319bc6dfd5e091f3b09.png
[4a66500c5ded45d3a8c89cf9df331fbe.png]: https://img-blog.csdnimg.cn/4a66500c5ded45d3a8c89cf9df331fbe.png
[1cf74e6be59e40cc9c0e6e2dc462fcb6.png]: https://img-blog.csdnimg.cn/1cf74e6be59e40cc9c0e6e2dc462fcb6.png
[110a598f51f843ac9e9f3641f029c31c.png]: https://img-blog.csdnimg.cn/110a598f51f843ac9e9f3641f029c31c.png
[ba178f0ede5f493fb187bc290ba644b1.png]: https://img-blog.csdnimg.cn/ba178f0ede5f493fb187bc290ba644b1.png
[0467114e86aa499cb79d06ead0a2e0a3.png]: https://img-blog.csdnimg.cn/0467114e86aa499cb79d06ead0a2e0a3.png
[bf751711ecb5433393092a9419cabe94.jpeg]: https://img-blog.csdnimg.cn/bf751711ecb5433393092a9419cabe94.jpeg
[53a948271e0946979eaf071b02c82c8e.png]: https://img-blog.csdnimg.cn/53a948271e0946979eaf071b02c82c8e.png
[7526cbb653ac4178910c24dd661450c4.png]: https://img-blog.csdnimg.cn/7526cbb653ac4178910c24dd661450c4.png
[df7370cb7cc841da84821c5fa81c3c0e.png]: https://img-blog.csdnimg.cn/df7370cb7cc841da84821c5fa81c3c0e.png
[975da0eda949450da714286fd0142700.png]: https://img-blog.csdnimg.cn/975da0eda949450da714286fd0142700.png
[debb0b1ca985452c84127635b8a9d250.png]: https://img-blog.csdnimg.cn/debb0b1ca985452c84127635b8a9d250.png
[95584a9d985240c38379ec4a645163aa.png]: https://img-blog.csdnimg.cn/95584a9d985240c38379ec4a645163aa.png
[452eaaf703ca457591a81316e17dba1c.png]: https://img-blog.csdnimg.cn/452eaaf703ca457591a81316e17dba1c.png
[75b09b6c0e7c415894612e21eb7bdd02.png]: https://img-blog.csdnimg.cn/75b09b6c0e7c415894612e21eb7bdd02.png
[4b1366b3e0bc4d3ca8ebe381ff2279ce.png]: https://img-blog.csdnimg.cn/4b1366b3e0bc4d3ca8ebe381ff2279ce.png
[ac29658e0d8b42e6b77d5712492f0bba.png]: https://img-blog.csdnimg.cn/ac29658e0d8b42e6b77d5712492f0bba.png
[bf49663a60074d3db8f8901a934b5476.png]: https://img-blog.csdnimg.cn/bf49663a60074d3db8f8901a934b5476.png
[ed31f4bf021e439cb9fd0a9b3c1a5679.png]: https://img-blog.csdnimg.cn/ed31f4bf021e439cb9fd0a9b3c1a5679.png
[65a21930ffbb42eead6da752f3a9c11f.png]: https://img-blog.csdnimg.cn/65a21930ffbb42eead6da752f3a9c11f.png
[38d64c5614d24c8db1c01f6cd6894bd1.png]: https://img-blog.csdnimg.cn/38d64c5614d24c8db1c01f6cd6894bd1.png
[6e5d4ac6192d4c24aaf9ce02680b1346.png]: https://img-blog.csdnimg.cn/6e5d4ac6192d4c24aaf9ce02680b1346.png
[Link 1]: https://blog.csdn.net/llg___/article/details/130047839
[fd6b485b7ac34e2e9167c0708a01a0d3.png]: https://img-blog.csdnimg.cn/fd6b485b7ac34e2e9167c0708a01a0d3.png
[d0c767c0de174a6caf5020bb3e29ba08.png]: https://img-blog.csdnimg.cn/d0c767c0de174a6caf5020bb3e29ba08.png
[741f6cb808044bd199a7b059f8ffdd8f.png]: https://img-blog.csdnimg.cn/741f6cb808044bd199a7b059f8ffdd8f.png
[05c236a9eba6450bb708b8b336aab5fa.png]: https://img-blog.csdnimg.cn/05c236a9eba6450bb708b8b336aab5fa.png