SpringBoot 整合 MinIO 實(shí)現(xiàn)視頻的分片上傳/斷點(diǎn)續(xù)傳（親測(cè)可行）
1、前言
之前做了一個(gè)慕課網(wǎng)上的仿短視頻開(kāi)發(fā)，里面有很多比較粗糙的實(shí)現(xiàn)，比如視頻上傳部分是直接由前端上傳云服務(wù)，沒(méi)考慮到客戶的網(wǎng)絡(luò)環(huán)境質(zhì)量等問(wèn)題，如果一個(gè)視頻快上傳完了，但是網(wǎng)斷了沒(méi)有上傳完成需要客戶重新上傳，這對(duì)于用戶體驗(yàn)是極差的。
那么我們對(duì)于視頻文件的上傳可以采取斷點(diǎn)續(xù)傳，上傳過(guò)程中，如果出現(xiàn)網(wǎng)絡(luò)異?；虺绦虮罎?dǎo)致文件上傳失敗時(shí)，將從斷點(diǎn)記錄處繼續(xù)上傳未上傳完成的部分，斷點(diǎn)續(xù)傳依賴于MD5和分片上傳，對(duì)于本demo分片上傳的流程如圖
通過(guò)文件唯一標(biāo)識(shí)MD5，在數(shù)據(jù)庫(kù)中查詢此前是否創(chuàng)建過(guò)該SysUploadTask，如果存在，直接返回TaskInfo；如果不存在，通過(guò)amazonS3獲取到UploadId并新建一個(gè)SysUploadTask返回。
前端將文件分好片后，通過(guò)服務(wù)器得到每一片的一個(gè)預(yù)地址，然后由前端直接向minio服務(wù)器發(fā)起真正的上傳請(qǐng)求，避免上傳時(shí)占用應(yīng)用服務(wù)器的帶寬，影響系統(tǒng)穩(wěn)定。最后再向后端服務(wù)器發(fā)起合并請(qǐng)求。
2、數(shù)據(jù)庫(kù)結(jié)構(gòu)
3、后端實(shí)現(xiàn)
3.1、根據(jù)MD5獲取是否存在相同文件
Controller層
/**
* 查詢是否上傳過(guò)，若存在，返回TaskInfoDTO
* @param identifier 文件md5
* @return
*/
@GetMapping("/{identifier}")
public GraceJSONResult taskInfo (@PathVariable("identifier") String identifier) {
return GraceJSONResult.ok(sysUploadTaskService.getTaskInfo(identifier));
}
Service層
/**
* 查詢是否上傳過(guò)，若存在，返回TaskInfoDTO
* @param identifier
* @return
*/
public TaskInfoDTO getTaskInfo(String identifier) {
SysUploadTask task = getByIdentifier(identifier);
if (task == null) {
return null;
}
TaskInfoDTO result = new TaskInfoDTO().setFinished(true).setTaskRecord(TaskRecordDTO.convertFromEntity(task)).setPath(getPath(task.getBucketName(), task.getObjectKey()));
boolean doesObjectExist = amazonS3.doesObjectExist(task.getBucketName(), task.getObjectKey());
if (!doesObjectExist) {
// 未上傳完，返回已上傳的分片
ListPartsRequest listPartsRequest = new ListPartsRequest(task.getBucketName(), task.getObjectKey(), task.getUploadId());
PartListing partListing = amazonS3.listParts(listPartsRequest);
result.setFinished(false).getTaskRecord().setExitPartList(partListing.getParts());
}
return result;
}
3.2、初始化一個(gè)上傳任務(wù)
Controller層
/**
* 創(chuàng)建一個(gè)上傳任務(wù)
* @return
*/
@PostMapping
public GraceJSONResult initTask (@Valid @RequestBody InitTaskParam param) {
return GraceJSONResult.ok(sysUploadTaskService.initTask(param));
}
Service層
/**
* 初始化一個(gè)任務(wù)
*/
public TaskInfoDTO initTask(InitTaskParam param) {
Date currentDate = new Date();
String bucketName = minioProperties.getBucketName();
String fileName = param.getFileName();
String suffix = fileName.substring(fileName.lastIndexOf(".")+1, fileName.length());
String key = StrUtil.format("{}/{}.{}", DateUtil.format(currentDate, "YYYY-MM-dd"), IdUtil.randomUUID(), suffix);
String contentType = MediaTypeFactory.getMediaType(key).orElse(MediaType.APPLICATION_OCTET_STREAM).toString();
ObjectMetadata objectMetadata = new ObjectMetadata();
objectMetadata.setContentType(contentType);
InitiateMultipartUploadResult initiateMultipartUploadResult = amazonS3
.initiateMultipartUpload(new InitiateMultipartUploadRequest(bucketName, key).withObjectMetadata(objectMetadata));
String uploadId = initiateMultipartUploadResult.getUploadId();
SysUploadTask task = new SysUploadTask();
int chunkNum = (int) Math.ceil(param.getTotalSize() * 1.0 / param.getChunkSize());
task.setBucketName(minioProperties.getBucketName())
.setChunkNum(chunkNum)
.setChunkSize(param.getChunkSize())
.setTotalSize(param.getTotalSize())
.setFileIdentifier(param.getIdentifier())
.setFileName(fileName)
.setObjectKey(key)
.setUploadId(uploadId);
sysUploadTaskMapper.insert(task);
return new TaskInfoDTO().setFinished(false).setTaskRecord(TaskRecordDTO.convertFromEntity(task)).setPath(getPath(bucketName, key));
}
3.3、獲取每個(gè)分片的預(yù)簽名上傳地址
Controller層
/**
* 獲取每個(gè)分片的預(yù)簽名上傳地址
* @param identifier
* @param partNumber
* @return
*/
@GetMapping("/{identifier}/{partNumber}")
public GraceJSONResult preSignUploadUrl (@PathVariable("identifier") String identifier, @PathVariable("partNumber") Integer partNumber) {
SysUploadTask task = sysUploadTaskService.getByIdentifier(identifier);
if (task == null) {
return GraceJSONResult.error("分片任務(wù)不存在");
}
Map<String, String> params = new HashMap<>();
params.put("partNumber", partNumber.toString());
params.put("uploadId", task.getUploadId());
return GraceJSONResult.ok(sysUploadTaskService.genPreSignUploadUrl(task.getBucketName(), task.getObjectKey(), params));
}
Service層
/**
* 生成預(yù)簽名上傳url
* @param bucket 桶名
* @param objectKey 對(duì)象的key
* @param params 額外的參數(shù)
* @return
*/
public String genPreSignUploadUrl(String bucket, String objectKey, Map<String, String> params) {
Date currentDate = new Date();
Date expireDate = DateUtil.offsetMillisecond(currentDate, PRE_SIGN_URL_EXPIRE.intValue());
GeneratePresignedUrlRequest request = new GeneratePresignedUrlRequest(bucket, objectKey)
.withExpiration(expireDate).withMethod(HttpMethod.PUT);
if (params != null) {
params.forEach((key, val) -> request.addRequestParameter(key, val));
}
URL preSignedUrl = amazonS3.generatePresignedUrl(request);
return preSignedUrl.toString();
}
3.4、合并分片
Controller層
/**
* 合并分片
* @param identifier
* @return
*/
@PostMapping("/merge/{identifier}")
public GraceJSONResult merge (@PathVariable("identifier") String identifier) {
sysUploadTaskService.merge(identifier);
return GraceJSONResult.ok();
}
Service層
/**
* 合并分片
* @param identifier
*/
public void merge(String identifier) {
SysUploadTask task = getByIdentifier(identifier);
if (task == null) {
throw new RuntimeException("分片任務(wù)不存");
}
ListPartsRequest listPartsRequest = new ListPartsRequest(task.getBucketName(), task.getObjectKey(), task.getUploadId());
PartListing partListing = amazonS3.listParts(listPartsRequest);
List<PartSummary> parts = partListing.getParts();
if (!task.getChunkNum().equals(parts.size())) {
// 已上傳分塊數(shù)量與記錄中的數(shù)量不對(duì)應(yīng)，不能合并分塊
throw new RuntimeException("分片缺失，請(qǐng)重新上傳");
}
CompleteMultipartUploadRequest completeMultipartUploadRequest = new CompleteMultipartUploadRequest()
.withUploadId(task.getUploadId())
.withKey(task.getObjectKey())
.withBucketName(task.getBucketName())
.withPartETags(parts.stream().map(partSummary -> new PartETag(partSummary.getPartNumber(), partSummary.getETag())).collect(Collectors.toList()));
CompleteMultipartUploadResult result = amazonS3.completeMultipartUpload(completeMultipartUploadRequest);
}
4、分片文件清理問(wèn)題
視頻上傳一半不上傳了，怎么清理碎片分片。
可以考慮在sys_upload_task表中新加一個(gè)status字段，表示是否合并分片，默認(rèn)為false，merge請(qǐng)求結(jié)束后變更為true，通過(guò)一個(gè)定時(shí)任務(wù)定期清理為status為false的記錄。另外MinIO自身對(duì)于臨時(shí)上傳的分片，會(huì)實(shí)施定時(shí)清理。
Demo地址
https://github.com/robinsyn/MinIO_Demo
來(lái)源：blog.csdn.net/weixin_44153131/
article/details/129249169
構(gòu)建高質(zhì)量的技術(shù)交流社群，歡迎從事編程開(kāi)發(fā)、技術(shù)招聘HR進(jìn)群，也歡迎大家分享自己公司的內(nèi)推信息，相互幫助，一起進(jìn)步！
文明發(fā)言，以
交流技術(shù)、職位內(nèi)推、行業(yè)探討為主
廣告人士勿入，切勿輕信私聊，防止被騙
評(píng)論
圖片
表情