发时间阻塞处理策略,调度过于密集即当前执行器正在执行任务还没有结束时来不及处理时的处理策略单机串行(默认)调度请求进入单机执行器后调度请求进入FIFO队列并以串行方式运行丢弃后续调度调度请求进入单机执行器后发现执行器存在运行的调度任务本次请求将会被丢弃并标记为失败覆盖之前调度(可能重复执行任务)调度请求进入单机执行器后发现执行器存在运行的调度任务将会终止运行中的调度任务并清空队列然后运行本地调度任务

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

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

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

幂等性:

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

场景:

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

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

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

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

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

分布式锁

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

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

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

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

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

synchronized(锁对象){

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

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

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

基于redis实现分布式锁:

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

使用zookeeper实现分布式锁(结构类似文件(节点)时只会有一个创建成功谁创建该结点成功谁就

获得锁

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

添加待处理任务

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

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

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

Transactional

mediaFilesMapper.selectById(fileMd5);if

(mediaFiles

拷贝基本信息BeanUtils.copyProperties(uploadFileParamsDto,

mediaFiles);mediaFiles.setId(fileMd5);mediaFiles.setFileId(fileMd5);mediaFiles.setCompanyId(companyId);//

bucket

objectName);mediaFiles.setBucket(bucket);mediaFiles.setFilePath(objectName);mediaFiles.setCreateDate(LocalDateTime.now());mediaFiles.setAuditStatus(002003);mediaFiles.setStatus(1);//

insert

mediaFilesMapper.insert(mediaFiles);if

(insert

mediaFiles.toString());XueChengPlusException.cast(保存文件信息失败);}//

添加待处理任务到待处理任务表addWaitingTask(mediaFiles);log.debug(保存文件信息到数据库成功,{},

mediaFiles;}

filename.substring(filename.lastIndexOf(.));//

文件mimeTypeString

如果是avi视频添加到视频待处理表if(mimeType.equals(video/x-msvideo)){MediaProcess

mediaProcess

MediaProcess();BeanUtils.copyProperties(mediaFiles,mediaProcess);mediaProcess.setStatus(1);//

1表示未处理mediaProcess.setFailCount(0);//

失败次数默认为0//

设置url为nullmediaProcess.setUrl(null);int

processInsert

mediaProcessMapper.insert(mediaProcess);if

(processInsert

{XueChengPlusException.cast(保存avi视频到待处理表失败);}}

}查询待处理任务

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

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

1)或处理失败但处理次数小于3

