devbox
Monodyee
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

springcloud+vue3实现分片上传，断点续传_vue3分片上传插件

作者：Monodyee | 2024-02-07 17:41:15

赞

踩

vue3分片上传插件

背景

        现在做一些应用级的服务平台时，有时会遇到用户需要上传一个较大的文件，但是上传失败后需要支持下次从失败的地方开始上传，这时候需要用到端点续传，分片上传的解决方案。本文介绍了一种策略来实现这个场景，涉及技术栈有 ng，springboot，vue，minio，amazonS3等框架和组件。

准备工作

1、ng设置文件大小

做这一步的目的是为了让文件分片后不会因为size而导致传不上去。

client_max_body_size 100m;  # 目前我们项目需要，这里设置的100m

  1. http {
  2.     include       /etc/nginx/mime.types;
  3.     default_type  application/octet-stream;
  4.  
  5.     log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
  6.                       '$status $body_bytes_sent "$http_referer" '
  7.                       '"$http_user_agent" "$http_x_forwarded_for"';
  8.  
  9.     access_log  /var/log/nginx/access.log  main;
  10.     # 在这里设置一下最大支持传输文件大小
  11.     client_max_body_size 100m;
  12.     sendfile        on;
  13.     #tcp_nopush     on;
  14.  
  15.     keepalive_timeout  65;
  16. ....
  17.  
  18.     }

2、springboot配置

  1. servlet:
  2.   multipart:
  3.     max-file-size: 100MB
  4.     # 最大支持请求大小
  5.     max-request-size: 500MB

接入

