Java分片上传，大文件上传

使用场景

• 大文件加速上传当文件大小超过5 GB时，使用分片上传可实现并行上传多个Part以加快上传速度。
• 网络环境较差网络环境较差时，建议使用分片上传。当出现上传失败的时候，仅需重传失败的Part。
• 文件大小不确定可以在需要上传的文件大小还不确定的情况下开始上传，这种场景在视频监控等行业应用中比较常见。

流程

实现

下面实现本地大文件分片上传。

文件接口：FileService

import com.demo.controller.file.dto.file.MultipartUploadVO;
import org.springframework.web.multipart.MultipartFile;

import java.util.List;

/**
 * 文件 Service 接口
 *
 **/
public interface FileService {

    /**
     * 根据文件名和文件大小生成文件唯一id
     * @param fileName 文件名
     * @param fileSize 文件大小
     * @return
     */
    MultipartUploadVO getPartUploadId(String fileName, Integer fileSize);

    /**
     * 上传分片文件
     * @param uploadId 分片上传id
     * @param content 分片文件内容
     * @param partNumber 分片号
     * @param headerRange header
     * @return
     * @throws Exception
     */
    String uploadPart(Long uploadId, byte[] content, Integer partNumber, String headerRange) throws Exception;

    /**
     * 合并分片文件
     * @param uploadId 分片上传id
     * @param path 文件唯一名称
     * @return
     * @throws Exception
     */
    String completeMultipartUpload(Long uploadId, String path) throws Exception;

}

初始化上传结果MultipartUploadVO

import io.swagger.annotations.ApiModelProperty;
import lombok.*;

@Getter
@Setter
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class MultipartUploadVO {

    /**
     * 分片上传执行步骤:
     * <ul>
     *     <li>NEXT: </li>执行分片上传
     *     <li>FINISHED: </li>文件已存在
     * </ul>
     */
    @ApiModelProperty(value = "分片上传执行步骤,NEXT: 执行分片上传;FINISHED: 文件已存在", example = "NEXT/FINISHED")
    private MultipartUploadStepEnum step;

    /**
     * 分片上传id,当step=NEXT,该字段有值
     */
    @ApiModelProperty(value = "分片上传id,当step=NEXT,该字段有值",notes = "当step=NEXT,该字段有值", example = "1024")
    private Long uploadId;

    /**
     * 文件访问地址,当step=FINISHED,该字段有值
     */
    @ApiModelProperty(value = "文件访问地址,当step=FINISHED,该字段有值",notes = "当step=FINISHED,该字段有值", example = "http://127.0.0.1:38086/infra/file/1638095079969169408/get/9f5951c4f20c62ca6e27c2b15dff54dc5d9506045a4c338646f269d6ac7cff23.pdf")
    private String url;

    
    public enum MultipartUploadStepEnum {

        /**
         * 执行分片上传
         */
        NEXT,
        /**
         * 该文件已存在,直接返回可访问的url
         */
        FINISHED;
    }
}

接口实现FileServiceImpl

import cn.hutool.core.io.IoUtil;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.StrUtil;
import com.demo.dto.file.MultipartUploadVO;
import com.demo.mapper.file.FileMapper;
import com.demo.service.file.FileService;
import com.demo.util.io.FileUtils;
import com.demo.CompleteMultipartUploadResult;
import com.demo.MultipartUploadPartParam;
import com.demo.FileClient;
import lombok.SneakyThrows;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import javax.annotation.Resource;

import java.util.ArrayList;
import java.util.List;

/**
 * 文件 Service 实现类
 **/
@Service
public class FileServiceImpl implements FileService {

    @Resource
    private FileMapper fileMapper;

