需求: 上传视频

1、需求分析

UI设计图：

• 进入媒资管理
  

• 点击上传视频
  
  

• 选择文件后，开始上传
  

• 上传成功后，可预览视频
  

2、断点续传

http协议本身对上传文件大小没有限制，但是客户的网络环境质量、电脑硬件环境等参差不齐，如果一个大文件快上传完了网断了没有上传完成，需要客户重新上传，用户体验非常差，所以对于大文件上传的要求最基本的是断点续传。

断点续传指的是在下载或上传时，将下载或上传任务（一个文件或一个压缩包）人为的划分为几个部分，每一个部分采用一个线程进行上传或下载，如果碰到网络故障，可以从已经上传或下载的部分开始继续上传下载未完成的部分，而没有必要从头开始上传下载，断点续传可以提高节省操作时间，提高用户体验性。

流程如下：

• 前端上传前先把文件分成块
• 一块一块的上传，上传中断后重新上传，已上传的分块则不用再上传
• 各分块上传完成最后在服务端合并文件

2.1 分块与合并的简单实现

文件分块的思路是:

• 获取源文件的长度–>File类中的length()方法
• 根据设定的分块大小计算块数
• 从源文件读数据并依次向每一块文件写数据

/*** 大文件处理测试*/
public class BigFileTest {//测试文件分块方法@Testpublic void testChunk() throws IOException {//源文件File sourceFile = new File("d:/develop/bigfile_test/nacos.mp4");//存放分块的目录String chunkPath = "d:/develop/bigfile_test/chunk/";File chunkFolder = new File(chunkPath);if (!chunkFolder.exists()) {chunkFolder.mkdirs();}//自己定的分块大小long chunkSize = 1024 * 1024 * 1;//分块数量long chunkNum = (long) Math.ceil(sourceFile.length() * 1.0 / chunkSize);System.out.println("分块总数："+chunkNum);//缓冲区大小byte[] b = new byte[1024];//使用RandomAccessFile访问文件, r即读RandomAccessFile raf_read = new RandomAccessFile(sourceFile, "r");//分块for (int i = 0; i < chunkNum; i++) {//创建分块文件File file = new File(chunkPath + i);if(file.exists()){file.delete();}//创建分块文件fileboolean newFile = file.createNewFile();//判断创建成功后开始写if (newFile) {//相当于创建了一个Writer流RandomAccessFile raf_write = new RandomAccessFile(file, "rw");int len = -1;//write方法向分块文件中写数据while ((len = raf_read.read(b)) != -1) {raf_write.write(b, 0, len);if (file.length() >= chunkSize) {break;}}raf_write.close();System.out.println("完成分块"+i);}}raf_read.close();}}

文件合并的思路是:

• 找到要合并的分块文件后进行排序
• 创建合并文件
• 依次从要合并的文件中读文件并写入上一步的合并文件中
• 使用源文件与合并后文件的md5值来判断是否合并成功

//测试文件合并方法@Testpublic void testMerge() throws IOException {//块文件目录File chunkFolder = new File("d:/develop/bigfile_test/chunk/");//原始文件File originalFile = new File("d:/develop/bigfile_test/nacos.mp4");//最后合并成的文件File mergeFile = new File("d:/develop/bigfile_test/nacos01.mp4");if (mergeFile.exists()) {mergeFile.delete();}//创建新的合并文件mergeFile.createNewFile();//用于写文件RandomAccessFile raf_write = new RandomAccessFile(mergeFile, "rw");//指针指向文件顶端raf_write.seek(0);//byte数组读写缓冲区byte[] b = new byte[1024];//获取分块目录下的文件列表//listFiles方法获取当前目录下的所有子文件，返回一个File[]数组File[] fileArray = chunkFolder.listFiles();// 转成集合，便于排序List<File> fileList = Arrays.asList(fileArray);// 从小到大排序Collections.sort(fileList, new Comparator<File>() {@Overridepublic int compare(File o1, File o2) {//按照文件名称进行排序,分块后文件名为1, 2 ,3...return Integer.parseInt(o1.getName()) - Integer.parseInt(o2.getName());}});//合并文件for (File chunkFile : fileList) {RandomAccessFile raf_read = new RandomAccessFile(chunkFile, "rw");int len = -1;//从分块中读,写入合并文件的File对象while ((len = raf_read.read(b)) != -1) {raf_write.write(b, 0, len);}raf_read.close();}raf_write.close();//校验文件try (FileInputStream fileInputStream = new FileInputStream(originalFile);FileInputStream mergeFileStream = new FileInputStream(mergeFile);) {//取出原始文件的md5String originalMd5 = DigestUtils.md5Hex(fileInputStream);//取出合并文件的md5进行比较String mergeFileMd5 = DigestUtils.md5Hex(mergeFileStream);if (originalMd5.equals(mergeFileMd5)) {System.out.println("合并文件成功");} else {System.out.println("合并文件失败");}}}

2.2 视频上传流程

注意根据流程去分析哪些地方需要后端开发接口，这个接口应该如何去实现

• 前端对文件进行分块
• 前端上传分块前请求媒资服务检查文件是否存在，存在则不上传
• 不存在则开始上传
• 媒资服务将分块上传至MinIO
• 前端将分块上传完毕请求媒资服务合并分块
• 媒资服务判断分块上传完成则请求MinIO合并文件
• 合并完成校验合并后的文件是否完整，如果不完整则删除文件

//将分块文件上传至minio
@Test
public void uploadChunk(){String chunkFolderPath = "D:\\\\develop\\\\upload\\\\chunk\\\\";File chunkFolder = new File(chunkFolderPath);//获取分块文件的数组File[] files = chunkFolder.listFiles();//将每个分块文件上传至miniofor (int i = 0; i < files.length; i++) {try {UploadObjectArgs uploadObjectArgs = UploadObjectArgs.builder().bucket("testbucket")//桶.object("chunk/" + i)//存放的子目录.filename(files[i].getAbsolutePath())//本地文件的路径.build();minioClient.uploadObject(uploadObjectArgs);System.out.println("上传分块成功"+i);} catch (Exception e) {e.printStackTrace();}}}

使用minion提供的方法进行合并文件：

//合并文件，要求分块文件最小5M
@Test
public void test_merge() throws Exception {//创建用于后面传入创建composeObjectArgs的source//也可for循环实现List<ComposeSource> sources = Stream.iterate(0, i -> ++i).limit(6)//这里的map映射，有种把面粉做成面条的感觉.map(i -> ComposeSource.builder().bucket("testbucket").object("chunk/".concat(Integer.toString(i))).build()).collect(Collectors.toList());ComposeObjectArgs composeObjectArgs = ComposeObjectArgs.builder().bucket("testbucket").object("merge01.mp4").sources(sources).build();minioClient.composeObject(composeObjectArgs);}
//清除分块文件
@Test
public void test_removeObjects(){//合并分块完成将分块文件清除List<DeleteObject> deleteObjects = Stream.iterate(0, i -> ++i).limit(6).map(i -> new DeleteObject("chunk/".concat(Integer.toString(i)))).collect(Collectors.toList());RemoveObjectsArgs removeObjectsArgs = RemoveObjectsArgs.builder().bucket("testbucket").objects(deleteObjects).build();Iterable<Result<DeleteError>> results = minioClient.removeObjects(removeObjectsArgs);results.forEach(r->{DeleteError deleteError = null;try {deleteError = r.get();} catch (Exception e) {e.printStackTrace();}});
}

上面使用流来造ComposeSource的集合，使用for循环等价于：

List<ComposeSource> sources = new ArrayList<>();
//循环次数的控制，流中用Stream.iterate+limit实现
for(int i = 0; i<6 ; i++){//这里在流中通过一个映射来实现ComposeSource composeSource = ComposeSource.builder().bucket("testbucket").object("chunk/".concat(Integer.toString(i))).build());sources.add(composeSource); //相当于最后的收集
}

3、接口定义

与前端约定结果类RestResponse.java，操作成功返回{code:0}否则返回{code:-1}

@Api(value = "大文件上传接口", tags = "大文件上传接口")
@RestController
public class BigFilesController {@ApiOperation(value = "文件上传前检查文件")@PostMapping("/upload/checkfile")public RestResponse<Boolean> checkfile(@RequestParam("fileMd5") String fileMd5) throws Exception {return null;}@ApiOperation(value = "分块文件上传前的检测")@PostMapping("/upload/checkchunk")public RestResponse<Boolean> checkchunk(@RequestParam("fileMd5") String fileMd5,@RequestParam("chunk") int chunk) throws Exception {return null;}@ApiOperation(value = "上传分块文件")@PostMapping("/upload/uploadchunk")public RestResponse uploadchunk(@RequestParam("file") MultipartFile file,@RequestParam("fileMd5") String fileMd5,@RequestParam("chunk") int chunk) throws Exception {return null;}@ApiOperation(value = "合并文件")@PostMapping("/upload/mergechunks")public RestResponse mergechunks(@RequestParam("fileMd5") String fileMd5,@RequestParam("fileName") String fileName,@RequestParam("chunkTotal") int chunkTotal) throws Exception {return null;}}

4、Dao层开发–上传分块

继承的BaseMapper足够使用

5、Service层开发–上传分块

先写检查文件或者分块是否存在的接口, Service层接口定义：

/*** @description 检查文件是否存在* @param fileMd5 文件的md5* @return RestResponse<java.lang.Boolean> false不存在，true存在
*/
public RestResponse<Boolean> checkFile(String fileMd5);/*** @description 检查分块是否存在* @param fileMd5  文件的md5* @param chunkIndex  分块序号* @return RestResponse<java.lang.Boolean> false不存在，true存在
*/
public RestResponse<Boolean> checkChunk(String fileMd5, int chunkIndex);

实现Service层方法：

@Override
public RestResponse<Boolean> checkFile(String fileMd5) {//在媒资表中查询文件信息MediaFiles mediaFiles = mediaFilesMapper.selectById(fileMd5);//数据库存在,继续查minion,以防只是数据库中的脏数据if (mediaFiles != null) {//桶String bucket = mediaFiles.getBucket();//存储目录String filePath = mediaFiles.getFilePath();//文件流InputStream stream = null;try {stream = minioClient.getObject(GetObjectArgs.builder().bucket(bucket).object(filePath).build());if (stream != null) {//文件已存在return RestResponse.success(true);}} catch (Exception e) {}}//文件不存在return RestResponse.success(false);
}

@Override
public RestResponse<Boolean> checkChunk(String fileMd5, int chunkIndex) { //得到分块文件目录String chunkFileFolderPath = getChunkFileFolderPath(fileMd5);//得到分块文件的路径String chunkFilePath = chunkFileFolderPath + chunkIndex;//文件流InputStream fileInputStream = null;try {fileInputStream = minioClient.getObject(GetObjectArgs.builder().bucket(bucket_videoFiles).object(chunkFilePath).build());if (fileInputStream != null) {//分块已存在return RestResponse.success(true);}} catch (Exception e) {}//分块未存在return RestResponse.success(false);
}//得到分块文件的目录
//我这里分块存储的路径是,md5值的前两位为两层目录.再加chunk
private String getChunkFileFolderPath(String fileMd5) {return fileMd5.substring(0, 1) + "/" + fileMd5.substring(1, 2) + "/" + fileMd5 + "/" + "chunk" + "/";
}

接下来写上传分块的接口, Service层接口定义为:

/*** @description 上传分块* @param fileMd5  文件md5* @param chunk  分块序号* @param bytes  文件字节
*/
public RestResponse uploadChunk(String fileMd5,int chunk,byte[] bytes);

写实现类:

@Override
public RestResponse uploadChunk(String fileMd5, int chunk, byte[] bytes) {//得到分块文件的目录路径String chunkFileFolderPath = getChunkFileFolderPath(fileMd5);//得到分块文件的路径String chunkFilePath = chunkFileFolderPath + chunk;try {//直接调用之前自己定义的addMediaFilesToMinIO方法,将文件存储至minIOaddMediaFilesToMinIO(bytes, bucket_videoFiles,chunkFilePath);return RestResponse.success(true);} catch (Exception ex) {ex.printStackTrace();log.debug("上传分块文件:{},失败:{}",chunkFilePath,e.getMessage());}return RestResponse.validfail(false,"上传分块失败");
}

6、完善controller层

@ApiOperation(value = "文件上传前检查文件")
@PostMapping("/upload/checkfile")
public RestResponse<Boolean> checkfile(@RequestParam("fileMd5") String fileMd5
) throws Exception {return mediaFileService.checkFile(fileMd5);
}@ApiOperation(value = "分块文件上传前的检测")
@PostMapping("/upload/checkchunk")
public RestResponse<Boolean> checkchunk(@RequestParam("fileMd5") String fileMd5,@RequestParam("chunk") int chunk) throws Exception {return mediaFileService.checkChunk(fileMd5,chunk);
}@ApiOperation(value = "上传分块文件")
@PostMapping("/upload/uploadchunk")
public RestResponse uploadchunk(@RequestParam("file") MultipartFile file,@RequestParam("fileMd5") String fileMd5,@RequestParam("chunk") int chunk) throws Exception {//创建临时文件File tempFile = File.createTempFile("minio", "temp");//上传的文件拷贝到临时文件file.transferTo(tempFile);//文件路径String absolutePath = tempFile.getAbsolutePath();//注意这里获取文件绝对路径的实现思路:创建临时文件--copy--getAbsolutPathreturn mediaFileService.uploadChunk(fileMd5,chunk,absolutePath);
}

7、Service层开发–合并分块

定义Service层接口:

/*** @description 合并分块* @param companyId  机构id* @param fileMd5  文件md5* @param chunkTotal 分块总和* @param uploadFileParamsDto 文件信息*/
public RestResponse mergechunks(Long companyId,String fileMd5,int chunkTotal,UploadFileParamsDto uploadFileParamsDto);

写接口的实现类, 思路大概如下:

• 找到分块文件调用minion的sdk进行文件合并
• 校验合并后的文件和源文件是否一致, 一致则成功
• 将文件信息入库
• 删除清理分块文件

@Override
public RestResponse mergechunks(Long companyId, String fileMd5, int chunkTotal, UploadFileParamsDto uploadFileParamsDto) {//=====获取分块文件路径=====String chunkFileFolderPath = getChunkFileFolderPath(fileMd5);//组成将分块文件路径组成 List<ComposeSource>List<ComposeSource> sourceObjectList = Stream.iterate(0, i -> ++i).limit(chunkTotal).map(i -> ComposeSource.builder().bucket(bucket_videoFiles).object(chunkFileFolderPath.concat(Integer.toString(i))).build()).collect(Collectors.toList());//=====合并=====//文件名称String fileName = uploadFileParamsDto.getFilename();//文件扩展名String extName = fileName.substring(fileName.lastIndexOf("."));//合并文件路径String mergeFilePath = getFilePathByMd5(fileMd5, extName);try {//合并文件ObjectWriteResponse response = minioClient.composeObject(ComposeObjectArgs.builder().bucket(bucket_videoFiles).object(mergeFilePath).sources(sourceObjectList).build());log.debug("合并文件成功:{}",mergeFilePath);} catch (Exception e) {log.debug("合并文件失败,fileMd5:{},异常:{}",fileMd5,e.getMessage(),e);return RestResponse.validfail(false, "合并文件失败。");}// ====验证md5====File minioFile = downloadFileFromMinIO(bucket_videoFiles,mergeFilePath);if(minioFile == null){log.debug("下载合并后文件失败,mergeFilePath:{}",mergeFilePath);return RestResponse.validfail(false, "下载合并后文件失败。");}try (InputStream newFileInputStream = new FileInputStream(minioFile)) {//minio上文件的md5值String md5Hex = DigestUtils.md5Hex(newFileInputStream);//比较md5值，不一致则说明文件不完整if(!fileMd5.equals(md5Hex)){return RestResponse.validfail(false, "文件合并校验失败，最终上传失败。");}//文件大小uploadFileParamsDto.setFileSize(minioFile.length());}catch (Exception e){log.debug("校验文件失败,fileMd5:{},异常:{}",fileMd5,e.getMessage(),e);return RestResponse.validfail(false, "文件合并校验失败，最终上传失败。");}finally {if(minioFile!=null){minioFile.delete();}}//文件入库currentProxy.addMediaFilesToDb(companyId,fileMd5,uploadFileParamsDto,bucket_videoFiles,mergeFilePath);//=====清除分块文件=====clearChunkFiles(chunkFileFolderPath,chunkTotal);return RestResponse.success(true);
}/*** 从minio下载文件* @param bucket 桶* @param objectName 对象名称* @return 下载后的文件*/
public File downloadFileFromMinIO(String bucket,String objectName){//临时文件File minioFile = null;FileOutputStream outputStream = null;try{InputStream stream = minioClient.getObject(GetObjectArgs.builder().bucket(bucket).object(objectName).build());//创建临时文件minioFile=File.createTempFile("minio", ".merge");outputStream = new FileOutputStream(minioFile);IOUtils.copy(stream,outputStream);return minioFile;} catch (Exception e) {e.printStackTrace();}finally {if(outputStream!=null){try {outputStream.close();} catch (IOException e) {e.printStackTrace();}}}return null;
}
/*** 得到合并后的文件的地址* @param fileMd5 文件id即md5值* @param fileExt 文件扩展名* @return*/
private String getFilePathByMd5(String fileMd5,String fileExt){return   fileMd5.substring(0,1) + "/" + fileMd5.substring(1,2) + "/" + fileMd5 + "/" +fileMd5 +fileExt;
}/*** 清除分块文件* @param chunkFileFolderPath 分块文件路径* @param chunkTotal 分块文件总数*/
private void clearChunkFiles(String chunkFileFolderPath,int chunkTotal){try {List<DeleteObject> deleteObjects = Stream.iterate(0, i -> ++i).limit(chunkTotal).map(i -> new DeleteObject(chunkFileFolderPath.concat(Integer.toString(i)))).collect(Collectors.toList());RemoveObjectsArgs removeObjectsArgs = RemoveObjectsArgs.builder().bucket("video").objects(deleteObjects).build();Iterable<Result<DeleteError>> results = minioClient.removeObjects(removeObjectsArgs);results.forEach(r->{DeleteError deleteError = null;try {deleteError = r.get();} catch (Exception e) {e.printStackTrace();log.error("清除分块文件失败,objectname:{}",deleteError.objectName(),e);}});} catch (Exception e) {e.printStackTrace();log.error("清除分块文件失败,chunkFileFolderPath:{}",chunkFileFolderPath,e);}

8、完善合并分块controller层

@ApiOperation(value = "合并文件")
@PostMapping("/upload/mergechunks")
public RestResponse mergechunks(@RequestParam("fileMd5") String fileMd5,@RequestParam("fileName") String fileName,@RequestParam("chunkTotal") int chunkTotal) throws Exception {Long companyId = 1232141425L;UploadFileParamsDto uploadFileParamsDto = new UploadFileParamsDto();uploadFileParamsDto.setFileType("001002");uploadFileParamsDto.setTags("课程视频");uploadFileParamsDto.setRemark("");uploadFileParamsDto.setFilename(fileName);return mediaFileService.mergechunks(companyId,fileMd5,chunkTotal,uploadFileParamsDto);}

9、Spring事务失效的情况

• 在方法中直接try…catch捕获异常, 没有抛出去
• 非事务方法调用事务方法
• 事务方法内部调用事务方法
• 使用@Transactional标记的方法不是public
• 抛出的异常与@Transactional注解中rollbackFor指定的异常不匹配(rollbackFor默认值为RunTimeException)
• 数据库表不支持事务, 如MySql的MyISAM
• Spring的传播行为导致事务失效,