0
0

minio分片上传

oss文件服务

一、前言

​ Minio是一个对象存储服务OSS(Object Storage Service)。是⼀种海量、安全、低成本、⾼可靠的云存储服务。本身的应用的并不复杂。

但是Minio的APi在对于大于5m的文件，自动采用了分片上传，它的分片上传我们无法得知上传的分片后的序号，也就是说，没上传一个分片，我们都需要自己去记录已上传分片的序号。这将导致一个文件一个文件分片5个，那么同样还需要调用5次后端接口去记录这5个分片的信息。这个无疑大大浪费了性能，且无法做到并发上传。

​ 因此基于Minio的javaAPI,我们采用另一种方案去替代。

二、初步流程：

image-20221107005827298

  1. 前端服务进行大文件分片处理，将分片信息传递给文件服务oss。
  2. oss通过redis和mysql检查分片信息是否已存在，若存在直接返回数据。
  3. 若不存在oss生成上传链接以及uploadID, 然后记录并返回所有分片的上传链接及uploadId。
  4. 前端服务直接请求Minio 服务器，并发上传分片。
  5. 所有分片上传完成后，使用uploadId 调用文件服务进行文件合并，oss同时更新上传状态。

三、具体实现

1 相关配置准备

1.1 数据库表设计

表名：file_record
字段名类型注释idbigint主键idfile_urltext上传分片的链接file_namevarchar文件名md5varcharMD5upload_idvarchar上传idis_uploadedint是否已上传total_chunksint分片总块数sizebigint文件大小（K）completed_partsint已完成片数created_atdatetime生成时间updated_atdatetime更新时间deleted_atdatetime删除时间（软删除）

1.2 minio集成配置

linux上部署minio，docker下载minio镜像

docker run -d -p 10000:10000 -p 11000:11000 --name minio2 -v ~/var/local/environment/data:/data -e "MINIO_ROOT_USER=root" -e "MINIO_ROOT_PASSWORD=minio123456" minio/minio server /data --console-address ":11000" --address ":10000"

配置说明：这里选择开放两个端口10000和11000。

–console-address ":11000"是将控制台的端口11000暴露

–address ":10000"是我们集成到项目中占用的端口。

nacos上配置oss-dev.yml

minio:url: http://192.168.137.129:10000accessKey: root
  secretKey: minio123456
  bucketName: test-bucket

minio控制台(可管理文件和bucket):

image-20221106231456674

项目依赖引入，这里使用的是当前的最新版8.3.5

<dependency><groupId>io.minio</groupId><artifactId>minio</artifactId><version>8.3.5</version></dependency>

2 后端实现

2.1 MinioProperty

MinioProperty类获取minio配置

importlombok.Data;importorg.springframework.boot.context.properties.ConfigurationProperties;importorg.springframework.context.annotation.Configuration;/**
 * 获取minio配置
 */@Data@Configuration@ConfigurationProperties(prefix ="minio")publicclassMinioProperty{/**
     * 服务地址
     */publicString url;/**
     * 用户名
     */publicString accessKey;/**
     * 密码
     */publicString secretKey;/**
     * 存储桶名称
     */publicString bucketName;/**
     *  如果是true，则用的是https而不是http,默认值是true
     */publicBoolean secure =false;}

2.2 MyMinioClient

MyMinioClient类继承MinioClient来暴露出父类方法供扩展使用。