前端上传组件 UploadParallel.vue

  1. <script setup>
  2. import { UploadFilled } from '@element-plus/icons-vue'
  3.  
  4. import md5 from "../lib/md5";
  5. import { taskInfo, initTask, preSignUrl, merge } from '../lib/api';
  6. import {ElNotification} from "element-plus";
  7. import Queue from 'promise-queue-plus';
  8. import axios from 'axios'
  9. import { ref } from 'vue'
  10.  
  11. // 文件上传分块任务的队列（用于移除文件时，停止该文件的上传队列） key：fileUid value： queue object
  12. const fileUploadChunkQueue = ref({}).value
  13.  
  14. /**
  15.  * 获取一个上传任务，没有则初始化一个
  16.  */
  17. const getTaskInfo = async (file) => {
  18.     let task;
  19.     const identifier = await md5(file)
  20.     const { code, data, msg } = await taskInfo(identifier)
  21.     if (code === 200) {
  22.         task = data
  23.         if (!task || Object.keys(task).length === 0) {
  24.             const initTaskData = {
  25.                 identifier,
  26.                 fileName: file.name,
  27.                 totalSize: file.size,
  28.                 chunkSize: 5 * 1024 * 1024
  29.             }
  30.             const { code, data, msg } = await initTask(initTaskData)
  31.             if (code === 200) {
  32.                 task = data
  33.             } else {
  34.                 ElNotification.error({
  35.                     title: '文件上传错误',
  36.                     message: msg
  37.                 })
  38.             }
  39.         }
  40.     } else {
  41.         ElNotification.error({
  42.             title: '文件上传错误',
  43.             message: msg
  44.         })
  45.     }
  46.     return task
  47. }
  48.  
  49. /**
  50.  * 上传逻辑处理，如果文件已经上传完成（完成分块合并操作），则不会进入到此方法中
  51.  */
  52. const handleUpload = (file, taskRecord, options) => {
  53.  
  54.     let lastUploadedSize = 0; // 上次断点续传时上传的总大小
  55.     let uploadedSize = 0 // 已上传的大小
  56.     const totalSize = file.size || 0 // 文件总大小
  57.     let startMs = new Date().getTime(); // 开始上传的时间
  58.     const { exitPartList, chunkSize, chunkNum, fileIdentifier } = taskRecord
  59.  
  60.     // 获取从开始上传到现在的平均速度（byte/s）
  61.     const getSpeed = () => {
  62.         // 已上传的总大小 - 上次上传的总大小（断点续传）= 本次上传的总大小（byte）
  63.         const intervalSize = uploadedSize - lastUploadedSize
  64.         const nowMs = new Date().getTime()
  65.         // 时间间隔（s）
  66.         const intervalTime = (nowMs - startMs) / 1000
  67.         return intervalSize / intervalTime
  68.     }
  69.  
  70.     const uploadNext = async (partNumber) => {
  71.         const start = new Number(chunkSize) * (partNumber - 1)
  72.         const end = start + new Number(chunkSize)
  73.         const blob = file.slice(start, end)
  74.         const { code, data, msg } = await preSignUrl({ identifier: fileIdentifier, partNumber: partNumber} )
  75.         if (code === 200 && data) {
  76.             await axios.request({
  77.                 url: data,
  78.                 method: 'PUT',
  79.                 data: blob,
  80.                 headers: {'Content-Type': 'application/octet-stream'}
  81.             })
  82.             return Promise.resolve({ partNumber: partNumber, uploadedSize: blob.size })
  83.         }
  84.         return Promise.reject(`分片${partNumber}， 获取上传地址失败`)
  85.     }
  86.  
  87.     /**
  88.      * 更新上传进度
  89.      * @param increment 为已上传的进度增加的字节量
  90.      */
  91.     const updateProcess = (increment) => {
  92.         increment = new Number(increment)
  93.         const { onProgress } = options
  94.         let factor = 1000; // 每次增加1000 byte
  95.         let from = 0;
  96.         // 通过循环一点一点的增加进度
  97.         while (from <= increment) {
  98.             from += factor
  99.             uploadedSize += factor
  100.             const percent = Math.round(uploadedSize / totalSize * 100).toFixed(2);
  101.             onProgress({percent: percent})
  102.         }
  103.  
  104.         const speed = getSpeed();
  105.         const remainingTime = speed != 0 ? Math.ceil((totalSize - uploadedSize) / speed) + 's' : '未知'
  106.         console.log('剩余大小：', (totalSize - uploadedSize) / 1024 / 1024, 'mb');
  107.         console.log('当前速度：', (speed / 1024 / 1024).toFixed(2), 'mbps');
  108.         console.log('预计完成：', remainingTime);
  109.     }
  110.  
  111.  
  112.     return new Promise(resolve => {
  113.         const failArr = [];
  114.         const queue = Queue(5, {
  115.             "retry": 3,               //Number of retries
  116.             "retryIsJump": false,     //retry now?
  117.             "workReject": function(reason,queue){
  118.                 failArr.push(reason)
  119.             },
  120.             "queueEnd": function(queue){
  121.                 resolve(failArr);
  122.             }
  123.         })
  124.         fileUploadChunkQueue[file.uid] = queue
  125.         for (let partNumber = 1; partNumber <= chunkNum; partNumber++) {
  126.             const exitPart = (exitPartList || []).find(exitPart => exitPart.partNumber == partNumber)
  127.             if (exitPart) {
  128.                 // 分片已上传完成，累计到上传完成的总额中,同时记录一下上次断点上传的大小，用于计算上传速度
  129.                 lastUploadedSize += new Number(exitPart.size)
  130.                 updateProcess(exitPart.size)
  131.             } else {
  132.                 queue.push(() => uploadNext(partNumber).then(res => {
  133.                     // 单片文件上传完成再更新上传进度
  134.                     updateProcess(res.uploadedSize)
  135.                 }))
  136.             }
  137.         }
  138.         if (queue.getLength() == 0) {
  139.             // 所有分片都上传完，但未合并，直接return出去，进行合并操作
  140.             resolve(failArr);
  141.             return;
  142.         }
  143.         queue.start()
  144.     })
  145. }
  146.  
  147. /**
  148.  * el-upload 自定义上传方法入口
  149.  */
  150. const handleHttpRequest = async (options) => {
  151.     const file = options.file
  152.     const task = await getTaskInfo(file)
  153.     if (task) {
  154.         const { finished, path, taskRecord } = task
  155.         const { fileIdentifier: identifier } = taskRecord
  156.         if (finished) {
  157.             return path
  158.         } else {
  159.             const errorList = await handleUpload(file, taskRecord, options)
  160.             if (errorList.length > 0) {
  161.                 ElNotification.error({
  162.                     title: '文件上传错误',
  163.                     message: '部分分片上次失败，请尝试重新上传文件'
  164.                 })
  165.                 return;
  166.             }
  167.             const { code, data, msg } = await merge(identifier)
  168.             if (code === 200) {
  169.                 return path;
  170.             } else {
  171.                 ElNotification.error({
  172.                     title: '文件上传错误',
  173.                     message: msg
  174.                 })
  175.             }
  176.         }
  177.     } else {
  178.         ElNotification.error({
  179.             title: '文件上传错误',
  180.             message: '获取上传任务失败'
  181.         })
  182.     }
  183. }
  184.  
  185. /**
  186.  * 移除文件列表中的文件
  187.  * 如果文件存在上传队列任务对象，则停止该队列的任务
  188.  */
  189. const handleRemoveFile = (uploadFile, uploadFiles) => {
  190.     const queueObject = fileUploadChunkQueue[uploadFile.uid]
  191.     if (queueObject) {
  192.         queueObject.stop()
  193.         fileUploadChunkQueue[uploadFile.uid] = undefined
  194.     }
  195. }
  196.  
  197. </script>
  198. <template>
  199.     <el-card style="width: 80%; margin: 80px auto" header="文件分片上传">
  200.         <el-upload
  201.             class="upload-demo"
  202.             drag
  203.             action="/"
  204.             multiple
  205.             :http-request="handleHttpRequest"
  206.             :on-remove="handleRemoveFile">
  207.             <el-icon class="el-icon--upload"><upload-filled /></el-icon>
  208.             <div class="el-upload__text">
  209.                 请拖拽文件到此处或 <em>点击此处上传</em>
  210.             </div>
  211.         </el-upload>
  212.     </el-card>
  213.  
  214. </template>

