Minio文件分片上传实现

资源准备
MacM1Pro 安装Parallels19.1.0请参考 https://blog.csdn.net/qq_41594280/article/details/135420241
MacM1Pro Parallels安装CentOS7.9请参考 https://blog.csdn.net/qq_41594280/article/details/135420461
部署Minio和整合SpringBoot请参考 https://blog.csdn.net/qq_41594280/article/details/135613722

一、准备表结构

1.1 文件上传信息表

CREATE TABLE minio_file_upload_info(
	`id` BIGINT(20) PRIMARY KEY AUTO_INCREMENT NOT NULL COMMENT '自增主键',
	`file_name` VARCHAR(255) NOT NULL DEFAULT '' COMMENT '文件名称',
	`file_md5` VARCHAR(128) NOT NULL DEFAULT '' COMMENT '文件MD5',
	`upload_id` VARCHAR(128) NOT NULL DEFAULT '' COMMENT '文件上传Id',
 	`file_url` VARCHAR(255) NOT NULL DEFAULT '' COMMENT '文件路径',
 	`total_chunk` INT(10) NOT NULL DEFAULT 0 COMMENT '文件总分块数',
 	`file_status` VARCHAR(50) NOT NULL DEFAULT '' COMMENT '文件状态',
 	`update_time` DATETIME DEFAULT NULL COMMENT '修改时间'
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='文件上传信息';

1.2 分块上传信息表

CREATE TABLE minio_chunk_upload_info(
	`id` BIGINT(20) PRIMARY KEY AUTO_INCREMENT NOT NULL COMMENT '自增主键',
	`chunk_number` INT(10) NOT NULL DEFAULT 0 COMMENT '文件分片号',
	`file_md5` VARCHAR(128) NOT NULL DEFAULT '' COMMENT '文件MD5',
	`upload_id` VARCHAR(128) NOT NULL DEFAULT '' COMMENT '文件上传Id',
 	`chunk_upload_url` VARCHAR(1000) NOT NULL DEFAULT '' COMMENT '文件分片路径',
 	`expiry_time` DATETIME DEFAULT NULL COMMENT '失效时间'
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='文件分片信息';

二、Minio文件相关基操

2.1 Entity

/**
 * 文件上传信息
 */
@TableName(schema = "minio_demo", value = "minio_file_upload_info")
@Data
public class MinioFileUploadInfo {

    /**
     * 自增ID
     */
    @TableId(type = IdType.AUTO)
    private Long id;

    /**
     * 文件名称
     */
    private String fileName;

    /**
     * 文件Md5值
     */
    private String fileMd5;

    /**
     * 文件上传ID
     */
    private String uploadId;

    /**
     * 文件路径
     */
    private String fileUrl;

    /**
     * 总分块数
     */
    private Integer totalChunk;

    /**
     * 文件上传状态
     */
    private String fileStatus;

    /**
     * 修改时间
     */
    private Date updateTime;

}

/**
 * 文件分片信息
 */
@TableName(schema = "minio_demo", value = "minio_chunk_upload_info")
@Data
public class MinioFileChunkUploadInfo implements Serializable {

    /**
     * 自增ID
     */
    @TableId(type = IdType.AUTO)
    private Long id;

    /**
     * 文件Md5值
     */
    private String fileMd5;

    /**
     * 上传ID
     */
    private String uploadId;

    /**
     * 文件块号
     */
    private Integer chunkNumber;

    /**
     * 文件块上传URL
     */
    private String chunkUploadUrl;

    /**
     * 过期时间
     */
    private LocalDateTime expiryTime;

}

2.2 Mapper

public interface MinioFileUploadInfoMapper extends MyBaseMapper<MinioFileUploadInfo> {
}

public interface MinioFileChunkUploadInfoMapper extends MyBaseMapper<MinioFileChunkUploadInfo> {
}

2.3 Service

public interface MinioFileUploadInfoService extends IService<MinioFileUploadInfo> {

    /**
     * 根据文件 md5 查询
     *
     * @param fileMd5 文件 md5
     */
    MinioFileUploadInfoDTO getByFileMd5(String fileMd5);

    /**
     * 保存
     *
     * @param param 参数对象
     */
    MinioFileUploadInfoDTO saveMinioFileUploadInfo(MinioFileUploadInfoParam param);

    /**
     * 修改文件状态
     *
     * @param param 参数对象
     */
    int updateFileStatusByFileMd5(MinioFileUploadInfoParam param);

}

@Service
public class MinioFileUploadInfoServiceImpl
        extends ServiceImpl<MinioFileUploadInfoMapper, MinioFileUploadInfo>
        implements MinioFileUploadInfoService {

    @Override
    public MinioFileUploadInfoDTO getByFileMd5(String fileMd5) {
        MinioFileUploadInfo minioFileUploadInfo = this.baseMapper.selectOne(
                new LambdaQueryWrapper<MinioFileUploadInfo>()
                        .eq(MinioFileUploadInfo::getFileMd5, fileMd5));
        if (null == minioFileUploadInfo) {
            return null;
        }
        return ExtBeanUtils.doToDto(minioFileUploadInfo, MinioFileUploadInfoDTO.class);
    }

    @Override
    public MinioFileUploadInfoDTO saveMinioFileUploadInfo(MinioFileUploadInfoParam param) {
        MinioFileUploadInfo minioFileUploadInfo;
        if (null == param.getId()) {
            minioFileUploadInfo = new MinioFileUploadInfo();
        } else {
            minioFileUploadInfo = this.baseMapper.selectById(param.getId());
            if (null == minioFileUploadInfo) {
                throw new MinioDemoException(MinioDemoExceptionTypes.DATA_NOT_EXISTED);
            }
            minioFileUploadInfo.setUpdateTime(new Date());
        }
        BeanUtils.copyProperties(param, minioFileUploadInfo, "id");
        int result;
        if (null == param.getId()) {
            result = this.baseMapper.insert(minioFileUploadInfo);
        } else {
            result = this.baseMapper.updateById(minioFileUploadInfo);
        }
        if (result == 0) {
            throw new MinioDemoException(MinioDemoExceptionTypes.USER_OPERATE_FAILED);
        }
        return ExtBeanUtils.doToDto(minioFileUploadInfo, MinioFileUploadInfoDTO.class);
    }

    @Override
    public int updateFileStatusByFileMd5(MinioFileUploadInfoParam param) {
        MinioFileUploadInfo minioFileUploadInfo = this.baseMapper.selectOne(
                new LambdaQueryWrapper<MinioFileUploadInfo>()
                        .eq(MinioFileUploadInfo::getFileMd5, param.getFileMd5()));
        if (null == minioFileUploadInfo) {
            throw new MinioDemoException(MinioDemoExceptionTypes.DATA_NOT_EXISTED);
        }
        minioFileUploadInfo.setFileStatus(param.getFileStatus());
        minioFileUploadInfo.setFileUrl(param.getFileUrl());
        return this.baseMapper.updateById(minioFileUploadInfo);
    }
}

public interface MinioFileChunkUploadInfoService extends IService<MinioFileChunkUploadInfo> {

    boolean saveMinioFileChunkUploadInfo(MinioFileChunkUploadInfoParam chunkUploadInfoParam);

    List<MinioFileChunkUploadInfoDTO> listByFileMd5AndUploadId(String fileMd5, String uploadId);
}

@Service
public class MinioFileChunkUploadInfoServiceImpl
        extends ServiceImpl<MinioFileChunkUploadInfoMapper, MinioFileChunkUploadInfo>
        implements MinioFileChunkUploadInfoService {

    @Override
    public boolean saveMinioFileChunkUploadInfo(MinioFileChunkUploadInfoParam param) {
        List<MinioFileChunkUploadInfo> list = new ArrayList<>();
        for (int i = 0; i < param.getUploadUrls().size(); i++) {
            MinioFileChunkUploadInfo tempObj = new MinioFileChunkUploadInfo();
            tempObj.setChunkNumber(i + 1);
            tempObj.setFileMd5(param.getFileMd5());
            tempObj.setUploadId(param.getUploadId());
            tempObj.setExpiryTime(param.getExpiryTime());
            tempObj.setChunkUploadUrl(param.getUploadUrls().get(i));
            list.add(tempObj);
        }
        int result = this.baseMapper.insertBatchSomeColumn(list);
        return result != 0;
    }

    @Override
    public List<MinioFileChunkUploadInfoDTO> listByFileMd5AndUploadId(String fileMd5, String uploadId) {
        List<MinioFileChunkUploadInfo> list = this.baseMapper.selectList(
                Wrappers.<MinioFileChunkUploadInfo>lambdaQuery()
                        .select(MinioFileChunkUploadInfo::getChunkUploadUrl)
                        .eq(MinioFileChunkUploadInfo::getFileMd5, fileMd5)
                        .eq(MinioFileChunkUploadInfo::getUploadId, uploadId));
        return ExtBeanUtils.doListToDtoList(list, MinioFileChunkUploadInfoDTO.class);
    }
}

至此 Entity、Mapper、Service 准备完毕

三、Minio分片实现

3.1 文件状态枚举

@Getter
public enum MinioFileStatus {

    UN_UPLOADED("UN_UPLOADED", "待上传"),
    UPLOADED("UPLOADED", "已上传"),
    UPLOADING("", "上传中")
    ;

    final String code;
    final String msg;

    MinioFileStatus(String code, String msg) {
        this.code = code;
        this.msg = msg;
    }
}

3.2 MinioService新增方法

public interface MinioService {

	/**
     * 初始化获取 uploadId
     *
     * @param objectName  文件名
     * @param partCount   分片总数
     * @param contentType contentType
     * @return uploadInfo
     */
    MinioUploadInfo initMultiPartUpload(String objectName,
                                        int partCount,
                                        String contentType);

    /**
     * 分片合并
     *
     * @param objectName 文件名
     * @param uploadId   uploadId
     * @return region
     */
    String mergeMultiPartUpload(String objectName, String uploadId);

    /**
     * 获取已上传的分片列表
     *
     * @param objectName 文件名
     * @param uploadId   uploadId
     * @return 分片列表
     */
    List<Integer> listUploadChunkList(String objectName, String uploadId);
}

@Component
@Slf4j
@RequiredArgsConstructor
public class MinioServiceImpl implements MinioService {

    public MinioUploadInfo initMultiPartUpload(String objectName, int partCount, String contentType) {
        HashMultimap<String, String> headers = HashMultimap.create();
        headers.put("Content-Type", contentType);

        String uploadId = "";
        List<String> partUrlList = new ArrayList<>();
        try {
            // 获取 uploadId
            uploadId = minioClient.getUploadId(minIoClientConfig.getBucketName(),
                    null,
                    objectName,
                    headers,
                    null);
            Map<String, String> paramsMap = new HashMap<>(2);
            paramsMap.put("uploadId", uploadId);
            for (int i = 1; i <= partCount; i++) {
                paramsMap.put("partNumber", String.valueOf(i));
                // 获取上传 url
                String uploadUrl = minioClient.getPresignedObjectUrl(GetPresignedObjectUrlArgs.builder()
                        // 注意此处指定请求方法为 PUT，前端需对应，否则会报 `SignatureDoesNotMatch` 错误
                        .method(Method.PUT)
                        .bucket(minIoClientConfig.getBucketName())
                        .object(objectName)
                        // 指定上传连接有效期
                        // .expiry(paramConfig.getChunkUploadExpirySecond(), TimeUnit.SECONDS)
                        .extraQueryParams(paramsMap).build());

                partUrlList.add(uploadUrl);
            }
        } catch (Exception e) {
            log.error("initMultiPartUpload Error:" + e);
            return null;
        }
        // 过期时间 TODO 过期
        LocalDateTime expireTime = LocalDateTime.now().minusHours(1);
        MinioUploadInfo result = new MinioUploadInfo();
        result.setUploadId(uploadId);
        result.setExpiryTime(expireTime);
        result.setUploadUrls(partUrlList);
        return result;
    }

    /**
     * 分片合并
     *
     * @param objectName 文件名
     * @param uploadId   uploadId
     */
    public String mergeMultiPartUpload(String objectName, String uploadId) {
        // todo 最大1000分片 这里好像可以改吧
        Part[] parts = new Part[1000];
        int partIndex = 0;
        ListPartsResponse partsResponse = listUploadPartsBase(objectName, uploadId);
        if (null == partsResponse) {
            log.error("查询文件分片列表为空");
            throw new RuntimeException("分片列表为空");
        }
        for (Part partItem : partsResponse.result().partList()) {
            parts[partIndex] = new Part(partIndex + 1, partItem.etag());
            partIndex++;
        }
        ObjectWriteResponse objectWriteResponse;
        try {
            objectWriteResponse = minioClient.mergeMultipart(minIoClientConfig.getBucketName(), null, objectName, uploadId, parts, null, null);
        } catch (Exception e) {
            log.error("分片合并失败：" + e);
            throw new RuntimeException("分片合并失败：" + e.getMessage());
        }
        if (null == objectWriteResponse) {
            log.error("合并失败，合并结果为空");
            throw new RuntimeException("分片合并失败");
        }
        return objectWriteResponse.region();
    }

    /**
     * 获取已上传的分片列表
     *
     * @param objectName 文件名
     * @param uploadId   uploadId
     */
    public List<Integer> listUploadChunkList(String objectName, String uploadId) {
        ListPartsResponse partsResponse = listUploadPartsBase(objectName, uploadId);
        if (null == partsResponse) {
            return Collections.emptyList();
        }
        return partsResponse.result().partList().stream()
                .map(Part::partNumber).collect(Collectors.toList());
    }


    private ListPartsResponse listUploadPartsBase(String objectName, String uploadId) {
        int maxParts = 1000;
        ListPartsResponse partsResponse;
        try {
            partsResponse = minioClient.listMultipart(minIoClientConfig.getBucketName(), null, objectName, maxParts, 0, uploadId, null, null);
        } catch (ServerException | InsufficientDataException | ErrorResponseException | NoSuchAlgorithmException |
                 IOException | XmlParserException | InvalidKeyException | InternalException |
                 InvalidResponseException e) {
            log.error("查询文件分片列表错误：{}，uploadId:{}", e, uploadId);
            return null;
        }
        return partsResponse;
    }

}

3.3 分片文件Service

public interface FileUploadService {

    /**
     * 获取分片上传信息
     *
     * @param param 参数
     * @return Minio上传信息
     */
    MinioUploadInfo getUploadId(GetMinioUploadInfoParam param);

    /**
     * 检查文件是否存在
     *
     * @param md5 md5
     * @return true存在 false不存在
     */
    MinioOperationResult checkFileExistsByMd5(String md5);

    /**
     * 查询已上传的分片序号
     *
     * @param objectName 文件名
     * @param uploadId   uploadId
     * @return 已上传的分片序号列表
     */
    List<Integer> listUploadParts(String objectName, String uploadId);

    /**
     * 分片合并
     *
     * @param param 参数
     * @return url
     */
    String mergeMultipartUpload(MergeMinioMultipartParam param);
}

@Slf4j
@Service
public class FileUploadServiceImpl implements FileUploadService {

    @Resource
    private MinioService minioService;
    @Resource
    private MinioFileUploadInfoService minioFileUploadInfoService;
    @Resource
    private MinioFileChunkUploadInfoService minioFileChunkUploadInfoService;

    @Override
    public MinioUploadInfo getUploadId(GetMinioUploadInfoParam param) {
        MinioUploadInfo uploadInfo;
        MinioFileUploadInfoDTO minioFileUploadInfo = this.minioFileUploadInfoService.getByFileMd5(param.getFileMd5());
        if (null == minioFileUploadInfo) {
            // 计算分片数量
            double partCount = Math.ceil(param.getFileSize() * 1.0 / param.getChunkSize());
            log.info("总分片数：" + partCount);
            uploadInfo = minioService.initMultiPartUpload(param.getFileName(), (int) partCount, param.getContentType());
            if (null != uploadInfo) {
                MinioFileUploadInfoParam saveParam = new MinioFileUploadInfoParam();
                saveParam.setUploadId(uploadInfo.getUploadId());
                saveParam.setFileMd5(param.getFileMd5());
                saveParam.setFileName(param.getFileName());
                saveParam.setTotalChunk((int) partCount);
                saveParam.setFileStatus(MinioFileStatus.UN_UPLOADED.getCode());
                // 保存文件上传信息
                MinioFileUploadInfoDTO minioFileUploadInfoDTO = minioFileUploadInfoService.saveMinioFileUploadInfo(saveParam);
                log.info("文件上传信息保存成功 {}", JSON.toJSONString(minioFileUploadInfoDTO));

                MinioFileChunkUploadInfoParam chunkUploadInfoParam = new MinioFileChunkUploadInfoParam();
                chunkUploadInfoParam.setUploadUrls(uploadInfo.getUploadUrls());
                chunkUploadInfoParam.setUploadId(uploadInfo.getUploadId());
                chunkUploadInfoParam.setExpiryTime(uploadInfo.getExpiryTime());
                chunkUploadInfoParam.setFileMd5(param.getFileMd5());
                chunkUploadInfoParam.setFileName(param.getFileName());
                // 保存分片上传信息
                boolean chunkUploadResult = minioFileChunkUploadInfoService.saveMinioFileChunkUploadInfo(chunkUploadInfoParam);
                log.info("文件分片信息保存{}", chunkUploadResult ? "成功" : "失败");
            }
            return uploadInfo;
        }
        // 查询分片上传地址
        List<MinioFileChunkUploadInfoDTO> list = minioFileChunkUploadInfoService.listByFileMd5AndUploadId(minioFileUploadInfo.getFileMd5(), minioFileUploadInfo.getUploadId());
        List<String> uploadUrlList = list.stream()
                .map(MinioFileChunkUploadInfoDTO::getChunkUploadUrl)
                .collect(Collectors.toList());
        uploadInfo = new MinioUploadInfo();
        uploadInfo.setUploadUrls(uploadUrlList);
        uploadInfo.setUploadId(minioFileUploadInfo.getUploadId());
        return uploadInfo;
    }

    @Override
    public MinioOperationResult checkFileExistsByMd5(String md5) {
        MinioOperationResult result = new MinioOperationResult();
        MinioFileUploadInfoDTO minioFileUploadInfo = this.minioFileUploadInfoService.getByFileMd5(md5);
        if (null == minioFileUploadInfo) {
            result.setStatus(MinioFileStatus.UN_UPLOADED.getCode());
            return result;
        }
        // 已上传
        if (Objects.equals(minioFileUploadInfo.getFileStatus(), MinioFileStatus.UPLOADED.getCode())) {
            result.setStatus(MinioFileStatus.UPLOADED.getCode());
            result.setUrl(minioFileUploadInfo.getFileUrl());
            return result;
        }
        // 查询已上传分片列表并返回已上传列表
        List<Integer> chunkUploadedList = listUploadParts(minioFileUploadInfo.getFileName(), minioFileUploadInfo.getUploadId());
        result.setStatus(MinioFileStatus.UPLOADING.getCode());
        result.setChunkUploadedList(chunkUploadedList);
        return result;
    }

    @Override
    public List<Integer> listUploadParts(String objectName, String uploadId) {
        return minioService.listUploadChunkList(objectName, uploadId);
    }

    @Override
    public String mergeMultipartUpload(MergeMinioMultipartParam param) {
        String result = minioService.mergeMultiPartUpload(param.getFileName(), param.getUploadId());
        if (!StringUtils.isBlank(result)) {
            MinioFileUploadInfoParam fileUploadInfoParam = new MinioFileUploadInfoParam();
            fileUploadInfoParam.setFileUrl(result);
            fileUploadInfoParam.setFileMd5(param.getMd5());
            fileUploadInfoParam.setFileStatus(MinioFileStatus.UPLOADED.getCode());

            // 更新状态
            int updateRows = minioFileUploadInfoService.updateFileStatusByFileMd5(fileUploadInfoParam);
            log.info("update file by file md5 updated count {}", updateRows);
        }
        return result;
    }
}

3.4 Controller新增

@RequestMapping(value = "file")
@RestController
public class FileController {

	@Resource
    private FileUploadService fileUploadService;

    @PostMapping("/upload")
    public R getUploadId(@Validate @RequestBody GetMinioUploadInfoParam param) {
        MinioUploadInfo minioUploadId = fileUploadService.getUploadId(param);
        return R.ok().setData(minioUploadId);
    }

    @GetMapping("/upload/check")
    public R checkFileUploadedByMd5(@RequestParam("md5") String md5) {
        return R.ok().setData(fileUploadService.checkFileExistsByMd5(md5));
    }

    @PostMapping("/upload/merge")
    public R mergeUploadFile(@Validated MergeMinioMultipartParam param) {
        String result = fileUploadService.mergeMultipartUpload(param);
        if (StringUtils.isEmpty(result)) {
            throw new MinioDemoException(MinioDemoExceptionTypes.CHUNK_MERGE_FAILED);
        }
        return R.ok().setData(result); // url
    }
}

3.5 文件分片上传测试

select * from minio_file_upload_info;

select * from minio_chunk_upload_info;