selectListByShardIndex(Param(shardTotal)

int

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

public

mediaProcessMapper;Overridepublic

ListMediaProcess

mediaProcessMapper.selectListByShardIndex(shardTotal,

shardIndex,

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

悲观锁:

性能好但是成功率低(多个线程同时执行时只有一个可以执行成功),还需要访问数据库造成数据库压力过大

media_process

版本号法在表中增加一个version字段更新时判断是否等于某个版本等于则更新否则更新失败update

set

自定义版本号字段status多个线程执行该SQL时只有一个线程成功执行2表示处理成功不用查询update

media_process

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

public

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

/***

mediaProcessMapper.startTask(id);return

result0?false:true;}

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

/***

mediaFilesMapper;AutowiredMediaProcessMapper

mediaProcessMapper;AutowiredMediaProcessHistoryMapper

mediaProcessHistoryMapper;TransactionalOverridepublic

void

mediaProcessMapper.selectById(taskId);if(mediaProcess

null){return

任务处理失败更新任务处理结果LambdaQueryWrapperMediaProcess

queryWrapperById

LambdaQueryWrapperMediaProcess().eq(MediaProcess::getId,

taskId);if(status.equals(3)){MediaProcess

mediaProcess_u

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

;}//

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();BeanUtils.copyProperties(mediaProcess,

mediaProcessHistory);mediaProcessHistoryMapper.insert(mediaProcessHistory);//

从待处理任务表中删除处理成功的任务mediaProcessMapper.deleteById(mediaProcess.getId());}

}视频转码处理

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

视频编码

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

编码格式:

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

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

MPEG系列视频编码:

Mpeg1(vcd),Mpeg2(DVD),Mpeg4(divxxvid),Mpeg4

AVC(热门)等音频编码:

H.261、H.262、H.263、H.263、H.263、H.264(MPEG4

FFmpeg

视频录制完成后需要使用视频编码软件对视频进行编码如FFmpeg,将ffmpeg.exe加入环境变量Path中后执行ffmpeg

-version测试,详情参考文档

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

视频处理工具类

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

ProcessBuilder

(x86)\\Tencent\\QQ\\Bin\\QQScLauncher.exe);

builder.redirectErrorStream(true);

执行命令

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

public

D:\\soft\\ffmpeg\\ffmpeg.exe;//

video_path

D:\\develop\\bigfile_test\\nacos01.avi;//

mp4_name

D:\\develop\\bigfile_test\\nacos01.mp4;//

创建工具类对象Mp4VideoUtil

Mp4VideoUtil(ffmpeg_path,video_path,mp4_name,mp4_path);//

videoUtil.generateMp4();System.out.println(s);

}public

mp4folder_path){super(ffmpeg_path);this.ffmpeg_path

mp4folder_path;}//

File(mp4_path);if(mp4File.exists()

mp4File.isFile()){mp4File.delete();}}/***

return

清除已生成的mp4clear_mp4(mp4folder_path);//

拼接命令ffmpeg.exe

ArrayListString();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

使用Java程序调用ffmpeg.exe命令将avi格式的视频转成mp4格式的文件try

{ProcessBuilder

ProcessBuilder();builder.command(commend);//

将标准输入流和错误输入流合并通过标准输入流程读取信息builder.redirectErrorStream(true);Process

waitFor(p);}

this.check_video_time(video_path,

mp4folder_path);if(!check_video_time){return

success;}}

即每个视频使用一个线程去处理所以每次处理的视频数量不要超过计算机的cpu核心数异步执行任务:

由于线程需要执行的具体任务是在后台异步执行的,所以线程池启动多个线程的动作瞬间完成的即我们定义的任务方法也会立刻完成,此时我们就需要设置一个计数器,保证所有线程都执行完任务后程序才会往下执行超时设置:

线程阻塞时还要设置一个超时时间,防止程序出现未知异常(断电),此时线程没有执行计数器减一的操作会导致其他线程无限期等待

Slf4j

mediaFileService;AutowiredMediaFileProcessService

ffmpeg.exe程序的位置Value(${videoprocess.ffmpegpath})String

ffmpegpath;XxlJob(videoJobHandler)public

void

XxlJobHelper.getShardIndex();int

shardTotal

XxlJobHelper.getShardTotal();ListMediaProcess

mediaProcessList

Runtime.getRuntime().availableProcessors();mediaProcessList

mediaFileProcessService.getMediaProcessList(shardIndex,

shardTotal,

mediaProcessList.size();log.debug(取出待处理视频任务{}条,

size);if

{e.printStackTrace();return;}//

创建一个包含size个线程的线程池将来每一个线程对应一个视频处理任务ExecutorService

threadPool

Executors.newFixedThreadPool(size);//

线程计数器,初始值就是我们的线程总数,每当一个线程执行完后该值会减1CountDownLatch

countDownLatch

将待处理任务加入线程池mediaProcessList.forEach(mediaProcess

{try

各个线程基于乐观锁的原理开始抢任务,只有获取到锁的线程才可以开启任务boolean

mediaFileProcessService.startTask(taskId);if

(!b)

{log.debug(抢占任务失败,任务id:{},taskId);return;}log.debug(开始执行任务:{},

mediaProcess);//

线程抢到任务后开始处理根据待处理任务中包含的视频文件信息将其从Minio下载到本地服务器上String

bucket

mediaProcess.getBucket();String

filePath

mediaProcess.getFileId();String

filename

mediaProcess.getFilename();File

originalFile

mediaFileService.downloadFileFromMinIO(mediaProcess.getBucket(),

(originalFile

{log.debug(下载待处理文件失败,originalFile:{},

mediaProcess.getBucket().concat(mediaProcess.getFilePath()));//

保存任务处理失败的结果mediaFileProcessService.saveProcessFinishStatus(mediaProcess.getId(),

fileId,

保存任务处理失败的结果mediaFileProcessService.saveProcessFinishStatus(mediaProcess.getId(),

fileId,

指定程序位置,源avi视频文件路径,转码后的文件名称,转码后的文件路径

Mp4VideoUtil

originalFile.getAbsolutePath(),

mp4File.getName(),

{e.printStackTrace();log.error(处理视频文件:{},出错:{},

e.getMessage());}if

{log.error(处理视频失败,视频地址:{},错误信息:{},

bucket

保存任务处理失败的结果mediaFileProcessService.saveProcessFinishStatus(mediaProcess.getId(),

fileId,

指定转码后的视频在Minio中的存储路径将转码后生成的视频上传至minioString

objectName

{mediaFileService.addMediaFilesToMinIO(mp4File.getAbsolutePath(),

video/mp4,

任务处理成功,将url保存到文件信息表并更新状态为成功同时将处理成功的任务记录删除并存入历史任务表mediaFileProcessService.saveProcessFinishStatus(mediaProcess.getId(),

fileId,

{log.error(上传视频失败或入库失败,视频地址:{},错误信息:{},

bucket

保存任务处理失败的结果mediaFileProcessService.saveProcessFinishStatus(mediaProcess.getId(),

fileId,

保证当前线程完成任务后将计数器的值减1,这行代码一定会执行countDownLatch.countDown();}});});//

阻塞即当所有线程都完成任务后程序才会下执行,此时需要设置线程的最大等待时间防止无限期等待countDownLatch.await(30,

String