前端api文件

  1. import axios from 'axios'
  2. import axiosExtra from 'axios-extra'
  3.  
  4. const baseUrl = 'http://172.16.10.74:10003/dsj-file'
  5.  
  6. const http = axios.create({
  7.     baseURL: baseUrl,
  8.     headers: {
  9.         'Dsj-Auth':'bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ0ZW5hbnRfaWQiOiIwMDAwMDAiLCJkc2pJZCI6ImFqZGhxaWRicSIsImRpc3RyaWN0Q29kZSI6IjQyMTEwMDAwMDAwMCIsInVzZXJfbmFtZSI6ImFkbWluIiwic29jaWFsQWNjb3VudElkIjpudWxsLCJyZWFsX25hbWUiOiLotoXnuqfnrqHnkIblkZgiLCJjbGllbnRfaWQiOiJzd29yZCIsInJvbGVfaWQiOiIxNDkzODIwMjY0OTY4OTk0ODE3IiwiaXNEZWZhdWx0UGFzc3dvcmQiOmZhbHNlLCJpZGVudGl0eVR5cGUiOiIxIiwic2NvcGUiOlsiYWxsIl0sImRlcHRTQ0NvZGUiOiIxMTQyMTEwMDc0NDYyMTAxMDgiLCJleHAiOjE2OTU4NjczMDUsImp0aSI6IjEwMmE3YzhhLTdiMWYtNDU4NC04ZWJjLWZiYmUyZTQyYmUzNCIsImlkZW50aXR5RHluYURhdGEiOnt9LCJhdmF0YXIiOiIxNjM1MTA4Nzk5MzQ3OTU3NzYyIiwiYXV0aG9yaXRpZXMiOlsiYWRtaW5pc3RyYXRvciJdLCJyb2xlX25hbWUiOiJhZG1pbmlzdHJhdG9yIiwiYWNjb3VudElkIjoiMTQ1Mzk5MzI5MjAxMDM5OTgxMSIsImxpY2Vuc2UiOiJwb3dlcmVkIGJ5IGRzaiIsInBvc3RfaWQiOiIxNTYwMTQ1MzUwMDY5NjY5ODg5IiwidXNlcl9pZCI6IjE1NDczOTY0ODAwNzkyMzcxMjEiLCJwaG9uZSI6IjE1ODcxOTI2MDczIiwibmlja19uYW1lIjoi6LaF57qn566h55CG5ZGYIiwiZGVwdF9pZCI6IjExNDIxMTAwNzQ0NjIxMDEwOCIsImFjY291bnQiOiJhZG1pbiIsImRlcHRDb2RlIjoiMTE0MjExMDA3NDQ2MjEwMTA4In0.5r85ctPgWxNarVdF9kwTNoub7IqQM6RxTHYIU-ajxio'
  10.     }
  11.  
  12. })
  13.  
  14. const httpExtra = axiosExtra.create({
  15.     maxConcurrent: 5, //并发为1
  16.     queueOptions: {
  17.         retry: 3, //请求失败时,最多会重试3次
  18.         retryIsJump: false //是否立即重试, 否则将在请求队列尾部插入重试请求
  19.     }
  20. })
  21.  
  22. http.interceptors.response.use(response => {
  23.     return response.data
  24. })
  25.  
  26. /**
  27.  * 根据文件的md5获取未上传完的任务
  28.  * @param identifier 文件md5
  29.  * @returns {Promise<AxiosResponse<any>>}
  30.  */
  31. const taskInfo = (identifier) => {
  32.     return http.get(`/parallel-upload/${identifier}`)
  33. }
  34.  
  35. /**
  36.  * 初始化一个分片上传任务
  37.  * @param identifier 文件md5
  38.  * @param fileName 文件名称
  39.  * @param totalSize 文件大小
  40.  * @param chunkSize 分块大小
  41.  * @returns {Promise<AxiosResponse<any>>}
  42.  */
  43. const initTask = ({identifier, fileName, totalSize, chunkSize}) => {
  44.     return http.post('/parallel-upload/init-task', {identifier, fileName, totalSize, chunkSize})
  45. }
  46.  
  47. /**
  48.  * 获取预签名分片上传地址
  49.  * @param identifier 文件md5
  50.  * @param partNumber 分片编号
  51.  * @returns {Promise<AxiosResponse<any>>}
  52.  */
  53. const preSignUrl = ({identifier, partNumber}) => {
  54.     return http.get(`/parallel-upload/${identifier}/${partNumber}`)
  55. }
  56.  
  57. /**
  58.  * 合并分片
  59.  * @param identifier
  60.  * @returns {Promise<AxiosResponse<any>>}
  61.  */
  62. const merge = (identifier) => {
  63.     return http.post(`/parallel-upload/merge/${identifier}`)
  64. }
  65.  
  66. export {
  67.     taskInfo,
  68.     initTask,
  69.     preSignUrl,
  70.     merge,
  71.     httpExtra
  72. }

