阿里云OSS实战：用Java SDK实现大文件分片上传和断点续传（附完整代码）

阿里云OSS实战：用Java SDK实现大文件分片上传和断点续传（附完整代码）

在当今数据爆炸的时代，处理大文件上传已成为后端开发者的必备技能。无论是视频平台的内容上传、设计协作工具的素材同步，还是企业级文档管理系统，都面临着如何高效、稳定传输大文件的挑战。传统单次上传方式在面对网络波动或服务中断时往往束手无策，而阿里云OSS的分片上传技术正是解决这一痛点的利器。

本文将深入剖析分片上传的核心机制，提供完整的Java实现方案。不同于基础教程，我们会重点关注生产环境中可能遇到的真实问题：如何设计可靠的断点续传逻辑、分片大小的黄金分割点选择、上传过程中的异常处理策略等。通过本文，您将掌握一套可直接集成到现有系统的企业级文件上传解决方案。

1. 环境准备与基础配置

1.1 初始化OSS客户端

首先确保已创建OSS Bucket并获取访问凭证。推荐使用RAM子账号的AccessKey，遵循最小权限原则：

// 生产环境建议从配置中心或环境变量获取敏感信息 String endpoint = "https://oss-cn-hangzhou.aliyuncs.com"; String accessKeyId = System.getenv("OSS_ACCESS_KEY_ID"); String accessKeySecret = System.getenv("OSS_ACCESS_KEY_SECRET"); String bucketName = "your-bucket-name"; // 创建OSSClient实例 OSS ossClient = new OSSClientBuilder().build(endpoint, accessKeyId, accessKeySecret);

重要安全提示：永远不要将AccessKey硬编码在代码中或提交到版本控制系统。建议使用KMS加密存储或临时安全令牌(STS)

1.2 分片上传核心参数配置

分片上传的性能与以下参数密切相关：

// 分片配置最佳实践 long partSize = 10 * 1024 * 1024; // 10MB int taskNum = 3; // 并发数 boolean enableCheckpoint = true; File checkpointFile = new File("/upload/checkpoints/" + fileMd5 + ".cp");

2. 分片上传完整实现

2.1 初始化分片上传任务

关键步骤是通过InitiateMultipartUploadRequest获取唯一的uploadId：