    @Override
    public MultipartUploadVO getPartUploadId(String fileName, Integer fileSize) {
        Assert.isFalse(fileSize < MultipartUploadPartParam.MIN_MULTIPART_SIZE, () -> new Exception("小于5M的文件不用分片") );
        String path = FileUtils.generatePath(fileName, fileSize);
        //查询数据库
        final FileDO fileDO = fileMapper.selectOne(FileDO::getPath, path);
        //秒传
        if (fileDO != null) {
            return MultipartUploadVO.builder()
                    .step(MultipartUploadVO.MultipartUploadStepEnum.FINISHED)
                    .url(fileDO.getUrl())
                    .build();
        }
        // 保存到数据库
        FileDO file = new FileDO();
        file.setId(12345L);
        file.setName(fileName);
        file.setPath(path);
        //合并文件后重置
        file.setUrl(StrUtil.EMPTY);
        //合并文件后重置
        file.setSize(fileSize);
        fileMapper.insert(file);
        //返回id，下一步执行
        return MultipartUploadVO.builder()
                .step(MultipartUploadVO.MultipartUploadStepEnum.NEXT)
                .uploadId(file.getId())
                .build();
    }

    @Override
    public String uploadPart(Long uploadId, byte[] content,
                                 Integer partNumber, String headerRange) throws Exception {
        Assert.isFalse(content.length < MultipartUploadPartParam.MIN_MULTIPART_SIZE, 
                       () -> new Exception("分片文件不能小于5M"));
        final FileDO fileDO = fileMapper.selectOne(FileDO::getId, uploadId);
        // 上传到文件存储器
        FileClient client = new LocalFileClient();
        Assert.notNull(client, "客户端(master) 不能为空");
        final MultipartUploadPartParam build = MultipartUploadPartParam.builder().uploadId(uploadId)
                .part(content).partNumber(partNumber).path(fileDO.getPath()).type(fileDO.getType()).build();
        String url = client.uploadPart(build).getStagingPath();
        return url;
    }

    @Override
    public String completeMultipartUpload(Long uploadId, String path) throws Exception {
        final FileDO fileDO = fileMapper.selectOne(FileDO::getId, uploadId);
        // 上传到文件存储器
        FileClient client = new LocalFileClient();
        Assert.notNull(client, "客户端(master) 不能为空");
        final CompleteMultipartUploadResult completed = client.completeMultipartUpload(uploadId, path);
        String url = completed.getUrl();
        fileDO.setUrl(url);
        //校正实际文件大小
        fileDO.setSize(completed.getTotalSize());
        // 更新数据库
        fileMapper.updateById(fileDO);
        return url;
    }
}

工具类FileUtils

public static String generatePath(String originalName, Integer fileSize) {
        String sha256Hex = DigestUtil.sha256Hex(originalName + fileSize);
        String extName = FileNameUtil.extName(originalName);
        return StrUtil.isBlank(extName) ? sha256Hex : sha256Hex + "." + extName;
    }

分片上传参数对象MultipartUploadPartParam

import cn.hutool.core.lang.Assert;
import lombok.*;

import javax.validation.constraints.NotNull;

/**
 * @desc 分片上传参数对象
 */
@Getter
@Setter
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class MultipartUploadPartParam {
    /**
     * 分片上传id
     */
    @NotNull(message = "分片上传id不能为空")
    private Long uploadId;
    /**
     * 分片文件不能为空, 最小5M, 最大5G
     */
    @NotNull(message = "分片文件不能为空, 最小5M, 最大5G")
    private byte[] part;
    /**
     * 分片文件加密文件名(保证唯一)
     */
    private String path;
    /**
     * 分片号
     */
    @NotNull(message = "分片号不能为空")
    private Integer partNumber;
    /**
     * 文件类型
     */
    private String type;

    // allowed minimum part size is 5MiB in multipart upload.
    public static final int MIN_MULTIPART_SIZE = 5 * 1024 * 1024;
    // allowed minimum part size is 5GiB in multipart upload.
    public static final long MAX_PART_SIZE = 5L * 1024 * 1024 * 1024;

    /**
     * 校验分片文件大小
     */
    public void validPartSize() {
        Assert.isFalse(part.length < MIN_MULTIPART_SIZE, () -> new Exception("分片文件不能小于5M"));
        Assert.isFalse(part.length > MAX_PART_SIZE, () -> new Exception("分片文件不能大于5G"));
    }
}