文件服务minio配置

  1.   minio:
  2.     endpoint: http://172.16.10.74:9000
  3.     address: http://172.16.10.74
  4.     port: 9000
  5.     secure: false
  6.     access-key: minioadmin
  7.     secret-key: XXXXXXXXXX
  8.     bucket-name: gpd
  9.     internet-address: http://XXXXXXXXX:9000

MinioProperties.java

  1. package com.dsj.prod.file.biz.properties;
  2.  
  3. import lombok.Data;
  4. import org.springframework.boot.context.properties.ConfigurationProperties;
  5. import org.springframework.cloud.context.config.annotation.RefreshScope;
  6. import org.springframework.stereotype.Component;
  7.  
  8. import java.util.List;
  9.  
  10.  
  11. @Data
  12. @RefreshScope
  13. @Component
  14. @ConfigurationProperties(prefix = "dsj.minio")
  15. public class MinioProperties {
  16.  
  17. 	/**
  18. 	 * The constant endpoint.
  19. 	 */
  20. 	public String endpoint;
  21. 	/**
  22. 	 * The constant address.
  23. 	 */
  24. 	public String address;
  25.  
  26. 	/**
  27. 	 * The constant port.
  28. 	 */
  29. 	public String port;
  30.  
  31. 	/**
  32. 	 * The constant accessKey.
  33. 	 */
  34. 	public String accessKey;
  35.  
  36. 	/**
  37. 	 * The constant secretKey.
  38. 	 */
  39. 	public String secretKey;
  40.  
  41. 	/**
  42. 	 * The constant bucketName.
  43. 	 */
  44. 	public String bucketName;
  45.  
  46. 	/**
  47. 	 * The constant internetAddress.
  48. 	 */
  49. 	public String internetAddress;
  50.  
  51. 	/**
  52. 	 * The Limit file extension.
  53. 	 * doc docx xls xlsx 图片() pdf
  54. 	 */
  55. 	public List<String> limitFileExtension;
  56.  
  57. }

