1.视频处理
1.1 需求
1.1.1 总体需求
视频上传成功需要对视频的格式进行转码处理，比如：avi转成mp4。如何用Java程序对视频进行处理呢？当视频比较多的时候我们如何可以高效处理
所以一般做文件存储的服务都需要对文件进行处理，例如对视频进行转码处理，可能由于文件量较大需要使用多线程等技术进行高效处理
1.1.2 什么是视频编码
视频上传成功后需要对视频进行转码处理
首先我们要分清文件格式和编码格式：
文件格式：是指.mp4、.avi 等 这些不同扩展名的视频文件的文件格式 ，视频文件的内容主要包括视频和音频，其文件格式是按照一 定的编码格式去编码，并且按照该文件所规定的封装格式将视频、音频、字幕等信息封装在一起，播放器会根据它们的封装格式去提取出编码，然后由播放器解码，最终播放音视频
本项目 使用FFmpeg对视频进行编码
下载：FFmpeg https://www.ffmpeg.org/download.html#build-windows，找到ffmpeg.exe，并将ffmpeg.exe加入环境变量path中
测试是否正常：cmd运行 ffmpeg -version
1.1.3 视频处理工具类
其中Mp4VideoUtil类是用于将视频转为mp4格式，是我们项目要使用的工具类
1.2 分布式任务处理
1.2.1 什么是分布式任务调度
对一个视频的转码可以理解为一个任务的执行，如果视频的数量比较多，如何去高效处理一批任务呢？
(有很多的视频，我们怎么高效的去处理所以我们要用到分布式任务调度)
1.多线程：多线程是充分利用单机的资源
2.分布式加多线程：充分利用多台计算机，每台计算机使用多线程处理(分布式任务调度的处理方案)
定时任务就是一种任务调度，苍穹外卖每到零点进行订单确认，cron表达式
多线程方式实现：
我们可以开启一个线程，每sleep一段时间，就去检查是否已到预期执行时间
以下代码简单实现了 按一定的间隔时间执行任务调度的功能：
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();
}
Timer方式实现：
Jdk也为我们提供了相关支持，如Timer、ScheduledExecutor
Timer 的优点在于简单易用，每个Timer对应一个线程，因此可以同时启动多个Timer并行执行多个任务，同一个Timer中的任务是串行执行
public static void main(String[] args){
Timer timer = new Timer();
timer.schedule(new TimerTask(){
@Override
public void run() {
//TODO：something
}
}, 1000, 2000); //1秒后开始调度，每2秒执行一次
}
ScheduledExecutor方式实现：
Java 5 推出了基于线程池设计的 ScheduledExecutor，其设计思想是，每一个被调度的任务都会由线程池中一个线程去执行，因此任务是并发执行的，相互之间不会受到干扰
imer 和 ScheduledExecutor 都仅能提供基于开始时间与重复间隔的任务调度，不能胜任更加复杂的调度需求。比如，设置每月第一天凌晨1点执行任务、复杂调度任务的管理、任务间传递数据等等
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);
}
第三方Quartz方式实现，项目地址：https://github.com/quartz-scheduler/quartz
Quartz 是一个功能强大的任务调度框架，它可以满足更多更复杂的调度需求，Quartz 设计的核心类包括 Scheduler, Job 以及 Trigger。其中，Job 负责定义需要执行的任务，Trigger 负责设置调度策略，Scheduler 将二者组装在一起，并触发任务开始执行。Quartz支持简单的按时间间隔调度、还支持按日历调度方式，通过设置CronTrigger表达式（包括：秒、分、时、日、月、周、年）进行任务调度
public static void main(String [] agrs) throws SchedulerException {
//创建一个Scheduler
SchedulerFactory schedulerFactory = new StdSchedulerFactory();
Scheduler scheduler = schedulerFactory.getScheduler();
//创建JobDetail
JobBuilder jobDetailBuilder = JobBuilder.newJob(MyJob.class);
jobDetailBuilder.withIdentity("jobName","jobGroupName");
JobDetail jobDetail = jobDetailBuilder.build();
//创建触发的CronTrigger 支持按日历调度
CronTrigger trigger = TriggerBuilder.newTrigger()
.withIdentity("triggerName", "triggerGroupName")
.startNow()
.withSchedule(CronScheduleBuilder.cronSchedule("0/2 * * * * ?"))
.build();
scheduler.scheduleJob(jobDetail,trigger);
scheduler.start();
}
public class MyJob implements Job {
@Override
public void execute(JobExecutionContext jobExecutionContext){
System.out.println("todo something");
}
}
任务调度顾名思义，就是对任务的调度，它是指系统为了完成特定业务，基于给定时间点，给定时间间隔或者给定执行次数自动执行任务
什么是分布式任务调度？
通常任务调度的程序是集成在应用中的，比如：优惠卷服务中包括了定时发放优惠卷的的调度程序，结算服务中包括了定期生成报表的任务调度程序
由于采用分布式架构，一个服务往往会部署多个冗余实例来运行我们的业务，在这种分布式系统环境下运行任务调度，我们称之为分布式任务调度，如下图：
分布式调度要实现的目标：
1.并行任务调度
并行任务调度实现靠多线程，如果有大量任务需要调度，此时光靠多线程就会有瓶颈了，因为一台计算机CPU的处理能力是有限的
如果将任务调度程序分布式部署，每个结点还可以部署为集群，这样就可以让多台计算机共同去完成任务调度，我们可以将任务分割为若干个分片，由不同的实例并行执行，来提高任务调度的处理效率
2.高可用
若某一个实例宕机，不影响其他实例来执行任务
3.弹性扩容
当集群中增加实例就可以提高并执行任务的处理效率
4.避免任务重复执行
当任务调度以集群方式部署，同一个任务调度可能会执行多次，比如在上面提到的电商系统中到点发优惠券的例子，就会发放多次优惠券，对公司造成很多损失，所以我们需要控制相同的任务在多个运行实例上只执行一次
1.2.2 XXL-JOB介绍
XXL-JOB是一个轻量级分布式任务调度平台，其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线，开箱即用(就是个定时任务)
官网：https://www.xuxueli.com/xxl-job/
文档：https://www.xuxueli.com/xxl-job/#%E3%80%8A%E5%88%86%E5%B8%83%E5%BC%8F%E4%BB%BB%E5%8A%A1%E8%B0%83%E5%BA%A6%E5%B9%B3%E5%8F%B0XXL-JOB%E3%80%8B
XXL-JOB主要有调度中心、执行器、任务：
**调度中心：**负责管理调度信息，按照调度配置发出调度请求，自身不承担业务代码，主要职责为执行器管理、任务管理、监控运维、日志管理等
**任务执行器：**负责接收调度请求并执行任务逻辑，只要职责是注册服务、任务执行服务(接收到任务后会放入线程池中的任务队列)、执行结果上报、日志服务等
**任务：**负责执行具体的业务处理
执行流程：
1.任务执行器根据配置的调度中心的地址，自动注册到调度中心
2.达到任务触发条件，调度中心下发任务
3.执行器基于线程池执行任务，并把执行结果放入内存队列中、把执行日志写入日志文件中
4.执行器消费内存队列中的执行结果，主动上报给调度中心
5.当用户在调度中心查看任务日志，调度中心请求任务执行器，任务执行器读取任务日志文件并返回日志详情
(执行器注册到调度中心，调度中心轮询发任务)
1.2.3 搭建XXL-JOB
1.2.3.1 调度中心
首先下载XXL-JOB码云：https://gitee.com/xuxueli0323/xxl-job
项目使用2.3.1版本： https://github.com/xuxueli/xxl-job/releases/tag/2.3.1
使用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 :文档资料，包含数据库脚本，在MySQL中已经创建了xxl_job_2.3.1数据库，导入sql脚本
在本机idea运行xxl-job调度中心
修改数据库配置文件，启动XxlJobAdminApplication即可
运行项目，访问本地 http://127.0.0.1:8080/xxl-job-admin/
账号和密码：admin/123456
1.2.3.2 执行器
下边配置执行器，执行器负责与调度中心通信接收调度中心发起的任务调度请求
1.下边进入调度中心添加执行器
点击新增，填写执行器信息，appname是前边在nacos中配置xxl信息时指定的执行器的应用名
media-process-service
2.首先在媒资管理模块的service工程添加依赖，在项目的父工程已约定了版本2.3.1
<dependency>
<groupId>com.xuxueli</groupId>
<artifactId>xxl-job-core</artifactId>
</dependency>
3.在nacos下的media-service-dev.yaml下配置xxl-job
注意配置中的appname这是执行器的应用名，port是执行器启动的端口，如果本地启动多个执行器注意端口不能重复
xxl:
job:
admin:
addresses: http://127.0.0.1:8080/xxl-job-admin/
executor:
appname: media-process-service
address:
ip:
port: 9999
logpath: /data/applogs/xxl-job/jobhandler
logretentiondays: 30
accessToken: default_token
4.配置xxl-job的执行器
xxl自带的，拷贝到xuecheng-plus-media-service模块的config包下即可
package com.xuecheng.media.config;
import com.xxl.job.core.executor.impl.XxlJobSpringExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class XxlJobConfig {
private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);
@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() {
logger.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;
}
}
5.测试是否通信
到此媒资管理模块service工程配置xxl-job执行器已经完成，测试执行器与调度中心是否正常通信，启动媒资管理模块的接口工程
启动后观察日志，出现下边的日志表示执行器在调度中心注册成功
同时观察调度中心中的执行器界面，在线机器地址处已显示1个执行器
1.2.3.3 执行任务
下边编写测试任务，参考示例工程中任务类的编写方法
在媒资服务service包下新建jobhandler包存放任务类，下边编写一个任务类
package com.xuecheng.media.service.jobhandler;
@Slf4j
@Component
public class SampleJob {
/**
* 1、简单任务示例（Bean模式）
*/
@XxlJob("testJob")//任务名称
public void sampleJobHandler() throws Exception{
log.info("开始执行.....");
}
}
下边在调度中心添加任务，进入任务管理，点击新增，填写任务信息
注意红色标记处
调度类型：
固定速度指按固定的间隔定时调度。
Cron，通过Cron表达式实现更丰富的定时调度策略。
Cron表达式是一个字符串，通过它可以定义调度策略，格式如下：
{秒数} {分钟} {小时} {日期} {月份} {星期} {年份(可为空)}
一些例子如下：
30 10 1 * * ? 每天1点10分30秒触发
0/3 * * * * ? 每3秒触发一次
* 0/10 * * * ? 每10分钟触发一次
xxl-job提供图形界面去配置：
运行模式有BEAN和GLUE，bean模式较常用就是在项目工程中编写执行器的任务代码，GLUE是将任务代码编写在调度中心
JobHandler即任务方法名，填写任务方法上边@XxlJob注解中的名称
路由策略：当执行器集群部署时，调度中心向哪个执行器下发任务，这里选择第一个表示只向第一个执行器下发任务，路由策略的其它选项稍后在分片广播章节详细解释
添加成功，启动任务
启动媒资管理的service工程，启动执行器
日志信息
如果要停止任务需要在调度中心操作
任务跑一段时间注意清理日志
1.3 视频处理技术方案
1.3.1 作业分片方案
视频处理任务添加成功后，对于要处理的任务会添加到待处理任务表中，现在启动多个执行器实例去查询这些待处理任务，此时如何保证多个执行器不会查询到重复的任务呢？
XXL-JOB并不直接提供数据处理的功能，它只会给执行器分配好分片序号，在向执行器任务调度的同时下发分片总数以及分片序号等参数，执行器收到这些参数根据自己的业务需求去利用这些参数(保证执行器查询到不重复的任务)
分片广播：分片是指是调度中心以执行器为维度进行分片，将集群中的执行器标上序号，广播是指每次调度会向集群中的所有执行器发送任务调度，请求中携带分片参数
1.定义作业分片的任务方法：ShardingJobHandler
分片广播和普通任务开发流程一致，不同之处在于可以获取分片参数进行分片业务处理
//2、分片广播任务
@XxlJob("shardingJobHandler")
public void shardingJobHandler() throws Exception {
// 分片参数
int shardIndex = XxlJobHelper.getShardIndex();// 执行器的序号，从0开始，当前分片序号
int shardTotal = XxlJobHelper.getShardTotal();// 执行器总数，总分片数
log.info("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);
}
2.在调度中心添加任务
首先在nacos中配置media-service的本地优先配置：
#配置本地优先
spring:
cloud:
config:
override-none: true
3.下边启动两个执行器实例，观察每个实例的执行情况
将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.3.2 保证任务不重复执行
通过作业分片方案保证了执行器之间查询到不重复的任务，如果一个执行器在处理一个视频还没有完成，此时调度中心又一次请求调度，为了不重复处理同一个视频该怎么办？
任务路由策略
第一个：每次都把任务分给第一个
最后第一个：每次都把任务分给最后第一个
轮询：按照顺序执行
一致HASH：任务有ID，ID是字符串，有hash值
故障转移：如果执行器集群中某一台机器故障，将会自动切换到一台正常的执行器发送调度请求
分片广播任务：一次任务调度将会广播触发集群中所有执行器执行一次任务，可根据分片参数开发分片任务(上面的那几个：每任务都只有一个执行器执行；分片广播：任务可以发送给多个执行器，提高效率)
调度过期策略(调度中心错过调度时间的补偿处理策略)
忽略：调度过期后，忽略过期的任务，从当前时间开始重新计算下次触发时间
立即执行一次：调度过期后，立即执行一次，并从当前时间开始重新计算下次触发时间
阻塞处理策略：调度过于密集执行器来不及处理时的处理策略
这里我们选择忽略，如果立即执行一次就可能重复执行相同的任务
阻塞处理策略(当前执行器正在执行任务还没有结束时，调度中心又调这个执行器执行任务，此时该如何处理)
单机串行(默认)：调度请求进入单机执行器后，调度请求进入FIFO队列并以串行方式运行(用队列模式，排队等待，完成了一个任务，接着完一个任务)
丢弃后续调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，本次请求将会被丢弃并标记为失败
覆盖之前调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，将会终止运行中的调度任务并清空队列，然后运行本地调度任务(当前任务不要了，立刻执行需要的任务)
如果选择覆盖之前调度则可能重复执行任务，这里选择 丢弃后续调度或单机串行方式来避免任务重复执行
只做这些配置可以保证任务不会重复执行吗？
做不到，还需要保证任务处理的幂等性
什么是任务的幂等性？任务的幂等性是指：对于数据的操作不论多少次，操作的结果始终是一致的。在本项目中要实现的是不论多少次任务调度同一个视频只执行一次成功的转码
什么是幂等性？
它描述了一次和多次请求某一个资源对于资源本身应该具有同样的结果。幂等性是为了解决重复提交问题。
解决幂等性常用的方案：
1.数据库约束，比如：唯一索引，主键
2.乐观锁，常用于数据库，更新数据时根据乐观锁状态去更新
3.唯一序列号，操作传递一个唯一序列号，操作时判断与该序列号相等则执行
基于以上分析，在执行器接收调度请求去执行视频处理任务时要实现视频处理的幂等性，要有办法去判断该视频是否处理完成，如果正在处理中或处理完则不再处理。
这里我们在数据库视频处理表中添加处理状态字段，视频处理完成更新状态为完成，执行视频处理前判断状态是否完成，如果完成则不再处理
1.3.3 视频处理方案
确定了分片方案，下边梳理整个视频上传及处理的业务流程，上传视频成功向 视频处理待处理表 添加记录
重要
视频处理的详细流程如下：
1.任务调度中心广播作业分片
2.执行器收到广播作业分片，从数据库取待处理任务，读取未处理及处理失败的任务
3.执行器更新任务为处理中，根据任务内容从MinIO下载要处理的文件。
4.执行器启动多线程去处理任务
5.任务处理完成，上传处理后的视频到MinIO
6.将更新任务处理结果，如果视频处理完成除了更新任务处理结果以外还要将文件的访问地址更新至任务处理表及文件表中，最后将任务完成记录写入历史表
1.4 查询待处理任务
1.4.1 需求分析
查询待处理任务只处理未提交及处理失败的任务，任务处理失败后进行重试，最多重试3次
任务处理成功将待处理记录移动到历史任务表
下图是待处理任务表：
历史任务表与待处理任务表的结构相同
1.4.2 添加待处理任务
上传视频成功向 视频处理待处理表 添加记录，暂时只添加对avi视频的处理记录
根据MIME Type去判断是否是avi视频，avi视频的MIME Type是video/x-msvideo
修改文件信息入库方法，如下：
package com.xuecheng.media.service.impl;
@Service
@Slf4j
public class MediaFileServiceImpl implements MediaFileService {
@Autowired
private MediaProcessMapper mediaProcessMapper;
/**
* 将文件信息添加到文件表
* companyId 机构id
* fileMd5 文件md5值
* uploadFileParamsDto 上传文件的信息
* bucket 桶
* objectName 对象名称
*/
@Transactional
public MediaFiles addMediaFilesToDb(Long companyId, String fileMd5, UploadFileParamsDto uploadFileParamsDto, String bucket, String objectName) {
//......
//保存文件信息到文件表
int insert = mediaFilesMapper.insert(mediaFiles);
if (insert < 0) {
log.error("保存文件信息到数据库失败,{}", mediaFiles.toString());
XueChengPlusException.cast("保存文件信息失败");
}
//添加这里
//添加到待处理任务表，如果是avi视频就写入待处理任务，向mediaProcess表插入数据
addWaitingTask(mediaFiles);
log.debug("保存文件信息到数据库成功,{}", mediaFiles.toString());
}
return mediaFiles;
}
/**
* 添加待处理任务
* mediaFiles 媒资文件信息
*/
private void addWaitingTask(MediaFiles mediaFiles){
//文件名称
String filename = mediaFiles.getFilename();
//文件扩展名
String extension = filename.substring(filename.lastIndexOf("."));
//文件mimeType
String mimeType = getMimeType(extension);
//如果是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);
}
}
}
进行前后端测试，上传3个avi视频，观察待处理任务表是否存在记录，记录是否完成
1.4.3 查询待处理任务
如何保证查询到的待处理视频记录不重复？
编写根据分片参数获取待处理任务的DAO方法，定义DAO接口如下：
package com.xuecheng.media.mapper;
public interface MediaProcessMapper extends BaseMapper<MediaProcess> {
/**
* 根据分片参数获取待处理任务
* shardTotal 分片总数
* shardindex 分片序号
* 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接口，查询待处理
package com.xuecheng.media.service;
/**
* 媒资文件处理业务方法
*/
public interface MediaFileProcessService {
/**
* 获取待处理任务
* shardIndex 分片序号
* shardTotal 分片总数
* count 获取记录数
*/
public List<MediaProcess> getMediaProcessList(int shardIndex,int shardTotal,int count);
}
service接口实现
package com.xuecheng.media.service.impl;
@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;
}
}
1.5 开始执行任务
1.5.1 分布式锁
为了避免多线程去争抢同一个任务可以使用synchronized同步锁去解决
synchronized(锁对象){
执行任务...
}
如果是多个执行器分布式部署，并不能保证同一个视频只有一个执行器去处理
现在要实现分布式环境下所有虚拟机中的线程去同步执行，就需要让多个虚拟机去共用一个锁，虚拟机可以分布式部署，锁也可以分布式部署，如下图：
虚拟机都去抢占同一个锁，锁是一个单独的程序提供加锁、解锁服务
该锁已不属于某个虚拟机，而是分布式部署，由多个虚拟机所共享，这种锁叫分布式锁（锁单独提出来了）
实现分布式锁的方案：
1.基于数据库实现分布锁
利用数据库主键唯一性的特点，多个线程同时去更新相同的记录，谁更新成功谁就抢到锁
2.基于redis实现锁
redis提供了分布式锁的实现方案，比如：SETNX、redisson等
SETNX命令的工作过程是去set一个不存在的key，多个线程去设置同一个key只会有一个线程设置成功，设置成功的的线程拿到锁(setnx：不存在，设置成功)
3.使用zookeeper实现
zookeeper是一个分布式协调服务，主要解决分布式程序之间的同步的问题。zookeeper的结构类似的文件目录，多线程向zookeeper创建一个子目录(节点)，只会有一个创建成功，利用此特点可以实现分布式锁，谁创建该结点成功谁就获得锁
1.5.2 开启任务
1.5.2.1 基于数据库方式实现分布锁
下边基于数据库方式实现分布锁，开始执行任务，将任务执行状态更新为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包裹的代码时需要首先获取锁，没有拿到锁则无法执行，总是悲观的认为别的线程会去抢，所以是悲观锁
乐观锁的思想是它不认为会有线程去争抢，尽管去执行，如果没有执行成功就再去重试
数据库的乐观锁实现方式是在表中增加一个version字段，更新时判断是否等于某个版本，等于则更新否则更新失败，如下方式
update t1 set t1.data1 = '',t1.version='2' where t1.version='1'
实现如下：
1.定义mapper
package com.xuecheng.media.mapper;
public interface MediaProcessMapper extends BaseMapper<MediaProcess> {
/**
* 开启一个任务
* 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);
}
2.在MediaFileProcessService中定义接口
package com.xuecheng.media.service;
public interface MediaFileProcessService {
/**
* 开启一个任务
* id 任务id
* @return true开启任务成功，false开启任务失败
*/
public boolean startTask(long id);
}
package com.xuecheng.media.service.impl;
@Slf4j
@Service
public class MediaFileProcessServiceImpl implements MediaFileProcessService {
public boolean startTask(long id) {
int result = mediaProcessMapper.startTask(id);
return result<=0?false:true;
}
}
1.6 更新任务状态
任务处理完成需要更新任务处理结果，任务执行成功则更新视频的URL、及任务处理结果，将待处理任务记录删除，同时向历史任务表添加记录
在MediaFileProcessService接口添加方法
package com.xuecheng.media.service;
public interface MediaFileProcessService {
/**
* 保存任务结果
* taskId 任务id
* status 任务状态
* fileId 文件id
* url url
* errorMsg 错误信息
*/
void saveProcessFinishStatus(Long taskId,String status,String fileId,String url,String errorMsg);
}
service接口方法实现如下：
package com.xuecheng.media.service.impl;
@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) {
//1.查出要更新的任务，如果不存在则直接返回
MediaProcess mediaProcess = mediaProcessMapper.selectById(taskId);
if (mediaProcess == null) {
return;
}
//2.如果任务执行失败，更新任务处理结果
LambdaQueryWrapper<MediaProcess> queryWrapperById = new LambdaQueryWrapper<MediaProcess>().eq(MediaProcess::getId, taskId);
//处理失败，3：失败
if (status.equals("3")) {
//更新mediaProcess表的状态
MediaProcess mediaProcess_u = new MediaProcess();
mediaProcess_u.setStatus("3");
mediaProcess_u.setErrormsg(errorMsg);
mediaProcess_u.setFailCount(mediaProcess.getFailCount() + 1);//失败次数加1
mediaProcessMapper.update(mediaProcess_u, queryWrapperById);
log.debug("更新任务处理状态为失败，任务信息:{}", mediaProcess_u);
return;
}
//----任务处理成功--------------------
//3.如果任务处理成功
MediaFiles mediaFiles = mediaFilesMapper.selectById(fileId);
if (mediaFiles != null) {
//更新媒资文件中的访问url(mediaFiles表)
mediaFiles.setUrl(url);
mediaFilesMapper.updateById(mediaFiles);
}
//处理成功，更新url和状态(mediaProcess)
mediaProcess.setUrl(url);
mediaProcess.setStatus("2");
mediaProcess.setFinishDate(LocalDateTime.now());
mediaProcessMapper.updateById(mediaProcess);
//添加到历史记录
MediaProcessHistory mediaProcessHistory = new MediaProcessHistory();
BeanUtils.copyProperties(mediaProcess, mediaProcessHistory);
mediaProcessHistoryMapper.insert(mediaProcessHistory);
//删除mediaProcess(从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;
}
}
1.7 视频处理
视频采用并发处理，每个视频使用一个线程去处理，每次处理的视频数量不要超过cpu核心数
所有视频处理完成结束本次执行，为防止代码异常出现无限期等待则添加超时设置，到达超时时间还没有处理完成仍结束任务
media.service模块的jobhander包中，定义任务类VideoTask 如下：
package com.xuecheng.media.service.jobhandler;
@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();
//-----1.查询待处理任务，size:任务数量-----
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;
}
//-----2.启动线程处理视频任务------
//启动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) {
log.debug("抢占任务失败，任务Id:{}", taskId);
return;
}
log.debug("开始执行任务:{}", mediaProcess);
//---下边是处理逻辑---
//桶
String bucket = mediaProcess.getBucket();
//存储路径
String filePath = mediaProcess.getFilePath();
//原始视频的md5值
String fileId = mediaProcess.getFileId();
//原始文件名称
String filename = mediaProcess.getFilename();
//将要处理的文件下载到minio服务器上
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;
}
}
1.8 测试
进入xxl-job调度中心添加执行器和视频处理任务
1.nacos选择对应的执行器，启动XxlJobAdminApplication，进入xxl-job配置好执行器
http://127.0.0.1:8080/xxl-job-admin/
2.配置任务，启动
注意：
1.和代码里面的任务名称保持一致videoJobHandler
2.配置阻塞处理策略为：丢弃后续调度
3.配置视频处理调度时间间隔不用根据视频处理时间去确定，可以配置的小一些，如：5分钟，即使到达调度时间如果视频没有处理完会丢弃调度请求
3.配置视频处理软件地址
4.首先上传至少3个视频，非mp4格式
2就是成功了
