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

分片技术方案

概述

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

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

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

在这里插入图片描述

任务幂等性

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

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

在这里插入图片描述

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

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

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

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

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

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

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

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

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

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

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

在这里插入图片描述

分布式锁

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

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

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

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

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

在这里插入图片描述

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

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

在这里插入图片描述

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

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

    • 多个线程同时向数据库表中插入一条主键相同的记录,哪个线程插入成功就代表哪个线程获取到锁
    • 多个线程同时去更新相同的记录,谁哪个线程更新成功就代表哪个线程抢到锁
  • 基于redis实现分布式锁: 基于setnx key valueset 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("."));
    // 文件mimeType
    String 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为null
        mediaProcess.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 {
    @Autowired
    MediaProcessMapper mediaProcessMapper;

    @Override
    public 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 {
    @Autowired
    MediaProcessMapper 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 {
    @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);
            // 根据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);
    // 开始视频转换,成功将返回success
    String 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;
    }
    // 清除已生成的mp4
    private 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(){
        // 清除已生成的mp4
        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(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 {
    @Autowired
    MediaFileService mediaFileService;
    @Autowired
    MediaFileProcessService 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);
    // 线程计数器,初始值就是我们的线程总数,每当一个线程执行完后该值会减1
    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);
                // 线程抢到任务后开始处理,根据待处理任务中包含的视频文件信息,将其从Minio下载到本地服务器上
                String bucket = mediaProcess.getBucket();
                String filePath = mediaProcess.getFilePath();// objectName
                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;
                }
       			// 利用工具类对视频进行转码
                try {
                    // 指定程序位置,源avi视频文件路径,转码后的文件名称,转码后的文件路径
Mp4VideoUtil videoUtil = new Mp4VideoUtil(ffmpegpath, originalFile.getAbsolutePath(), mp4File.getName(), mp4File.getAbsolutePath());
                    // 开始视频转换,成功将返回success
                    String 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中的存储路径,将转码后生成的视频上传至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 {
                // 保证当前线程完成任务后将计数器的值减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;
    }
}

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mfbz.cn/a/368806.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

机器学习-基础分类算法-KNN详解

KNN-k近邻算法 k-Nearest Neighbors 思想极度简单应用数学只是少效果好可以解释机器学习算法使用过程中的很多细节问题更完整的刻画机器学习应用的流程 创建简单测试用例 import numpy as np import matplotlib.pyplot as plt raw_data_X [[3.393533211, 2.331273381],[3.1…

Flutter实现轮播图功能

一、在pubspec.yaml中添加&#xff1a; dependencies:# 轮播图card_swiper: ^3.0.1card_swiper: ^3.0.1&#xff0c;要获取最新版本&#xff1a;https://pub-web.flutter-io.cn/packages/card_swiper/versions&#xff0c;这个里面有文档可以看&#xff0c;如下图&#xff1a;…

大模型ReAct智能体开发实战

哆啦A梦是很多人都熟悉的角色&#xff0c;包括我自己。 在成长过程中&#xff0c;我常常对他口袋里的许多小玩意感到惊讶&#xff0c;而且他知道何时使用它们。 随着大型语言模型 (LLM) 的发展趋势&#xff0c;你也可以构建一个具有相同行为方式的模型&#xff01; 我们将构建…

高中数学立体几何练习题3

用到的基础知识&#xff1a; 1. 2.

MATLAB计算多边形质心/矩心

前言&#xff1a;不规则四边形的中心 不规则四边形的出心有多种定义&#xff0c;以下是最常见的三种&#xff1a; 1.重心&#xff1a;重心是四边形内部所有顶点连线交点的平均位置。可以通过求解四个顶点坐标的平均值来找到重心。 2.质心&#xff1a;质心是四边形内部所有质点…

Python机器学习库(numpy库)

文章目录 Python机器学习库&#xff08;numpy库&#xff09;1. 数据的维度2. numpy基础知识2.1 numpy概述2.1 numpy概述2.1 numpy概述2.2 numpy库的引用 3. ndarray数组的创建3.1 N维数组对象ndarray3.2 创建ndarray数组3.2.1 使用Python列表、元组创建ndarray数组3.2.2 使用nu…

029 命令行传递参数

1.循环输出args字符串数组 public class D001 {public static void main(String[] args) {for (String arg : args) {System.out.println(arg);}} } 2.找打这个类的路径&#xff0c;打开cmd cmd C:\Users\Admin\IdeaProjects\JavaSE学习之路\scanner\src\com\yxm\demo 3. 编译…

Servlet+Ajax实现对数据的列表展示(极简入门)

目录 1.准备工作 1.数据库源&#xff08;这里以Mysql为例&#xff09; 2.映射实体类 3.模拟三层架构&#xff08;Dao、Service、Controller&#xff09; Dao接口 Dao实现 Service实现&#xff08;这里省略Service接口&#xff09; Controller层&#xff08;或叫Servlet层…

2024济南生物发酵展:会议日程安排和技术装备亮点预告

2024济南发酵展/2024生物发酵展/2024山东发酵展/2024济南生物制药展/2024生物技术展/2024食品设备展/2024食品加工展/2024济南细胞工程展 由中国生物发酵产业协会主办&#xff0c;上海信世展览服务有限公司承办的2024第12届国际生物发酵产品与技术装备展览会&#xff08;济南&a…

深入理解Istio服务网格数据平面Envoy

一、服务网格概述(service mesh) 在传统的微服务架构中&#xff0c;服务间的调用&#xff0c;业务代码需要考虑认证、熔断、服务发现等非业务能力&#xff0c;在某种程度上&#xff0c;表现出了一定的耦合性 服务网格追求高级别的服务流量治理能力&#xff0c;认证、熔断、服…

2023.12 淘天-数科 已offer

文章目录 岗位信息1面ld 12.17 1H2面 VP 12.18 40min3面 HR 12.2012.21offer薪资方案沟通 岗位信息 1面ld 12.17 1H &#xff08;是一个从业估计很长时间前辈&#xff0c;很平和&#xff0c;感觉能学到很多东西&#xff09; 自我介绍项目深究1.说下自己工作里最有成就感的事和…

图论练习3

内容&#xff1a;过程中视条件改变边权&#xff0c;利用树状数组区间加处理 卯酉东海道 题目链接 题目大意 个点&#xff0c;条有向边&#xff0c;每条边有颜色和费用总共有种颜色若当前颜色与要走的边颜色相同&#xff0c;则花费为若当前颜色与要走的边颜色不同&#xff0c;…

MYSQL——MySQL8.3无法启动

在新电脑上装了个MySQL&#xff0c;但是无法使用net start mysql启动&#xff0c;很是纳闷&#xff0c;使用mysqld --console去查看报错&#xff0c;也是没报错的&#xff0c;但是奇怪的是&#xff0c;我输入完这个mysqld --console之后&#xff0c;就等于启动了mysql了&#x…

第十一篇【传奇开心果系列】Python的OpenCV技术点案例示例:三维重建

传奇开心果短博文系列 系列短博文目录Python的OpenCV技术点案例示例系列短博文目录一、前言二、OpenCV三维重建介绍三、基于区域的SGBM示例代码四、BM(Block Matching)算法介绍和示例代码五、基于能量最小化的GC(Graph Cut)算法介绍和示例代码六、相机标定介绍和示例代码七…

【数据结构与算法】之排序系列-20240203

这里写目录标题 一、628. 三个数的最大乘积二、645. 错误的集合三、747. 至少是其他数字两倍的最大数四、905. 按奇偶排序数组五、922. 按奇偶排序数组 II六、976. 三角形的最大周长 一、628. 三个数的最大乘积 简单 给你一个整型数组 nums &#xff0c;在数组中找出由三个数组…

Leetcode刷题笔记题解(C++):36. 有效的数独

思路一&#xff1a;暴力破解&#xff0c;两个二维数组记录行、列对应的数字出现的次数&#xff0c;比如rows[i][index]表示的数字index在i行出现的次数&#xff0c;三维数组记录每个块中对应数字出现的次数&#xff0c;比如boxes[i/3][j/3][index]表示的数字index在[i/3][j/3]个…

Hugging Face推出自定义AI聊天Assistants;谷歌推出图像生成工具 ImageFX

&#x1f989; AI新闻 &#x1f680; 谷歌推出图像生成工具 ImageFX 摘要&#xff1a;谷歌在 Imagen 2 的基础上推出新的图像生成工具 ImageFX&#xff0c;通过简单的文字提示可以生成高质量图像。该工具包含了提示界面&#xff0c;让用户可以快速尝试创作和想法的相邻维度。…

数据结构—基础知识:哈夫曼树

文章目录 数据结构—基础知识&#xff1a;哈夫曼树哈夫曼树的基本概念哈夫曼树的构造算法哈夫曼树的构造过程哈夫曼算法的实现算法&#xff1a;构造哈夫曼树 数据结构—基础知识&#xff1a;哈夫曼树 哈夫曼树的基本概念 哈夫曼&#xff08;Huffman&#xff09;树又称最优树&…

通过 ChatGPT 的 Function Call 查询数据库

用 Function Calling 的方式实现手机流量包智能客服的例子。 def get_sql_completion(messages, model"gpt-3.5-turbo"):response client.chat.completions.create(modelmodel,messagesmessages,temperature0,tools[{ # 摘自 OpenAI 官方示例 https://github.com/…

ASP.NET Core 自定义解压缩提供程序

写在前面 在了解ASP.NET Core 自定义请求解压缩中间件的应用时&#xff0c;依据官方文档操作下来碰到了几个问题&#xff0c;这边做个记录。 关键点就是配置 Content-Encoding&#xff0c;参数需要和代码中添加的提供程序的Key保持一致&#xff1b; builder.Services.AddRequ…