控制器 ParallelUploadController.java

  1. package com.dsj.prod.file.biz.controller;
  2.  
  3.  
  4. import com.dsj.plf.arch.tool.api.R;
  5. import com.dsj.prod.file.api.dto.parallelUpload.InitTaskParam;
  6. import com.dsj.prod.file.api.dto.parallelUpload.TaskInfoDTO;
  7. import com.dsj.prod.file.api.entity.ParallelUploadTask;
  8. import com.dsj.prod.file.biz.service.ParallelUploadTaskService;
  9. import com.github.xiaoymin.knife4j.annotations.ApiOperationSupport;
  10. import io.swagger.annotations.Api;
  11. import io.swagger.annotations.ApiOperation;
  12. import org.springframework.web.bind.annotation.*;
  13.  
  14. import javax.annotation.Resource;
  15. import javax.validation.Valid;
  16. import java.util.HashMap;
  17. import java.util.Map;
  18.  
  19.  
  20.  
  21. @Api(value = "文件分片上传接口", tags = "文件分片上传接口")
  22. @RestController
  23. @RequestMapping("/parallel-upload")
  24. public class ParallelUploadController {
  25. 	@Resource
  26. 	private ParallelUploadTaskService sysUploadTaskService;
  27.  
  28.  
  29. 	@ApiOperationSupport(order = 1)
  30. 	@ApiOperation(value = "获取上传进度", notes = "传入 identifier：文件md5")
  31. 	@GetMapping("/{identifier}")
  32. 	public R<TaskInfoDTO> taskInfo(@PathVariable("identifier") String identifier) {
  33. 		TaskInfoDTO result = sysUploadTaskService.getTaskInfo(identifier);
  34. 		return R.data(result);
  35. 	}
  36.  
  37. 	/**
  38. 	 * 创建一个上传任务
  39. 	 *
  40. 	 * @param param the param
  41. 	 * @return result
  42. 	 */
  43. 	@ApiOperationSupport(order = 2)
  44. 	@ApiOperation(value = "创建一个上传任务", notes = "传入 param")
  45. 	@PostMapping("/init-task")
  46. 	public R<TaskInfoDTO> initTask(@Valid @RequestBody InitTaskParam param) {
  47. 		return R.data(sysUploadTaskService.initTask(param));
  48. 	}
  49.  
  50. 	@ApiOperationSupport(order = 3)
  51. 	@ApiOperation(value = "获取每个分片的预签名上传地址", notes = "传入 identifier文件md5,partNumber分片序号")
  52. 	@GetMapping("/{identifier}/{partNumber}")
  53. 	public R preSignUploadUrl(@PathVariable("identifier") String identifier, @PathVariable("partNumber") Integer partNumber) {
  54. 		ParallelUploadTask task = sysUploadTaskService.getByIdentifier(identifier);
  55. 		if (task == null) {
  56. 			return R.fail("分片任务不存在");
  57. 		}
  58. 		Map<String, String> params = new HashMap<>();
  59. 		params.put("partNumber", partNumber.toString());
  60. 		params.put("uploadId", task.getUploadId());
  61. 		return R.data(sysUploadTaskService.genPreSignUploadUrl(task.getBucketName(), task.getObjectKey(), params));
  62. 	}
  63.  
  64. 	@ApiOperationSupport(order = 4)
  65. 	@ApiOperation(value = "合并分片", notes = "传入 identifier文件md5")
  66. 	@PostMapping("/merge/{identifier}")
  67. 	public R merge(@PathVariable("identifier") String identifier) {
  68. 		sysUploadTaskService.merge(identifier);
  69. 		return R.success("合并成功");
  70. 	}
  71.  
  72. }

