当前位置：首页 > news >正文

基于SpringBoot文件管理系统中的分片上传实现

news 2025/6/27 12:08:52

分片上传

一、引言

现代Web应用中，大文件上传是一个常见需求。传统单文件上传方式在网络不稳定或文件过大时存在诸多问题。分片上传技术通过将大文件分割为小块进行传输，完美解决了大文件上传的痛点。本文将深入探讨分片上传的核心实现方案。

二、功能概述

分片上传功能用于处理大文件上传，支持断点续传、MD5校验秒传和并发上传，核心特点包括：

断点续传：通过记录已上传分片，支持中断后继续上传
秒传功能：基于文件MD5值实现文件快速上传
分片校验：通过分片编号和总数控制上传完整性
并发控制：使用线程池(20线程)处理并发上传任务
多存储支持：通过UFOPFactory适配本地存储、阿里云OSS等多种后端
数据一致性：事务保证上传过程原子性，避免数据不一致

三、核心实现代码分析

1. 极速上传检查（秒传功能）

FiletransferController.java 中的极速上传接口实现文件MD5校验和秒传功能：

@Operation(summary = "极速上传", description = "校验文件MD5判断是否可秒传", tags = {"filetransfer"})
@RequestMapping(value = "/uploadfile", method = RequestMethod.GET)
public RestResult<UploadFileVo> uploadFileSpeed(UploadFileDTO uploadFileDTO) {// 检查文件是否已存在（通过MD5标识符）Map<String, Object> param = new HashMap<>();param.put("identifier", uploadFileDTO.getIdentifier());List<FileBean> list = fileMapper.selectByMap(param);if (list != null && !list.isEmpty()) {// 文件已存在，直接返回秒传成功uploadFileVo.setSkipUpload(true);return RestResult.success().data(uploadFileVo);}// 检查已上传的分片，支持断点续传List<Integer> uploaded = uploadTaskDetailMapper.selectUploadedChunkNums(uploadFileDTO.getIdentifier());uploadFileVo.setUploaded(uploaded);return RestResult.success().data(uploadFileVo);
}

2. 分片上传处理

FiletransferService.java 中的分片上传实现：

@Override
@Transactional(rollbackFor = Exception.class)
public void uploadFile(HttpServletRequest request, UploadFileDTO uploadFileDTO, String userId) {// 1. 检查分片是否已上传List<Integer> uploadedChunks = uploadTaskDetailMapper.selectUploadedChunkNums(uploadFileDTO.getIdentifier());if (uploadedChunks.contains(uploadFileDTO.getChunkNumber())) {log.info("分片{}已上传，跳过处理", uploadFileDTO.getChunkNumber());return;}// 2. 上传当前分片Uploader uploader = ufopFactory.getUploader();UploadFileResult result = uploader.uploadChunk(uploadFileDTO, request.getInputStream());// 3. 记录已上传分片UploadTaskDetail taskDetail = new UploadTaskDetail();taskDetail.setIdentifier(uploadFileDTO.getIdentifier());taskDetail.setChunkNumber(uploadFileDTO.getChunkNumber());taskDetail.setChunkSize((int)uploadFileDTO.getChunkSize());taskDetail.setTotalChunks(uploadFileDTO.getTotalChunks());uploadTaskDetailMapper.insert(taskDetail);// 4. 检查是否所有分片都已上传完成List<Integer> allUploadedChunks = uploadTaskDetailMapper.selectUploadedChunkNums(uploadFileDTO.getIdentifier());if (allUploadedChunks.size() == uploadFileDTO.getTotalChunks()) {// 5. 合并所有分片mergeChunks(uploadFileDTO.getIdentifier(), userId);}
}

3. 分片合并实现

private void mergeChunks(String identifier, String userId) {// 1. 获取所有分片信息并排序List<UploadTaskDetail> chunks = uploadTaskDetailMapper.selectByIdentifier(identifier);chunks.sort(Comparator.comparingInt(UploadTaskDetail::getChunkNumber));// 2. 创建合并文件File mergedFile = new File(tempPath + File.separator + identifier);try (FileOutputStream out = new FileOutputStream(mergedFile)) {for (UploadTaskDetail chunk : chunks) {// 3. 读取每个分片并写入合并文件Downloader downloader = ufopFactory.getDownloader(storageType);DownloadFile downloadFile = new DownloadFile();downloadFile.setFileUrl(chunk.getChunkUrl());InputStream in = downloader.getInputStream(downloadFile);IOUtils.copy(in, out);in.close();}// 4. 保存合并后的文件到存储系统saveMergedFile(mergedFile, identifier, userId);// 5. 清理临时分片文件cleanTempChunks(chunks);// 6. 更新上传任务状态updateUploadTaskStatus(identifier, UploadStatus.COMPLETED);} catch (IOException e) {log.error("分片合并失败", e);// 7. 失败处理：更新状态，发送通知updateUploadTaskStatus(identifier, UploadStatus.FAILED);notifyUserUploadFailure(userId, identifier);throw new UploadException("文件合并失败");}
}

四、关键技术点

1. 断点续传机制

通过UploadTaskDetail表记录已上传分片编号
前端根据返回的uploaded数组跳过已上传分片
核心代码：List<Integer> uploaded = uploadTaskDetailMapper.selectUploadedChunkNums(identifier);

2. 并发控制

使用线程池管理上传任务：public static Executor exec = Executors.newFixedThreadPool(20);
通过exec.execute()异步处理文件合并和后续操作
分片上传使用乐观锁避免资源冲突

3. 存储抽象

通过UFOPFactory获取对应存储类型的操作对象：