public String initiateMultipartUpload(String objectName) { InitiateMultipartUploadRequest request = new InitiateMultipartUploadRequest(bucketName, objectName); // 设置元数据（可选） ObjectMetadata metadata = new ObjectMetadata(); metadata.setContentType("application/octet-stream"); request.setObjectMetadata(metadata); InitiateMultipartUploadResult result = ossClient.initiateMultipartUpload(request); return result.getUploadId(); }

2.2 分片切割与并发上传

采用线程池实现高效并发上传，每个分片独立上传：

ExecutorService executor = Executors.newFixedThreadPool(taskNum); List<Future<PartETag>> futures = new ArrayList<>(); try (FileInputStream fis = new FileInputStream(localFile)) { byte[] buffer = new byte[(int)partSize]; int bytesRead; int partNumber = 1; while ((bytesRead = fis.read(buffer)) > 0) { ByteArrayInputStream partStream = new ByteArrayInputStream(buffer, 0, bytesRead); UploadPartRequest uploadRequest = new UploadPartRequest(); uploadRequest.setBucketName(bucketName); uploadRequest.setKey(objectName); uploadRequest.setUploadId(uploadId); uploadRequest.setPartNumber(partNumber); uploadRequest.setInputStream(partStream); uploadRequest.setPartSize(bytesRead); futures.add(executor.submit(() -> { return ossClient.uploadPart(uploadRequest).getPartETag(); })); partNumber++; } // 等待所有分片完成 List<PartETag> partETags = new ArrayList<>(); for (Future<PartETag> future : futures) { partETags.add(future.get()); } return partETags; } finally { executor.shutdown(); }

2.3 分片合并与完成上传

所有分片上传成功后，需要调用complete接口进行合并：

public void completeMultipartUpload( String objectName, String uploadId, List<PartETag> partETags) { // 分片必须按序号排序 partETags.sort(Comparator.comparingInt(PartETag::getPartNumber)); CompleteMultipartUploadRequest completeRequest = new CompleteMultipartUploadRequest( bucketName, objectName, uploadId, partETags); ossClient.completeMultipartUpload(completeRequest); }

3. 断点续传高级实现

3.1 检查点文件设计

可靠的断点续传需要持久化以下信息：

{ "uploadId": "D3E6B7F12C4A4B7B9D2E1F3G5H6J8K0", "objectName": "videos/2023/sample.mp4", "filePath": "/data/uploads/sample.mp4", "fileMd5": "a1b2c3d4e5f6g7h8i9j0", "partSize": 10485760, "completedParts": [1,2,5], "partETags": { "1": "ETagForPart1", "2": "ETagForPart2" } }

3.2 断点恢复逻辑

上传中断后重新启动时，先加载检查点文件：

public boolean resumeUpload(File checkpointFile) { CheckpointInfo checkpoint = loadCheckpoint(checkpointFile); if (checkpoint == null) return false; // 获取已上传分片列表 ListPartsRequest listPartsRequest = new ListPartsRequest( bucketName, checkpoint.getObjectName(), checkpoint.getUploadId()); PartListing partListing = ossClient.listParts(listPartsRequest); List<PartSummary> existingParts = partListing.getParts(); // 构建需要跳过的分片序号 Set<Integer> skipParts = existingParts.stream() .map(PartSummary::getPartNumber) .collect(Collectors.toSet()); // 重新上传缺失分片 try (FileInputStream fis = new FileInputStream(checkpoint.getFilePath())) { byte[] buffer = new byte[(int)checkpoint.getPartSize()]; int bytesRead; int partNumber = 1; while ((bytesRead = fis.read(buffer)) > 0) { if (skipParts.contains(partNumber)) { partNumber++; fis.skip(bytesRead); continue; } // 上传缺失分片... partNumber++; } return true; } catch (IOException e) { logger.error("Resume upload failed", e); return false; } }

4. 生产环境优化策略

4.1 分片大小动态调整

根据文件类型和网络状况智能调整分片大小：

public long calculateOptimalPartSize(File file, NetworkSpeed speed) { long fileSize = file.length(); long minPartSize = 5 * 1024 * 1024; // 5MB long maxPartSize = 100 * 1024 * 1024; // 100MB // 基础算法：文件大小/100，限制在5MB-100MB之间 long partSize = Math.max(minPartSize, Math.min(maxPartSize, fileSize / 100)); // 根据网络状况调整 if (speed == NetworkSpeed.SLOW) { partSize = Math.min(partSize, 10 * 1024 * 1024); } else if (speed == NetworkSpeed.FAST) { partSize = Math.max(partSize, 20 * 1024 * 1024); } return partSize; }

4.2 上传监控与告警

实现上传进度实时监控：

public class UploadProgressListener implements ProgressListener { private long bytesWritten = 0; private long totalBytes; private long lastLogTime = 0; public UploadProgressListener(long totalBytes) { this.totalBytes = totalBytes; } @Override public void progressChanged(ProgressEvent progressEvent) { bytesWritten += progressEvent.getBytes(); long now = System.currentTimeMillis(); // 每5秒打印一次进度 if (now - lastLogTime > 5000) { double percent = (double)bytesWritten / totalBytes * 100; logger.info(String.format("Upload progress: %.2f%%", percent)); lastLogTime = now; } // 达到阈值触发告警 if (progressEvent.getEventType() == ProgressEventType.TRANSFER_FAILED_EVENT) { alertService.sendAlert("Upload failed at " + bytesWritten + " bytes"); } } }

4.3 客户端SDK封装建议

为方便团队使用，推荐封装为统一的文件服务：

public interface FileUploadService { /** * 上传文件（自动选择普通上传或分片上传） * @param file 待上传文件 * @param objectName OSS对象名称 * @param metadata 文件元数据 * @return 文件访问URL */ String upload(File file, String objectName, ObjectMetadata metadata); /** * 断点续传 * @param checkpointFile 检查点文件路径 * @return 是否恢复成功 */ boolean resumeUpload(String checkpointFile); /** * 获取上传进度 * @param uploadId 上传任务ID * @return 进度百分比(0-100) */ int getUploadProgress(String uploadId); }

5. 完整代码示例

以下整合了所有关键技术的完整实现：

public class AdvancedOssUploader { private final OSS ossClient; private final String bucketName; private final int taskNum; public AdvancedOssUploader(OSS ossClient, String bucketName, int taskNum) { this.ossClient = ossClient; this.bucketName = bucketName; this.taskNum = taskNum; } public String uploadWithResume(File localFile, String objectName) { // 1. 计算文件指纹（用于断点续传标识） String fileMd5 = calculateFileMd5(localFile); File checkpointFile = getCheckpointFile(fileMd5); // 2. 尝试恢复上传 if (checkpointFile.exists()) { boolean resumed = resumeUpload(checkpointFile); if (resumed) return getFileUrl(objectName); } // 3. 初始化分片上传 String uploadId = initiateMultipartUpload(objectName); saveCheckpoint(checkpointFile, uploadId, objectName, localFile); // 4. 执行分片上传（代码见2.2节） List<PartETag> partETags = uploadParts(localFile, objectName, uploadId); // 5. 完成上传 completeMultipartUpload(objectName, uploadId, partETags); deleteCheckpoint(checkpointFile); return getFileUrl(objectName); } private String getFileUrl(String objectName) { Date expiration = new Date(System.currentTimeMillis() + 3600 * 1000 * 24 * 365 * 10); URL url = ossClient.generatePresignedUrl(bucketName, objectName, expiration); return url.toString(); } // 其他辅助方法... }

在实际项目中部署时，建议将检查点文件保存在分布式存储（如Redis或数据库）中，而非本地文件系统，以确保多实例部署时的可靠性。对于超大规模文件（如TB级别），可以考虑结合OSS的TransferAccelerate功能进一步提升上传速度。