服务层 ParallelUploadTaskService.java

  1. package com.dsj.prod.file.biz.service;
  2.  
  3. import com.baomidou.mybatisplus.extension.service.IService;
  4. import com.dsj.prod.file.api.dto.parallelUpload.InitTaskParam;
  5. import com.dsj.prod.file.api.dto.parallelUpload.TaskInfoDTO;
  6. import com.dsj.prod.file.api.entity.ParallelUploadTask;
  7.  
  8. import java.util.Map;
  9.  
  10. /**
  11.  * 分片上传-分片任务记录(ParallelUploadTask)表服务接口
  12.  *
  13.  * @since 2022-08-22 17:47:30
  14.  */
  15. public interface ParallelUploadTaskService extends IService<ParallelUploadTask> {
  16.  
  17.     /**
  18.      * 根据md5标识获取分片上传任务
  19.       * @param identifier
  20.      * @return
  21.      */
  22.     ParallelUploadTask getByIdentifier (String identifier);
  23.  
  24.     /**
  25.      * 初始化一个任务
  26.      */
  27.     TaskInfoDTO initTask (InitTaskParam param);
  28.  
  29.     /**
  30.      * 获取文件地址
  31.      * @param bucket
  32.      * @param objectKey
  33.      * @return
  34.      */
  35.     String getPath (String bucket, String objectKey);
  36.  
  37.     /**
  38.      * 获取上传进度
  39.      * @param identifier
  40.      * @return
  41.      */
  42.     TaskInfoDTO getTaskInfo (String identifier);
  43.  
  44.     /**
  45.      * 生成预签名上传url
  46.      * @param bucket 桶名
  47.      * @param objectKey 对象的key
  48.      * @param params 额外的参数
  49.      * @return
  50.      */
  51.     String genPreSignUploadUrl (String bucket, String objectKey, Map<String, String> params);
  52.  
  53.     /**
  54.      * 合并分片
  55.      * @param identifier
  56.      */
  57.     void merge (String identifier);
  58. }