Uploader uploader = ufopFactory.getUploader(fileBean.getStorageType());

支持本地存储、阿里云OSS等多种后端无缝切换
统一接口设计，扩展新的存储类型只需实现接口

4. 数据一致性保障

使用@Transactional注解确保分片上传和合并操作的原子性
合并文件时使用SequenceInputStream保证分片顺序
异常处理机制：
- 分片上传失败自动重试（最多3次）
- 合并失败自动回滚并通知用户
- 超时处理机制（30秒无响应自动终止）

五、数据库设计

分片上传功能依赖两个核心数据表：

1. UploadTask（上传任务）

public class UploadTask {private String taskId;          // 任务ID (主键)private String identifier;      // 文件唯一标识(MD5)private String fileName;        // 文件名private long totalSize;         // 文件总大小private int totalChunks;        // 总分片数private int status;             // 任务状态(0-进行中,1-完成,2-失败)private String userId;          // 所属用户IDprivate int storageType;        // 存储类型 (0-本地,1-OSS,2-S3)private Date createTime;        // 创建时间private Date updateTime;        // 更新时间
}

2. UploadTaskDetail（分片详情）

public class UploadTaskDetail {private String detailId;        // 分片ID (主键)private String identifier;      // 文件标识 (外键)private int chunkNumber;        // 分片序号 (1-based)private String chunkUrl;        // 分片存储路径private long chunkSize;         // 分片大小 (字节)private String chunkMd5;        // 分片MD5值private int status;             // 分片状态(0-未上传,1-已上传,2-校验失败)private Date createTime;        // 创建时间
}

数据库关系图

六、性能优化策略

1. 分片大小动态调整

根据网络状况动态调整分片大小（256KB-5MB）
计算公式：chunkSize = max(256KB, min(5MB, 网络带宽×0.8×预估上传时间))

2. 并行上传优化

使用CompletableFuture实现更精细的并发控制：

List<CompletableFuture<Void>> futures = chunksToUpload.stream().map(chunkNum -> CompletableFuture.runAsync(() -> uploadChunk(chunkNum), uploadThreadPool).collect(Collectors.toList());CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

3. 内存优化

使用NIO的FileChannel减少大文件操作的内存占用
流式处理避免一次性加载整个分片到内存
分片合并使用缓冲区技术（默认8KB缓冲区）

4. 前端优化

Web Worker计算MD5避免界面卡顿
增量MD5计算：分片计算后合并为完整MD5
上传暂停/恢复功能实现：

// 暂停上传
handlePause() {this.uploadController.abort();this.isPaused = true;
}// 恢复上传
handleResume() {this.uploadController = new AbortController();this.uploadChunks(this.resumeChunks);
}

七、安全设计

1. 安全验证机制

分片上传请求必须携带JWT令牌
服务端验证用户对上传目录的权限
文件类型白名单验证（防止上传可执行文件）

2. 分片校验

每个分片单独计算MD5并存储
合并前验证全部分片的MD5值
防篡改机制：最终MD5 = MD5(分片1-MD5 + 分片2-MD5 + ...)

3. 恶意上传防护

单用户并发上传任务限制（最多5个）
分片上传频率限制（每秒不超过50个分片）
黑名单机制（检测到恶意行为自动封禁IP）

八、错误处理与监控

1. 错误分类处理

错误类型	处理方式	重试策略
网络中断	自动断点续传	立即重试3次
分片校验失败	重新上传该分片	延迟5秒重试
存储空间不足	终止上传并通知	不重试
权限变更	终止上传并通知	不重试

2. 监控指标

分片上传成功率：(成功分片数/总分片数)×100%
平均上传速度：总上传大小/总耗时
秒传率：秒传文件数/总上传请求数×100%
并发任务数：实时监控活跃上传任务

3. 日志监控

// 关键操作记录详细日志
log.info("开始合并分片，标识: {}, 分片数: {}", identifier, chunks.size());
// 错误日志记录完整上下文
log.error("分片合并失败，标识: {}, 错误: {}", identifier, e.getMessage(), e);

九、扩展能力

1. 云存储特殊处理

阿里云OSS分片上传优化：

// 初始化分片上传
InitiateMultipartUploadRequest request = new InitiateMultipartUploadRequest(bucketName, objectName);
InitiateMultipartUploadResult result = ossClient.initiateMultipartUpload(request);
String uploadId = result.getUploadId();// 上传分片
UploadPartRequest uploadPartRequest = new UploadPartRequest();
uploadPartRequest.setBucketName(bucketName);
uploadPartRequest.setKey(objectName);
uploadPartRequest.setUploadId(uploadId);
uploadPartRequest.setPartNumber(partNumber);
uploadPartRequest.setInputStream(partInputStream);
UploadPartResult uploadPartResult = ossClient.uploadPart(uploadPartRequest);// 完成上传
CompleteMultipartUploadRequest completeRequest = new CompleteMultipartUploadRequest(bucketName, objectName, uploadId, partETags);
ossClient.completeMultipartUpload(completeRequest);

2. 客户端SDK封装

提供统一的上传SDK，封装复杂的分片逻辑：

import { ChunkUploader } from '@sdk/file-upload';const uploader = new ChunkUploader({target: '/api/upload',chunkSize: 1024 * 1024, // 1MBmaxConcurrent: 3, // 并发数headers: {'Authorization': 'Bearer <token>'}
});uploader.upload(file).onProgress(progress => {console.log(`上传进度: ${progress}%`);}).onSuccess(() => {console.log('上传成功');}).onError(error => {console.error('上传失败', error);});