合并文件结果对象CompleteMultipartUploadResult

import lombok.*;

/**
 * @desc 合并文件结果对象
 */
@Getter
@Setter
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class CompleteMultipartUploadResult {

    /**
     * 分片上传id
     */
    private Long uploadId;

    /**
     * 合并后文件访问地址
     */
    private String url;

    /**
     * 合并后总文件大小
     */
    private Integer totalSize;
}

分片上传系统配置MultipartUploadProperties

import lombok.Getter;
import lombok.Setter;
import org.springframework.boot.context.properties.ConfigurationProperties;

import java.io.File;

/**
 * @desc
 */
@Getter
@Setter
@ConfigurationProperties(prefix = MultipartUploadProperties.PREFIX)
public class MultipartUploadProperties {

    public static final String PREFIX = "multipart-upload";


    /**
     * 分片文件暂存目录
     */
    private String stagingDirectory = "temp";

    /**
     * 缓存时间,默认60秒
     */
    private Long cacheOutSeconds = 60L;

    public String getStagingDirectory() {
        if (!stagingDirectory.endsWith(File.separator)) {
            stagingDirectory += File.separator;
        }
        return stagingDirectory;
    }

    /**
     * 获取指定暂存目录
     * @param uploadId
     * @return
     */
    public String getStagingDirectory(Long uploadId) {
        return getStagingDirectory() + uploadId + File.separator;
    }
}

本地文件存储客户端LocalFileClient

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.io.FileUtil;
import com.demo.CompleteMultipartUploadResult;
import com.demo.MultipartUploadPart;
import com.demo.MultipartUploadPartParam;
import com.demo.AbstractFileClient;

import java.io.File;
import java.io.FileOutputStream;
import java.io.OutputStream;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 本地文件存储客户端
 **/
public class LocalFileClient implements FileClient {

        /**
     * 分片缓存
     */
    protected static LoadingCache<Long, List<MultipartUploadPart>> multipartUploadCache;
    protected static final ReentrantReadWriteLock LOCK = new ReentrantReadWriteLock();

        /**
     * 初始化
     */
    public final void init() {
        //加载配置
        properties = SpringUtil.getBean(MultipartUploadProperties.class);
        //加载缓存池
        multipartUploadCache = CacheUtils.buildAsyncReloadingCache(
                Duration.ofSeconds(properties.getCacheOutSeconds()),
                new CacheLoader<>() {
                    @Override
                    public List<MultipartUploadPart> load(Long uploadId) throws Exception {
                        return CollUtil.newArrayList();
                    }
                }
        );
        log.info("[init][配置({}) 初始化完成]", config);
    }

    @Override
    public String upload(byte[] content, String path, String type) throws Exception {
        // 执行写入
        String filePath = getFilePath(path);
        FileUtil.writeBytes(content, filePath);
        // 拼接返回路径
        return super.formatFileUrl(config.getDomain(), path);
    }

    private String getFilePath(String path) {
        return config.getBasePath() + path;
    }

    @Override
    public MultipartUploadPart uploadPart(MultipartUploadPartParam partParam) throws Exception {
        //分片任务id
        Long uploadId = partParam.getUploadId();
        //分片号
        Integer partNumber = partParam.getPartNumber();
        //获取该id下已上传的分片列表
        List<MultipartUploadPart> parts = listCacheParts(uploadId, partNumber);
        //获取系统配置路径
        String pathPrefix = properties.getStagingDirectory(uploadId);
        // 执行写入
        String filePath = pathPrefix + partNumber + ".part";
        LOCK.writeLock().lock();
        try (OutputStream outputStream = new FileOutputStream(filePath)) {
            outputStream.write(partParam.getPart());
            final MultipartUploadPart multipartUploadPart = MultipartUploadPart.builder()
                    .uploadId(uploadId)
                    .partNumber(partNumber)
                    .stagingPath(filePath)
                    .partSize(partParam.getPart().length)
                    .build();
            parts.add(multipartUploadPart);
            //放入缓存
            multipartUploadCache.put(uploadId, parts);
            return multipartUploadPart;
        } finally {
            LOCK.writeLock().unlock();
        }
    }