实现层 ParallelUploadTaskServiceImpl 

  1. package com.dsj.prod.file.biz.service.impl;
  2.  
  3. import cn.hutool.core.date.DateUtil;
  4. import cn.hutool.core.util.IdUtil;
  5. import cn.hutool.core.util.StrUtil;
  6. import com.amazonaws.HttpMethod;
  7. import com.amazonaws.services.s3.AmazonS3;
  8. import com.amazonaws.services.s3.model.*;
  9. import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
  10. import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
  11. import com.dsj.prod.file.api.constants.MinioConstant;
  12. import com.dsj.prod.file.api.dto.parallelUpload.InitTaskParam;
  13. import com.dsj.prod.file.api.dto.parallelUpload.TaskInfoDTO;
  14. import com.dsj.prod.file.api.dto.parallelUpload.TaskRecordDTO;
  15. import com.dsj.prod.file.api.entity.ParallelUploadTask;
  16. import com.dsj.prod.file.biz.mapper.ParallelUploadMapper;
  17. import com.dsj.prod.file.biz.properties.MinioProperties;
  18. import com.dsj.prod.file.biz.service.ParallelUploadTaskService;
  19. import lombok.extern.slf4j.Slf4j;
  20. import org.springframework.http.MediaType;
  21. import org.springframework.http.MediaTypeFactory;
  22. import org.springframework.stereotype.Service;
  23.  
  24. import javax.annotation.Resource;
  25. import java.net.URL;
  26. import java.util.Date;
  27. import java.util.List;
  28. import java.util.Map;
  29. import java.util.stream.Collectors;
  30.  
  31. /**
  32.  * 分片上传-分片任务记录(ParallelUploadTask)表服务实现类
  33.  *
  34.  * @since 2022-08-22 17:47:31
  35.  */
  36. @Slf4j
  37. @Service("sysUploadTaskService")
  38. public class ParallelUploadTaskServiceImpl extends ServiceImpl<ParallelUploadMapper, ParallelUploadTask> implements ParallelUploadTaskService {
  39.  
  40. 	@Resource
  41. 	private AmazonS3 amazonS3;
  42.  
  43. 	@Resource
  44. 	private MinioProperties minioProperties;
  45.  
  46. 	@Resource
  47. 	private ParallelUploadMapper sysUploadTaskMapper;
  48.  
  49. 	@Override
  50. 	public ParallelUploadTask getByIdentifier(String identifier) {
  51. 		return sysUploadTaskMapper.selectOne(new QueryWrapper<ParallelUploadTask>().lambda().eq(ParallelUploadTask::getFileIdentifier, identifier));
  52. 	}
  53.  
  54.  
  55. 	@Override
  56. 	public TaskInfoDTO initTask(InitTaskParam param) {
  57.  
  58. 		Date currentDate = new Date();
  59. 		String bucketName = minioProperties.getBucketName();
  60. 		String fileName = param.getFileName();
  61. 		String suffix = fileName.substring(fileName.lastIndexOf(".") + 1);
  62. 		String key = StrUtil.format("{}/{}.{}", DateUtil.format(currentDate, "YYYY-MM-dd"), IdUtil.randomUUID(), suffix);
  63. 		String contentType = MediaTypeFactory.getMediaType(key).orElse(MediaType.APPLICATION_OCTET_STREAM).toString();
  64. 		ObjectMetadata objectMetadata = new ObjectMetadata();
  65. 		objectMetadata.setContentType(contentType);
  66. 		InitiateMultipartUploadResult initiateMultipartUploadResult = amazonS3.initiateMultipartUpload(new InitiateMultipartUploadRequest(bucketName, key)
  67. 			.withObjectMetadata(objectMetadata));
  68. 		String uploadId = initiateMultipartUploadResult.getUploadId();
  69.  
  70. 		ParallelUploadTask task = new ParallelUploadTask();
  71. 		int chunkNum = (int) Math.ceil(param.getTotalSize() * 1.0 / param.getChunkSize());
  72. 		task.setBucketName(minioProperties.getBucketName())
  73. 			.setChunkNum(chunkNum)
  74. 			.setChunkSize(param.getChunkSize())
  75. 			.setTotalSize(param.getTotalSize())
  76. 			.setFileIdentifier(param.getIdentifier())
  77. 			.setFileName(fileName)
  78. 			.setObjectKey(key)
  79. 			.setUploadId(uploadId);
  80. 		sysUploadTaskMapper.insert(task);
  81. 		return new TaskInfoDTO().setFinished(false).setTaskRecord(TaskRecordDTO.convertFromEntity(task)).setPath(getPath(bucketName, key));
  82. 	}
  83.  
  84. 	@Override
  85. 	public String getPath(String bucket, String objectKey) {
  86. 		return StrUtil.format("{}/{}/{}", minioProperties.getEndpoint(), bucket, objectKey);
  87. 	}
  88.  
  89. 	@Override
  90. 	public TaskInfoDTO getTaskInfo(String identifier) {
  91. 		ParallelUploadTask task = getByIdentifier(identifier);
  92. 		if (task == null) {
  93. 			return null;
  94. 		}
  95. 		TaskInfoDTO result = new TaskInfoDTO().setFinished(true).setTaskRecord(TaskRecordDTO.convertFromEntity(task)).setPath(getPath(task.getBucketName(), task.getObjectKey()));
  96.  
  97. 		boolean doesObjectExist = amazonS3.doesObjectExist(task.getBucketName(), task.getObjectKey());
  98. 		if (!doesObjectExist) {
  99. 			// 未上传完，返回已上传的分片
  100. 			ListPartsRequest listPartsRequest = new ListPartsRequest(task.getBucketName(), task.getObjectKey(), task.getUploadId());
  101. 			PartListing partListing = amazonS3.listParts(listPartsRequest);
  102. 			result.setFinished(false).getTaskRecord().setExitPartList(partListing.getParts());
  103. 		}
  104. 		return result;
  105. 	}
  106.  
  107. 	@Override
  108. 	public String genPreSignUploadUrl(String bucket, String objectKey, Map<String, String> params) {
  109. 		Date currentDate = new Date();
  110. 		Date expireDate = DateUtil.offsetMillisecond(currentDate, MinioConstant.PRE_SIGN_URL_EXPIRE.intValue());
  111. 		GeneratePresignedUrlRequest request = new GeneratePresignedUrlRequest(bucket, objectKey)
  112. 			.withExpiration(expireDate).withMethod(HttpMethod.PUT);
  113. 		if (params != null) {
  114. 			params.forEach(request::addRequestParameter);
  115. 		}
  116. 		URL preSignedUrl = amazonS3.generatePresignedUrl(request);
  117. 		return preSignedUrl.toString();
  118. 	}
  119.  
  120. 	@Override
  121. 	public void merge(String identifier) {
  122. 		ParallelUploadTask task = getByIdentifier(identifier);
  123. 		if (task == null) {
  124. 			log.error("分片任务不存在，任务id：{}", identifier);
  125. 			throw new RuntimeException("分片任务不存在");
  126. 		}
  127.  
  128. 		log.info("开始合并分片，任务id：{}", task.getId());
  129. 		ListPartsRequest listPartsRequest = new ListPartsRequest(task.getBucketName(), task.getObjectKey(), task.getUploadId());
  130. 		PartListing partListing = amazonS3.listParts(listPartsRequest);
  131. 		List<PartSummary> parts = partListing.getParts();
  132. 		if (!task.getChunkNum().equals(parts.size())) {
  133. 			// 已上传分块数量与记录中的数量不对应，不能合并分块
  134. 			log.error("分片缺失，任务id：{}，已上传分块数量：{}，记录中的数量：{}", task.getId(), parts.size(), task.getChunkNum());
  135. 			throw new RuntimeException("分片缺失，请重新上传");
  136. 		}
  137. 		CompleteMultipartUploadRequest completeMultipartUploadRequest = new CompleteMultipartUploadRequest()
  138. 			.withUploadId(task.getUploadId())
  139. 			.withKey(task.getObjectKey())
  140. 			.withBucketName(task.getBucketName())
  141. 			.withPartETags(parts.stream().map(partSummary -> new PartETag(partSummary.getPartNumber(), partSummary.getETag())).collect(Collectors.toList()));
  142. 		CompleteMultipartUploadResult result = amazonS3.completeMultipartUpload(completeMultipartUploadRequest);
  143. 		log.info("合并分片完成，返回结果：{}", result);
  144. 	}
  145. }

