SpringBoot基于Minio实现分片上传、断点续传的实现

MinioClient client = MinioClient.builder()
                .endpoint("http://192.168.xx.133:9000")  // 服务端IP+端口
                .credentials(minioProperties.getAccessKey(), // 服务端用户名
                             minioProperties.getSecretKey()) // 服务端密码
                .build();

/**
     * 将文件进行分片上传
     * <p>有一个未处理的bug（虽然概率很低很低）：</p>
     * 当两个线程同时上传md5相同的文件时，由于两者会定位到同一个桶的同一个临时目录，两个线程会相互产生影响！
     * 
     * @param file 分片文件
     * @param currIndex 当前文件的分片索引
     * @param totalPieces 切片总数（对于同一个文件，请确保切片总数始终不变）
     * @param md5 整体文件MD5
     * @return 剩余未上传的文件索引集合
     */
    public FragResult uploadFileFragment(MultipartFile file,
                                  Integer currIndex, Integer totalPieces, String md5) throws Exception {
        checkNull(currIndex, totalPieces, md5);
        // 临时文件存放桶
        if ( !this.bucketExists(DEFAULT_TEMP_BUCKET_NAME) ) {
            this.createBucket(DEFAULT_TEMP_BUCKET_NAME);
        }
        // 得到已上传的文件索引
        Iterable<Result<Item>> results = this.getFilesByPrefix(DEFAULT_TEMP_BUCKET_NAME, md5.concat("/"), false);
        Set<Integer> savedIndex = Sets.newHashSet();
        boolean fileExists = false;
        for (Result<Item> item : results) {
            Integer idx = Integer.valueOf( getContentAfterSlash(item.get().objectName()) );
            if (currIndex.equals( idx )) {
                fileExists = true;
            }
            savedIndex.add( idx );
        }
        // 得到未上传的文件索引
        Set<Integer> remainIndex = Sets.newTreeSet();
        for (int i = 0; i < totalPieces; i++) {
            if ( !savedIndex.contains(i) ) {
                remainIndex.add(i);
            }
        }
        if (fileExists) {
            return new FragResult(false, remainIndex, "index [" + currIndex + "] exists");
        }
        this.uploadFileStream(DEFAULT_TEMP_BUCKET_NAME, this.getFileTempPath(md5, currIndex, totalPieces), file.getInputStream());
        // 还剩一个索引未上传，当前上传索引刚好是未上传索引，上传完当前索引后就完全结束了。
        if ( remainIndex.size() == 1 && remainIndex.contains(currIndex) ) {
            return new FragResult(true, null, "completed");
        }
        return new FragResult(false, remainIndex, "index [" + currIndex + "] has been uploaded");
    }

/**
     * 合并分片文件，并放到指定目录
     * 前提是之前已把所有分片上传完毕。
     * 
     * @param bucketName 目标文件桶名
     * @param targetName 目标文件名（含完整路径）
     * @param totalPieces 切片总数（对于同一个文件，请确保切片总数始终不变）
     * @param md5 文件md5
     * @return minio原生对象，记录了文件上传信息
     */
    public boolean composeFileFragment(String bucketName, String targetName, 
                                                   Integer totalPieces, String md5) throws Exception {
        checkNull(bucketName, targetName, totalPieces, md5);
        // 检查文件索引是否都上传完毕
        Iterable<Result<Item>> results = this.getFilesByPrefix(DEFAULT_TEMP_BUCKET_NAME, md5.concat("/"), false);
        Set<String> savedIndex = Sets.newTreeSet();
        for (Result<Item> item : results) {
            savedIndex.add( item.get().objectName() );
        }
        if (savedIndex.size() == totalPieces) {
            // 文件路径 转 文件合并对象
            List<ComposeSource> sourceObjectList = savedIndex.stream()
                    .map(filePath -> ComposeSource.builder()
                            .bucket(DEFAULT_TEMP_BUCKET_NAME)
                            .object( filePath )
                            .build())
                    .collect(Collectors.toList());
            ObjectWriteResponse objectWriteResponse = client.composeObject(
                    ComposeObjectArgs.builder()
                            .bucket(bucketName)
                            .object(targetName)
                            .sources(sourceObjectList)
                            .build());
            // 上传成功，则删除所有的临时分片文件
            List<String> filePaths = Stream.iterate(0, i -> ++i)
                    .limit(totalPieces)
                    .map(i -> this.getFileTempPath(md5, i, totalPieces) )
                    .collect(Collectors.toList());
            Iterable<Result<DeleteError>> deleteResults = this.removeFiles(DEFAULT_TEMP_BUCKET_NAME, filePaths);
            // 遍历错误集合（无元素则成功）
            for (Result<DeleteError> result : deleteResults) {
                DeleteError error = result.get();
                System.err.printf("[Bigfile] 分片'%s'删除失败! 错误信息: %s", error.objectName(), error.message());
            }
            return true;
        }
        throw new GlobalException("The fragment index is not complete. Please check parameters [totalPieces] or [md5]");
    }

    /**
     * 分片上传
     * @param user 用户对象
     * @param fileAddDto file: 分片文件, 
     *                   currIndex: 当前分片索引, 
     *                   totalPieces: 分片总数，
     *                   md5: 文件md5
     * @return 前端需上传的下一个分片序号（-1表示上传完成）
     */
    @PostMapping("/file/big/upload")
    public ResultData<String> uploadBigFile(User user, BigFileAddDto fileAddDto) {
        // 1.文件为空，返回失败 (一般不是用户的问题)
        if (fileAddDto.getFile() == null) {
            throw new GlobalException();
        }
        // 2.名字为空，或包含特殊字符，则提示错误
        String fileName = fileAddDto.getFile().getOriginalFilename();
        if (StringUtils.isEmpty(fileName) || fileName.matches(FileSysConstant.NAME_EXCEPT_SYMBOL)) {
            throw new GlobalException(ResultCode.INCORRECT_FILE_NAME);
        }
        // 3. 执行分片上传
        String result = fileSystemService.uploadBigFile(user, fileAddDto);
        return GlobalResult.success(result);
    }

    @Override
    public String uploadBigFile(User user, BigFileAddDto fileAddDto) {
        try {
            MultipartFile file = fileAddDto.getFile();
            Integer currIndex = fileAddDto.getCurrIndex();
            Integer totalPieces = fileAddDto.getTotalPieces();
            String md5 = fileAddDto.getMd5();
            log.info("[Bigfile] 上传文件md5: {} ,分片索引: {}", md5, currIndex);
            FragResult fragResult = minioUtils.uploadFileFragment(file, currIndex, totalPieces, md5);
            // 分片全部上传完毕
            if ( fragResult.isAllCompleted() ) {
                FileInfo fileInfo = getFileInfo(fileAddDto, user.getId());
                DBUtils.checkOperation( fileSystemMapper.insertFile(fileInfo) );
                String realPath = generateRealPath(generateVirtPath(fileAddDto.getParentPath(), file.getOriginalFilename()));
                // 发起文件合并请求, 无异常则成功
                minioUtils.composeFileFragment(getBucketByUsername(user.getUsername()), realPath, totalPieces, md5);
                return "-1";
            } else {
                Iterator<Integer> iterator = fragResult.getRemainIndex().iterator();
                if (iterator.hasNext()) {
                    String nextIndex = iterator.next().toString();
                    log.info("[BigFile] 下一个需上传的文件索引是:{}", nextIndex);
                    return nextIndex;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        log.error("[Bigfile] 上传文件时出现异常");
        throw new GlobalException(ResultCode.FILE_UPLOAD_ERROR);
    }