JAVA大文件上传源代码

package com.ud.uploadfiledemo.controller; import com.ud.uploadfiledemo.model.ChunkInfo; import com.ud.uploadfiledemo.model.MergeStatus; import lombok.extern.slf4j.Slf4j; import org.apache.commons.io.FileUtils; import org.springframework.beans.factory.annotation.Value; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.*; import org.springframework.web.multipart.MultipartFile; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import java.time.Instant; import java.io.*; import java.nio.channels.FileChannel; import java.nio.file.Path; import java.nio.file.Paths; import java.util.HashSet; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @RestController @RequestMapping("/upload") @Slf4j public class FileUploadController { @Value("${file.upload.path}") private String uploadPath; // 用于记录正在合并的文件 private final Set<String> mergingFiles = ConcurrentHashMap.newKeySet(); // 用于存储合并进度的Map private final ConcurrentHashMap<String, MergeStatus> mergeStatusMap = new ConcurrentHashMap<>(); private final TaskScheduler taskScheduler; public FileUploadController() { ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); scheduler.setPoolSize(1); scheduler.initialize(); this.taskScheduler = scheduler; } @PostMapping("/chunk") public ResponseEntity<String> uploadChunk(@RequestParam("file") MultipartFile file, ChunkInfo chunkInfo) { try { // 创建文件块存储目录 String chunkDirPath = uploadPath + File.separator + chunkInfo.getFileMd5(); File chunkDir = new File(chunkDirPath); if (!chunkDir.exists()) { chunkDir.mkdirs(); } // 写入文件块 File chunkFile = new File(chunkDirPath + File.separator + chunkInfo.getChunkNumber()); file.transferTo(chunkFile); // 检查是否所有块都已上传，并且当前文件不在合并过程中 if (checkIfAllChunksUploaded(chunkInfo) && !mergingFiles.contains(chunkInfo.getFileMd5())) { try { // 标记文件正在合并 mergingFiles.add(chunkInfo.getFileMd5()); mergeChunks(chunkInfo); } finally { // 合并完成后移除标记 mergingFiles.remove(chunkInfo.getFileMd5()); } } return ResponseEntity.ok("Chunk uploaded successfully"); } catch (IOException e) { log.error("Upload chunk failed", e); return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body("Upload failed: " + e.getMessage()); } } @GetMapping("/chunk/verify") public ResponseEntity<Set<Integer>> verifyChunk(@RequestParam String fileMd5) { // 获取已上传的块号 Set<Integer> uploadedChunks = getUploadedChunks(fileMd5); return ResponseEntity.ok(uploadedChunks); } private Set<Integer> getUploadedChunks(String fileMd5) { Set<Integer> uploadedChunks = new HashSet<>(); File chunkDir = new File(uploadPath + File.separator + fileMd5); if (chunkDir.exists()) { File[] files = chunkDir.listFiles(); if (files != null) { for (File file : files) { uploadedChunks.add(Integer.parseInt(file.getName())); } } } return uploadedChunks; } private boolean checkIfAllChunksUploaded(ChunkInfo chunkInfo) { Set<Integer> uploadedChunks = getUploadedChunks(chunkInfo.getFileMd5()); return uploadedChunks.size() == chunkInfo.getTotalChunks(); } @GetMapping("/merge/status") public ResponseEntity<MergeStatus> getMergeStatus(@RequestParam String fileMd5) { MergeStatus status = mergeStatusMap.getOrDefault(fileMd5, new MergeStatus("waiting", 0)); return ResponseEntity.ok(status); } private void mergeChunks(ChunkInfo chunkInfo) throws IOException { String fileMd5 = chunkInfo.getFileMd5(); try { mergeStatusMap.put(fileMd5, new MergeStatus("merging", 0)); String chunkDirPath = uploadPath + File.separator + fileMd5; // 处理文件路径，替换非法字符 String sanitizedFilename = chunkInfo.getFilename().replace('\\', '/'); Path targetPath = Paths.get(uploadPath, sanitizedFilename); File targetDir = targetPath.getParent().toFile(); if (!targetDir.exists()) { targetDir.mkdirs(); } log.info("Merging chunks to: {}", targetPath); // 获取总块数用于计算进度 int totalChunks = chunkInfo.getTotalChunks(); int processedChunks = 0; try (FileChannel outChannel = new FileOutputStream(targetPath.toFile()).getChannel()) { for (int i = 1; i <= totalChunks; i++) { File chunk = new File(chunkDirPath + File.separator + i); if (!chunk.exists()) { throw new IOException("Chunk file missing: " + i); } try (FileChannel inChannel = new FileInputStream(chunk).getChannel()) { inChannel.transferTo(0, inChannel.size(), outChannel); } processedChunks++; // 更新合并进度 int progress = (processedChunks * 100) / totalChunks; mergeStatusMap.put(fileMd5, new MergeStatus("merging", progress)); } } // 删除临时文件 try { Thread.sleep(100); deleteChunkDir(new File(chunkDirPath)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); log.warn("Interrupted while waiting to delete chunks", e); } // 更新状态为完成 mergeStatusMap.put(fileMd5, new MergeStatus("completed", 100)); log.info("File merged successfully: {}", targetPath); } catch (Exception e) { mergeStatusMap.put(fileMd5, new MergeStatus("error", 0, e.getMessage())); throw e; } finally { // 5秒后清除状态 taskScheduler.schedule( () -> mergeStatusMap.remove(fileMd5), Instant.now().plusSeconds(5) ); } } private void deleteChunkDir(File dir) { if (dir.exists()) { File[] files = dir.listFiles(); if (files != null) { for (File file : files) { if (!file.delete()) { log.warn("Failed to delete chunk file: {}", file.getAbsolutePath()); } } } if (!dir.delete()) { log.warn("Failed to delete chunk directory: {}", dir.getAbsolutePath()); } } } }