importcom.google.common.collect.Multimap;importio.minio.CreateMultipartUploadResponse;importio.minio.ListPartsResponse;importio.minio.MinioClient;importio.minio.ObjectWriteResponse;importio.minio.errors.*;importio.minio.messages.Part;importjava.io.IOException;importjava.security.InvalidKeyException;importjava.security.NoSuchAlgorithmException;publicclassMyMinioClientextendsMinioClient{protectedMyMinioClient(MinioClient client){super(client);}/**
     * 创建分片上传请求
     * @param bucketName       存储桶
     * @param region           区域
     * @param objectName       对象名
     * @param headers          消息头
     * @param extraQueryParams 额外查询参数
     */@OverridepublicCreateMultipartUploadResponsecreateMultipartUpload(String bucketName,String region,String objectName,Multimap<String,String> headers,Multimap<String,String> extraQueryParams)throwsNoSuchAlgorithmException,InsufficientDataException,IOException,InvalidKeyException,ServerException,XmlParserException,ErrorResponseException,InternalException,InvalidResponseException{returnsuper.createMultipartUpload(bucketName, region, objectName, headers, extraQueryParams);}/**
     * 完成分片上传，执行合并文件
     * @param bucketName       存储桶
     * @param region           区域
     * @param objectName       对象名
     * @param uploadId         上传ID
     * @param parts            分片
     * @param extraHeaders     额外消息头
     * @param extraQueryParams 额外查询参数
     */@OverridepublicObjectWriteResponsecompleteMultipartUpload(String bucketName,String region,String objectName,String uploadId,Part[] parts,Multimap<String,String> extraHeaders,Multimap<String,String> extraQueryParams)throwsNoSuchAlgorithmException,InsufficientDataException,IOException,InvalidKeyException,ServerException,XmlParserException,ErrorResponseException,InternalException,InvalidResponseException{returnsuper.completeMultipartUpload(bucketName, region, objectName, uploadId, parts, extraHeaders, extraQueryParams);}/**
     * 查询分片数据
     * @param bucketName       存储桶
     * @param region           区域
     * @param objectName       对象名
     * @param uploadId         上传ID
     * @param extraHeaders     额外消息头
     * @param extraQueryParams 额外查询参数
     */publicListPartsResponselistMultipart(String bucketName,String region,String objectName,Integer maxParts,Integer partNumberMarker,String uploadId,Multimap<String,String> extraHeaders,Multimap<String,String> extraQueryParams)throwsNoSuchAlgorithmException,InsufficientDataException,IOException,InvalidKeyException,ServerException,XmlParserException,ErrorResponseException,InternalException,InvalidResponseException{returnsuper.listParts(bucketName, region, objectName, maxParts, partNumberMarker, uploadId, extraHeaders, extraQueryParams);}/**
     * 上传分片上传请求，返回uploadId
     * @param bucketName       存储桶
     * @param region           区域
     * @param objectName       对象名
     * @param headers           消息头
     * @param extraQueryParams 额外查询参数
     */publicCreateMultipartUploadResponseuploadId(String bucketName,String region,String objectName,Multimap<String,String> headers,Multimap<String,String> extraQueryParams)throwsNoSuchAlgorithmException,InsufficientDataException,IOException,InvalidKeyException,ServerException,XmlParserException,ErrorResponseException,InternalException,InvalidResponseException{returnsuper.createMultipartUpload(bucketName, region, objectName, headers, extraQueryParams);}/**
     * 返回临时带签名、过期时间为1天的PUT请求方式的访问URL
     * @param bucketName  桶名
     * @param filePath    Oss文件路径
     * @param queryParams 查询参数
     * @return 临时带签名、过期时间为1天的PUT请求方式的访问URL
     */@SneakyThrowspublicStringgetPresignedObjectUrl(String bucketName,String filePath,Map<String,String> queryParams){returnsuper.getPresignedObjectUrl(GetPresignedObjectUrlArgs.builder().method(Method.PUT).bucket(bucketName).object(filePath).expiry(1,TimeUnit.DAYS).extraQueryParams(queryParams).build());}}

2.3 MinioConfig

MinioConfig将MyMinioClient交给spring容器管理方便调用

importio.minio.MinioClient;importlombok.SneakyThrows;importorg.springframework.boot.autoconfigure.condition.ConditionalOnBean;importorg.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;@ConfigurationpublicclassMinoConfig{@Bean@SneakyThrows@ConditionalOnMissingBean(MyMinioClient.class)publicMyMinioClientminioClient(MinioProperty minioProperty){MinioClient minioClient =MinioClient.builder().endpoint(minioProperty.getUrl()).credentials(minioProperty.getAccessKey(), minioProperty.getSecretKey()).build();returnnewMyMinioClient(minioClient);}}

2.4 核心接口实现

2.4.1 createMultipartUpload方法

返回分片上传需要的签名数据URL及 uploadId。

1、fileMultipartDTO 是分片传输实体，包括:文件名fileName、分片数chunkSize、所属文件夹名bucketName。

2、检查redis中是否存在，若存在直接返回对应数据；若redis中不存在检查数据库中是否存在，若存在直接返回对应数据。

3、若redis、数据库中都不存在，处理生成分片。

4、调用minioAPI获得uploadId和签名url

5、相关数据在redis中缓存，并保存到数据库。返回给前端