亚马逊S3工具类AmazonS3Config.java

  1. package com.dsj.prod.file.biz.config;
  2.  
  3. import com.amazonaws.ClientConfiguration;
  4. import com.amazonaws.Protocol;
  5. import com.amazonaws.auth.AWSCredentials;
  6. import com.amazonaws.auth.AWSStaticCredentialsProvider;
  7. import com.amazonaws.auth.BasicAWSCredentials;
  8. import com.amazonaws.client.builder.AwsClientBuilder;
  9. import com.amazonaws.regions.Regions;
  10. import com.amazonaws.services.s3.AmazonS3;
  11. import com.amazonaws.services.s3.AmazonS3ClientBuilder;
  12. import com.dsj.prod.file.biz.properties.MinioProperties;
  13. import org.springframework.context.annotation.Bean;
  14. import org.springframework.context.annotation.Configuration;
  15.  
  16. import javax.annotation.Resource;
  17.  
  18. @Configuration
  19. public class AmazonS3Config {
  20.  
  21.     @Resource
  22.     private MinioProperties minioProperties;
  23.  
  24.     @Bean(name = "amazonS3Client")
  25.     public AmazonS3 amazonS3Client () {
  26.         ClientConfiguration config = new ClientConfiguration();
  27.         config.setProtocol(Protocol.HTTP);
  28.         config.setConnectionTimeout(60000);
  29.         config.setUseExpectContinue(true);
  30.         AWSCredentials credentials = new BasicAWSCredentials(minioProperties.getAccessKey(), minioProperties.getSecretKey());
  31.         AwsClientBuilder.EndpointConfiguration end_point = new AwsClientBuilder.EndpointConfiguration(minioProperties.getEndpoint(), Regions.CN_NORTH_1.name());
  32.         AmazonS3 amazonS3 = AmazonS3ClientBuilder.standard()
  33.                 .withClientConfiguration(config)
  34.                 .withCredentials(new AWSStaticCredentialsProvider(credentials))
  35.                 .withEndpointConfiguration(end_point)
  36.                 .withPathStyleAccessEnabled(true).build();
  37.         return amazonS3;
  38.     }
  39.  
  40. }

整体效果预览

大致流程。

1、初始化一个任务，获取预期链接。（会通过md5校验，来判断OSS中是否有分片或整个文件。存在断点分片则从最近分片开始上传，存在整个文件则直接返回链接。）

2、文件切片上传

3、合并分片

4、完成并提示合并成功

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Monodyee/article/detail/66616
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号