    @Override
    public CompleteMultipartUploadResult completeMultipartUpload(Long uploadId, String path) throws Exception {
        //查询缓存中的全部分片列表
        List<MultipartUploadPart> parts = listCacheParts(uploadId);
        final String url = mergePartFile(uploadId, path, parts);
        return CompleteMultipartUploadResult.builder()
                .uploadId(uploadId)
                .url(url)
                .totalSize(parts.stream().collect(Collectors.summingInt(MultipartUploadPart::getPartSize)))
                .build();
    }

    /**
     * 合并分片文件
     *
     * @param uploadId
     * @param path
     * @return
     */
    public String mergePartFile(Long uploadId, String path, List<MultipartUploadPart> parts) throws Exception {
        if (CollUtil.isEmpty(parts)) {
            return null;
        }
        // 执行写入
        String filePath = getFilePath(path);
        //合并文件
        final File file = FileUtil.touch(filePath);
        File partFile;
        try (FileOutputStream out = new FileOutputStream(file, true)) {
            //追加写入文件
            for (MultipartUploadPart part : parts) {
                partFile = new File(part.getStagingPath());
                out.write(FileUtil.readBytes(partFile));
                out.flush();
            }
        } catch (Exception e) {
            throw e;
        }
        //删除缓存
        multipartUploadCache.invalidate(uploadId);
        //删除分片文件
        final String stagingDirectory = properties.getStagingDirectory(uploadId);
        FileUtil.del(stagingDirectory);
        // 拼接返回路径
        return super.formatFileUrl(config.getDomain(), path);
    }
        /**
     * 合并分片时获取缓存数据
     *
     * @param uploadId 分片上传id
     * @return
     * @throws Exception
     */
    protected List<MultipartUploadPart> listCacheParts(Long uploadId) throws Exception {
        List<MultipartUploadPart> parts;
        LOCK.readLock().lock();
        try {
            //防止读取的时候还有别的线程在写入
            parts = multipartUploadCache.get(uploadId);
            if (CollUtil.isEmpty(parts)) {
                throw new BizException(StrUtil.format("当前uploadId={}的分片文件不存在", uploadId));
            }
        } finally {
            LOCK.readLock().unlock();
        }
        parts.sort(Comparator.comparing(MultipartUploadPart::getPartNumber));
        final MultipartUploadPart last = CollUtil.getLast(parts);
        if (parts.size() != last.getPartNumber() + 1) {
            throw new BizException(StrUtil.format("当前uploadId={}的分片文件不完整", uploadId));
        }
        return parts;
    }

    /**
     * 分片上传时获取缓存数据
     *
     * @param uploadId   分片上传id
     * @param partNumber 分片号
     * @return
     * @throws Exception
     */
    protected List<MultipartUploadPart> listCacheParts(Long uploadId, Integer partNumber) throws Exception {
        List<MultipartUploadPart> parts;
        LOCK.readLock().lock();
        try {
            parts = multipartUploadCache.get(uploadId);
            if (CollUtil.contains(parts, item -> Objects.equals(item.getPartNumber(), partNumber))) {
                throw new BizException(StrUtil.format("当前uploadId={}的分片号partNumber={}已存在", uploadId, partNumber));
            }
        } finally {
            LOCK.readLock().unlock();
        }
        return parts;
    }
}

时序图