publicMap<String,Object>createMultipartUpload(FileMultipartDTO fileMultipartDTO){
    log.info("fileMultipartDTO:{}", fileMultipartDTO);String fileName = fileMultipartDTO.getFileName();Integer chunkSize = fileMultipartDTO.getChunkSize();String bucketName = fileMultipartDTO.getBucketName();// 1. 根据文件名创建签名// 2. 获取uploadIdString contentType ="application/octet-stream";HashMultimap<String,String> headers =HashMultimap.create();
    headers.put("Content-Type", contentType);CreateMultipartUploadResponse response = minioClient.uploadId(bucketName,null, fileName,null,null);String uploadId = response.result().uploadId();Map<String,Object> result =newHashMap<>(3,1);String md5 =MD5Utils.encryptToMd5(fileName + chunkSize + bucketName);//查询是否已存在,若存在返回对应数据Map<String,Object> checkResult;//检查redis中是否已存在
    checkResult =checkIsExistInRedis(md5);if(checkResult !=null&&!checkResult.isEmpty()){
        log.info(">>>>>>>>>>>>redis中存在");return checkResult;}//检查数据库中是否已存在
    checkResult =checkIsExistInDatabase(md5);if(checkResult !=null&&!checkResult.isEmpty()){
        log.info(">>>>>>>>>>>>数据库中存在");//redis中补充
        checkResult.put("fileName", fileName);
        redisTemplate.opsForValue().set(RedisPrefixForKey.MINIO_KEY + md5, checkResult,1,TimeUnit.HOURS);return checkResult;}//redis中和数据库中都没有，需要生成
    result.put("uploadId", uploadId);// 3. 请求Minio 服务，获取每个分块带签名的上传URLMap<String,String> reqParams =newHashMap<>(3,1);
    reqParams.put("uploadId", uploadId);List<String> uploadUrlList =newArrayList<>();// 4. 循环分块数 从1开始for(int i =1; i <= chunkSize; i++){
        reqParams.put("partNumber",String.valueOf(i));// 获取URLString uploadUrl = minioCilent.getPresignedObjectUrl(bucketName, fileName, reqParams);// 添加到集合
        result.put("chunk_"+(i -1), uploadUrl);
        uploadUrlList.add(uploadUrl);}
    log.info(">>>>分片数据入redis和数据库");
    result.put("fileName", fileName);
    redisTemplate.opsForValue().set(RedisPrefixForKey.MINIO_KEY + md5, result,8,TimeUnit.HOURS);//入库String uploadUrls = JSON.toJSONString(uploadUrlList);FileRecord fileRecord =newFileRecord().setFileName(fileName).setFileUrl(uploadUrls).setUploadId(uploadId).setTotalChunks(chunkSize).setMd5(md5)//上传完成后，接收到合并请求时再改以下参数.setCompletedParts(0).setSize(0).setIsUploaded(0);
    fileRecordMapper.insert(fileRecord);return result;}
2.4.2 completeMultipartUpload方法

分片上传完后合并。

FileCompleteDTO是合并请求参数实体，包括：所属文件夹名bucketName、上传iduploadId、文件名objectName。

1、调用minioAPI获取该文件的所有分片partList。

2、将partList中part合并成parts。

3、调用minioAPI完成合并。

4、合并成功后修改数据库信息。

publicResultcompleteMultipartUpload(FileCompleteDTO fileObject){String bucketName = fileObject.getBucketName();String uploadId = fileObject.getUploadId();String objectName = fileObject.getObjectName();int completedParts;int size =0;try{Part[] parts =newPart[10000];ListPartsResponse partResult = minioClient.listMultipart(bucketName,null, objectName,1000,0, uploadId,null,null);List<Part> partList = partResult.result().partList();
        completedParts = partList.size();int partNumber =1;
        log.info("总片数：================"+ completedParts +"========");for(Part part : partList){
            parts[partNumber -1]=newPart(partNumber, part.etag());//etag就是分片信息
            partNumber++;
            size += part.partSize();}
        minioClient.completeMultipartUpload(bucketName,null, objectName, uploadId, parts,null,null);}catch(Exception e){
        e.printStackTrace();returnResult.fail("合并失败："+ e.getMessage());}//合并成功，入库//通过uploadId修改对应的记录行
    fileRecordMapper.updateStatusByUploadId(uploadId, completedParts, size);returnResult.success();}

3 前端实现

此处对前端实现不做重点描述，项目提供了一个demo，后续可做参考修改。
在这里插入图片描述

四、效果展示

当前端选择完文件后：

image-20221106234039847

image-20221106234221457

数据库中新增数据：

image-20221106234346147

redis中新增数据

image-20221106234426565

当前端点击开始上传按钮后，前端服务直接请求Minio服务

image-20221106234606478

完成上传后，数据库更新is_uploaded、Size、completed_parts

image-20221106234715043

查看minio控制台，可以看到该文件。

image-20221106234840310

标签： 数据库 java 服务器
本文转载自: https://blog.csdn.net/qq_42852943/article/details/127734905
版权归原作者 __Loren 所有， 如有侵权，请联系我们删除。
登录后发布评论

“minio分片上传”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

【网络】套接字编程——UDP通信

IDEA卡顿解决

解决 Git The remote end hung up unexpectedly 错误

基于Hadoop的汽车大数据分析系统设计与实现【爬虫、数据预处理、MapReduce、echarts、Flask】

（杭州大数据RD面经）字节、阿里、滴滴问题汇总

不吹不黑，客观理性深入讨论国产中间件

spring AOP详解

文章导航