学成在线:采用XXL-JOB任务调度方案使用FFmpeg处理视频转码业务

分片技术方案

概述

XXL-JOB并不直接提供数据处理的功能，它只会给所有注册的执行器分配好分片序号，在向执行器下发任务调度的同时携带分片总数和当前分片序号等参数

设计作业分片方案保证多个执行器之间不会查询到重复的任务,保证任务不会重复执行

• 任务添加成功后，这些要处理的任务都会添加到待处理任务表中，然后启动的多个执行器实例会去查询并处理这些待处理任务
• 每个执行器从任务列表获取任务时可以让任务id模上分片总数，取余结果对应需要执行该任务执行器的分片序号,每个执行器查询的任务都是唯一的

任务幂等性

基于作业分片方案可以保证每一个执行器查询到的待处理任务不会重复,但对于同一个执行器并不能保证其不会重复处理其领取到的任务`

一个执行器正在处理的调度任务还没有完成时，此时调度中心可能又下发了一次任务调度请求，此时为了保证执行器不重复处理同一个任务需要进行一些配置

策略	选项
调度过期策略，调度中心错过调度时间的补偿处理策略	忽略：调度过期后忽略过期的任务，从当前时间开始重新计算下次触发时间 立即执行一次(可能重复执行相同的任务)：调度过期后立即执行一次，从当前时间开始重新计算下次触发时间
阻塞处理策略,调度过于密集即当前执行器正在执行任务还没有结束时来不及处理时的处理策略	单机串行(默认)：调度请求进入单机执行器后，调度请求进入FIFO队列并以串行方式运行 丢弃后续调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，本次请求将会被丢弃并标记为失败 覆盖之前调度(可能重复执行任务)：调度请求进入单机执行器后，发现执行器存在运行的调度任务，将会终止运行中的调度任务并清空队列，然后运行本地调度任务

基于以上配置还是无法保同一个执行器不会重复执行任务，因为我们虽然配置了忽略任务,但等到下次触发时间时可能还会执行相同的任务

任务的幂等性：对于数据的操作不论多少次最终结果始终是一致的，如处理视频转码业务时不论任务调度多少次,同一个视频只会执行一次成功的转码

• 执行过的任务可以打一个状态标记已完成,下次再次调度该任务时如果该任务已完成就不再执行

幂等性: 一次和多次请求某一个资源时对于资源(如视频)本身应该具有同样的结果,即使重复调度处理相同的任务也不能重复处理相同的视频

• 场景: 重复提交问题，如恶意刷单，重复支付等问题,如无论执行添加语句多少次最终只会向数据库中插入一条记录

• 数据库约束：比如唯一索引，主键

• 乐观锁：常用于数据库，更新数据时根据乐观锁状态去更新

• 唯一序列号：操作时传递一个唯一序列号, 如在Redis中存储一个序列号当第一次操作完成后就删除该序列号，下回操作时由于获取不到该序列号就无法操作

实现视频处理的幂等性：执行器接收调度请求去执行视频处理任务时需要先判断该视频是否处理完成，如果处理中或处理成功则不再处理

• 在数据库视频处理表中添加处理状态字段，视频处理完成后更新status字段的值，执行器执行任务前会先判断视频的处理状态
• 随着任务的累计,视频处理表中的记录可能会越来越多,此时我们可以将处理成功的任务转移到任务处理历史表(结构一样)中,提高执行器每次查询任务的速度

分布式锁

通过每个执行器从任务列表获取任务时让任务id模上分片总数，取余结果对应需要执行该任务执行器的分片序号，该方式理论上每个执行器分到的任务是不重复的

由于任务调度中心支持执行器弹性扩容的机制，所以无法绝对避免任务不重复执行,此时需要给每个任务配一把锁,只有获取到锁的线程才能执行任务

• 如原来有四个执行器正在执行任务，此时0、1号执行器正在执行视频处理任务，但由于网络问题无法与调度中心通信,此时调度中心就会认为执行器个数减少了
• 调度中心就会对执行器重新编号，那么原来的3、4执行器编号就会变成0、1，他们就会查询并执行和0、1号执行器相同的任务

同步锁：为了避免多线程去争抢同一个任务可以使用synchronized同步锁去解决

• 缺点：synchronized只能保证同一台计算机中的多个线程去争抢同一把锁

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

分布式锁：如果多个执行器分布式部署即多台计算机，此时需要每台计算机上的所有线程争抢(共用)同一把锁(分布式锁)，保证同一个视频只有一个执行器去处理

分布式锁是由一个单独的程序提供加锁、解锁服务，实现的方案有很多

• 基于数据库实现分布式锁：利用数据库主键的唯一性或利用数据库唯一索引、行级锁的特点

  • 多个线程同时向数据库表中插入一条主键相同的记录，哪个线程插入成功就代表哪个线程获取到锁
  • 多个线程同时去更新相同的记录，谁哪个线程更新成功就代表哪个线程抢到锁

• 基于redis实现分布式锁: 基于setnx key value和set key value nx命令和redisson框架等方案

  • 添加一个String类型的键值对，前提是这个key不存在否则不执行，多个线程设置同一个key只会有一个线程设置成功，设置成功的的线程拿到锁

• 使用zookeeper实现分布式锁(结构类似文件目录)：多线程向zookeeper中创建一个子目录(节点)时只会有一个创建成功，谁创建该结点成功谁就 获得锁

操作视频待处理任务

上传视频成功后向视频待处理任务表(media_process)添加视频待处理任务记录,上传视频和添加待处理任务这两个操作需要保证事务的一致性

添加待处理任务

上传视频成功后需要向视频待处理任务表添加视频待处理任务记录,这里暂时只处理avi格式的视频,对于其他格式的文件不会添加待处理任务记录

• 因为上传视频成功后一定会将上传文件的信息添加到media_files文件信息表，所以我们可以将添加文件信息和添加待处理任务记录的操作控制在一个事务中

视频上传完后在addMediaFilesToDb方法中编写addWaitingTask方法添加待处理任务,然后前后端测试上传4个avi视频，观察待处理任务表是否存在任务记录

@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 extension = filename.substring(filename.lastIndexOf("."));// 文件mimeTypeString mimeType = getMimeType(extension);// 如果是avi视频添加到视频待处理表if(mimeType.equals("video/x-msvideo")){MediaProcess mediaProcess = new MediaProcess();BeanUtils.copyProperties(mediaFiles,mediaProcess);mediaProcess.setStatus("1");// 1表示未处理mediaProcess.setFailCount(0);// 失败次数默认为0// 设置url为nullmediaProcess.setUrl(null);int processInsert = mediaProcessMapper.insert(mediaProcess);if (processInsert <= 0) {XueChengPlusException.cast("保存avi视频到待处理表失败");}}
}

查询待处理任务

在MediaProcessMapper中编写根据分片参数获取待处理任务的DAO方法，保证各个执行器查询到的待处理任务记录不重复

• 用任务id对分片总数取模，如果等于该执行器的分片序号则执行
• 同时为了避免同一个任务被同一个执行器执行两次,我们需要额外指定任务状态为未处理(status = 1)或处理失败但处理次数小于3

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

编写MediaFileProcessService接口及其实现类查询待处理任务表中的的待处理任务,指定分片参数和获取记录数(不能超过cpu核心数)

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

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

基于数据库方式实现分布锁

当一个线程开始执行视频处理任务时将任务记录的status字段的值更新为4表示处理中

• 悲观锁: 悲观锁比较适合插入数据,简单粗暴但是性能一般
• 乐观锁: 比较适合更新数据, 性能好但是成功率低(多个线程同时执行时只有一个可以执行成功),还需要访问数据库造成数据库压力过大

 # 多个线程去执行该sql都将会执行成功update media_process m set m.status='4' where  m.id=?# 版本号法，在表中增加一个version字段，更新时判断是否等于某个版本，等于则更新否则更新失败update t1 set t1.data1 = '',t1.version='2' where t1.version='1'# 自定义版本号字段status，多个线程执行该SQL时只有一个线程成功执行，2表示处理成功不用查询update media_process m set m.status='4' where (m.status='1' or m.status='3') and m.fail_count<3 and m.id=?# 更新失败重试，尝试增加版本号字段的值update t1 set t1.count = count+1,t1.version='2' where t1.version='1'update t1 set t1.count = count+1,t1.version='3' where t1.version='2'

在MediaProcessMapper中定义方法,基于乐观锁的原理实现分布式锁,保证最终只有一个线程可以成功执行SQL即获取到锁

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

编写MediaFileProcessService接口及其实现类,开启一个任务,只有抢到锁的线程才可以成功开启任务

/***  开启一个任务* @param id 任务id* @return true开启任务成功，false开启任务失败*/
public boolean startTask(long id);

@Slf4j
@Service
public class MediaFileProcessServiceImpl implements MediaFileProcessService {@AutowiredMediaProcessMapper mediaProcessMapper;public boolean startTask(long id) {int result = mediaProcessMapper.startTask(id);return result<=0?false:true;}
}

更新待处理任务结果

任务处理完成需要更新待处理任务表中status字段的值,如果任务执行成功还需要更新视频的URL,将待处理任务记录从表中删除,同时向历史任务表添加记录

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

@Slf4j
@Service
public class MediaFileProcessServiceImpl implements MediaFileProcessService {@AutowiredMediaFilesMapper mediaFilesMapper;@AutowiredMediaProcessMapper mediaProcessMapper;@AutowiredMediaProcessHistoryMapper mediaProcessHistoryMapper;@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);// 根据Id更新任务处理结果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);// 添加到历史任务记录表MediaProcessHistory mediaProcessHistory = new MediaProcessHistory();BeanUtils.copyProperties(mediaProcess, mediaProcessHistory);mediaProcessHistoryMapper.insert(mediaProcessHistory);// 从待处理任务表中删除处理成功的任务mediaProcessMapper.deleteById(mediaProcess.getId());}
}

视频转码处理

视频上传成功需要对视频格式进行处理,这里我们需要使用Java程序对视频进行处理

视频编码

文件格式: mp4、.avi、rmvb等这些不同扩展名的视频文件的文件格式

编码格式: 视频文件的内容主要包括视频和音频,它们都会按照一定的编码格式去编码,播放器播放音视频时需要根据它们的封装格式去提取出编码并解析

音视频编码格式：通过音视频的压缩技术可以将原始视频格式的文件转换成另一种视频格式的文件,即将视频的编码格式转换成另一种编码格式,目前最常用的编码标准是视频H.264，音频AAC

• MPEG系列视频编码: Mpeg1(vcd),Mpeg2(DVD),Mpeg4(divx，xvid),Mpeg4 AVC(热门)等
• 音频编码: MPEG Audio Layer 1/2、MPEG Audio Layer 3(mp3)、MPEG-2 AAC 、MPEG-4 AAC等
• H.26X系列视频编码: H.261、H.262、H.263、H.263+、H.263++、H.264(MPEG4 AVC合作的结晶)

FFmpeg

视频录制完成后需要使用视频编码软件对视频进行编码如FFmpeg,将ffmpeg.exe加入环境变量Path中后执行ffmpeg -version测试,详情参考文档

ffmpeg.exe -i 1.avi 1.mp4/mp3/gif将一个.avi文件转成mp4、mp3、gif等文件

视频处理工具类

测试使用java.lang.ProcessBuilder执行Windows命令

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

在base工程的util包下创建Mp4VideoUtil类是用于将视频转为mp4格式,使用Java程序调用ffmpeg.exe命令将avi格式的视频转成mp4格式的文件

public static void main(String[] args) throws IOException {// ffmpeg.exe命令的位置String ffmpeg_path = "D:\\soft\\ffmpeg\\ffmpeg.exe";// 源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);// 开始视频转换，成功将返回successString s = videoUtil.generateMp4();System.out.println(s);
}

public class Mp4VideoUtil extends VideoUtil {String ffmpeg_path;String video_path;String mp4_name;String mp4folder_path;public Mp4VideoUtil(String ffmpeg_path, String video_path, String mp4_name, String mp4folder_path){super(ffmpeg_path);this.ffmpeg_path = ffmpeg_path;this.video_path = video_path;this.mp4_name = mp4_name;this.mp4folder_path = mp4folder_path;}// 清除已生成的mp4private void clear_mp4(String mp4_path){// 删除原来已经生成的m3u8及ts文件File mp4File = new File(mp4_path);if(mp4File.exists() && mp4File.isFile()){mp4File.delete();}}/*** 将视频编码生成对应的mp4文件* @return 成功返回success，失败返回控制台日志*/public String generateMp4(){// 清除已生成的mp4clear_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.mp4List<String> commend = new ArrayList<String>();commend.add(ffmpeg_path);commend.add("-i");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);String outstring = null;// 使用Java程序调用`ffmpeg.exe`命令将avi格式的视频转成mp4格式的文件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);if(!check_video_time){return outstring;}else{return "success";}}
}

视频处理任务类

定义任务类VideoTask编写任务的逻辑代码

• 并发处理: 即每个视频使用一个线程去处理，所以每次处理的视频数量不要超过计算机的cpu核心数
• 异步执行任务: 由于线程需要执行的具体任务是在后台异步执行的,所以线程池启动多个线程的动作瞬间完成的即我们定义的任务方法也会立刻完成,此时我们就需要设置一个计数器,保证所有线程都执行完任务后程序才会往下执行
• 超时设置: 线程阻塞时还要设置一个超时时间,防止程序出现未知异常(断电),此时线程没有执行计数器减一的操作会导致其他线程无限期等待

@Slf4j
@Component
public class VideoTask {@AutowiredMediaFileService mediaFileService;@AutowiredMediaFileProcessService mediaFileProcessService;// ffmpeg.exe程序的位置@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();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);// 线程计数器,初始值就是我们的线程总数,每当一个线程执行完后该值会减1CountDownLatch countDownLatch = new CountDownLatch(size);// 将待处理任务加入线程池mediaProcessList.forEach(mediaProcess -> {threadPool.execute(() -> {try {// 任务idLong taskId = mediaProcess.getId();// 各个线程基于乐观锁的原理开始抢任务,只有获取到锁的线程才可以开启任务boolean b = mediaFileProcessService.startTask(taskId);if (!b) {log.debug("抢占任务失败,任务id:{}",taskId);return;}log.debug("开始执行任务:{}", mediaProcess);// 线程抢到任务后开始处理，根据待处理任务中包含的视频文件信息，将其从Minio下载到本地服务器上String bucket = mediaProcess.getBucket();String filePath = mediaProcess.getFilePath();// objectNameString 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;}// 利用工具类对视频进行转码try {// 指定程序位置,源avi视频文件路径,转码后的文件名称,转码后的文件路径
Mp4VideoUtil videoUtil = new Mp4VideoUtil(ffmpegpath, originalFile.getAbsolutePath(), mp4File.getName(), mp4File.getAbsolutePath());// 开始视频转换，成功将返回successString 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;}// 指定转码后的视频在Minio中的存储路径，将转码后生成的视频上传至minioString 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 {// 保证当前线程完成任务后将计数器的值减1,这行代码一定会执行countDownLatch.countDown();}});});// 阻塞即当所有线程都完成任务后程序才会下执行,此时需要设置线程的最大等待时间防止无限期等待countDownLatch.await(30, TimeUnit.MINUTES);}// 获取文件在Minio中完整的存储路径private String getFilePath(String fileMd5,String fileExt){return   fileMd5.substring(0,1) + "/" + fileMd5.substring(1,2) + "/" + fileMd5 + "/" +fileMd5 +fileExt;}
}