feat(oss): 增强大文件分片上传的稳定性和可靠性

- 减小分片大小从8MB到5MB以降低单次上传失败概率
- 添加最大重试次数和重试间隔配置
- 实现上传失败后的指数退避重试机制
- 在分片上传过程中增加重试日志记录
- 完善异常处理逻辑，区分业务异常和技术异常
- 优化OSS客户端连接池配置提升并发性能
- 修复批处理写入器中的潜在问题并注释旧实现

ruoyi-common/ruoyi-common-batch/src/main/java/org/dromara/common/batch/processer/GameAthleteProcessor.java

@@ -1,5 +1,8 @@
 package org.dromara.common.batch.processer;
 
+import org.dromara.system.domain.GameAthlete;
+import org.springframework.batch.item.ItemProcessor;
+
 import java.util.Map;
 
 public class GameAthleteProcessor implements ItemProcessor<GameAthlete, GameAthlete> {

ruoyi-common/ruoyi-common-batch/src/main/java/org/dromara/common/batch/writer/GameAthleteWriter.java

@@ -1,7 +1,13 @@
 package org.dromara.common.batch.writer;
 
+import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import lombok.RequiredArgsConstructor;
+import org.dromara.system.domain.GameAthlete;
+import org.dromara.system.mapper.GameAthleteMapper;
+import org.springframework.batch.item.Chunk;
+import org.springframework.batch.item.ItemWriter;
 
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
@@ -13,36 +19,41 @@ public class GameAthleteWriter implements ItemWriter<GameAthlete> {
     private final GameAthleteMapper gameAthleteMapper;
 
     @Override
-    public void write(List<? extends GameAthlete> items) throws Exception {
-        // 1. 提取所有非空的athleteCode
-        Set<String> codes = items.stream()
-            .map(GameAthlete::getAthleteCode)
-            .filter(Objects::nonNull)
-            .collect(Collectors.toSet());
+    public void write(Chunk<? extends GameAthlete> chunk) throws Exception {
 
-        if (!codes.isEmpty()) {
-            // 2. 查询已存在的记录
-            Map<String, Long> codeToIdMap = gameAthleteMapper.selectList(
-                Wrappers.lambdaQuery(GameAthlete.class)
-                    .in(GameAthlete::getAthleteCode, codes)
-                    .select(GameAthlete::getAthleteId, GameAthlete::getAthleteCode)
-            ).stream().collect(Collectors.toMap(
-                GameAthlete::getAthleteCode,
-                GameAthlete::getAthleteId
-            ));
-
-            // 3. 设置已存在记录的ID
-            items.forEach(item -> {
-                if (item.getAthleteCode() != null) {
-                    Long id = codeToIdMap.get(item.getAthleteCode());
-                    if (id != null) {
-                        item.setAthleteId(id);
-                    }
-                }
-            });
-        }
-
-        // 4. 批量保存或更新
-        gameAthleteMapper.insertOrUpdateBatch(items);
     }
+
+//    @Override
+//    public void write(List<? extends GameAthlete> items) throws Exception {
+//        // 1. 提取所有非空的athleteCode
+//        Set<String> codes = items.stream()
+//            .map(GameAthlete::getAthleteCode)
+//            .filter(Objects::nonNull)
+//            .collect(Collectors.toSet());
+//
+//        if (!codes.isEmpty()) {
+//            // 2. 查询已存在的记录
+//            Map<String, Long> codeToIdMap = gameAthleteMapper.selectList(
+//                Wrappers.lambdaQuery(GameAthlete.class)
+//                    .in(GameAthlete::getAthleteCode, codes)
+//                    .select(GameAthlete::getAthleteId, GameAthlete::getAthleteCode)
+//            ).stream().collect(Collectors.toMap(
+//                GameAthlete::getAthleteCode,
+//                GameAthlete::getAthleteId
+//            ));
+//
+//            // 3. 设置已存在记录的ID
+//            items.forEach(item -> {
+//                if (item.getAthleteCode() != null) {
+//                    Long id = codeToIdMap.get(item.getAthleteCode());
+//                    if (id != null) {
+//                        item.setAthleteId(id);
+//                    }
+//                }
+//            });
+//        }
+//
+//        // 4. 批量保存或更新
+//        gameAthleteMapper.insertOrUpdateBatch(items);
+//    }
 }

ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/core/OssClient.java

@@ -96,7 +96,12 @@ public class OssClient {
                 .region(of())
                 .forcePathStyle(isStyle)
                 .httpClient(NettyNioAsyncHttpClient.builder()
-                    .connectionTimeout(Duration.ofSeconds(60)).build())
+                    .connectionTimeout(Duration.ofSeconds(60))
+                    .writeTimeout(Duration.ofMinutes(5))
+                    .readTimeout(Duration.ofMinutes(5))
+                    .maxConcurrency(50)
+                    .maxPendingConnectionAcquires(100)
+                    .build())
                 .build();
 
             //AWS基于 CRT 的 S3 AsyncClient 实例用作 S3 传输管理器的底层客户端

ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/impl/SysOssServiceImpl.java

@@ -56,8 +56,14 @@ public class SysOssServiceImpl implements ISysOssService, OssService {
     // 分片上传阈值：50MB
     private static final long MULTIPART_UPLOAD_THRESHOLD = 50 * 1024 * 1024;
 
-    // 分片大小：8MB
-    private static final long PART_SIZE = 8 * 1024 * 1024;
+    // 分片大小：5MB（减小分片大小以降低单次上传失败概率）
+    private static final long PART_SIZE = 5 * 1024 * 1024;
+    
+    // 最大重试次数
+    private static final int MAX_RETRY_TIMES = 3;
+    
+    // 重试间隔（毫秒）
+    private static final long RETRY_INTERVAL_MS = 1000;
 
     // Redis键前缀
     private static final String UPLOAD_PARTS_PREFIX = "oss:upload:parts:";
@@ -332,35 +338,59 @@ public class SysOssServiceImpl implements ISysOssService, OssService {
         // 生成唯一的文件名，避免冲突
         String fileName = generateUniqueFileName(suffix);
 
-        try {
-            // 初始化分片上传
-            String uploadId = initMultipartUpload(fileName, file.length());
+        int retryCount = 0;
+        Exception lastException = null;
+
+        while (retryCount <= MAX_RETRY_TIMES) {
+            try {
+                // 初始化分片上传
+                String uploadId = initMultipartUpload(fileName, file.length());
 
-            // 计算分片数量
-            long fileSize = file.length();
-            int partCount = (int) Math.ceil((double) fileSize / PART_SIZE);
+                // 计算分片数量
+                long fileSize = file.length();
+                int partCount = (int) Math.ceil((double) fileSize / PART_SIZE);
 
-            List<String> partETags = new ArrayList<>();
+                List<String> partETags = new ArrayList<>();
 
-            // 上传所有分片
-            try (java.io.FileInputStream fis = new java.io.FileInputStream(file)) {
-                for (int i = 1; i <= partCount; i++) {
-                    byte[] partData = new byte[(int) Math.min(PART_SIZE, fileSize - (i - 1) * PART_SIZE)];
-                    int bytesRead = fis.read(partData);
+                // 上传所有分片
+                try (java.io.FileInputStream fis = new java.io.FileInputStream(file)) {
+                    for (int i = 1; i <= partCount; i++) {
+                        byte[] partData = new byte[(int) Math.min(PART_SIZE, fileSize - (i - 1) * PART_SIZE)];
+                        int bytesRead = fis.read(partData);
 
-                    if (bytesRead > 0) {
-                        String eTag = uploadPart(uploadId, i, partData);
-                        partETags.add(eTag);
+                        if (bytesRead > 0) {
+                            String eTag = uploadPart(uploadId, i, partData);
+                            partETags.add(eTag);
+                        }
                     }
                 }
-            }
-
-            // 完成分片上传
-            return completeMultipartUpload(uploadId, partETags, fileName, originalfileName);
 
-        } catch (Exception e) {
-            throw new ServiceException("分片上传失败: " + e.getMessage());
+                // 完成分片上传
+                SysOssVo result = completeMultipartUpload(uploadId, partETags, fileName, originalfileName);
+                log.info("大文件分片上传成功，fileName: {}, 重试次数: {}", fileName, retryCount);
+                return result;
+
+            } catch (ServiceException e) {
+                throw e; // 重新抛出业务异常
+            } catch (Exception e) {
+                lastException = e;
+                retryCount++;
+                log.error("上传大文件失败，fileName: {}, 重试次数: {}, 错误: {}", fileName, retryCount, e.getMessage(), e);
+
+                // 如果不是最后一次重试，则等待一段时间后重试
+                if (retryCount <= MAX_RETRY_TIMES) {
+                    try {
+                        Thread.sleep(RETRY_INTERVAL_MS * retryCount); // 指数退避
+                    } catch (InterruptedException ie) {
+                        Thread.currentThread().interrupt();
+                        throw new ServiceException("上传大文件失败: 线程被中断");
+                    }
+                }
+            }
         }
+
+        // 所有重试都失败
+        throw new ServiceException("分片上传失败: " + lastException.getMessage());
     }
 
     /**
@@ -401,34 +431,55 @@ public class SysOssServiceImpl implements ISysOssService, OssService {
      */
     @Override
     public String uploadPart(String uploadId, int partNumber, byte[] partData) {
-        try {
-            // 获取上传信息
-            Map<String, Object> uploadInfo = RedisUtils.getCacheObject(UPLOAD_INFO_PREFIX + uploadId);
-            if (uploadInfo == null) {
-                throw new ServiceException("上传会话已过期，请重新开始上传");
-            }
-
-            String fileName = (String) uploadInfo.get("fileName");
-            if (fileName == null) {
-                throw new ServiceException("上传信息不完整，请重新开始上传");
-            }
-
-            OssClient storage = OssFactory.instance();
-
-            // 调用OSS的分片上传API
-            String eTag = storage.uploadPart(fileName, uploadId, partNumber, partData);
+        // 获取上传信息
+        Map<String, Object> uploadInfo = RedisUtils.getCacheObject(UPLOAD_INFO_PREFIX + uploadId);
+        if (uploadInfo == null) {
+            throw new ServiceException("上传会话已过期，请重新开始上传");
+        }
 
-            // 在Redis中记录已上传的分片
-            String partsKey = UPLOAD_PARTS_PREFIX + uploadId;
-            RedisUtils.setCacheObject(partsKey + ":" + partNumber, eTag, Duration.ofHours(24));
+        String fileName = (String) uploadInfo.get("fileName");
+        if (fileName == null) {
+            throw new ServiceException("上传信息不完整，请重新开始上传");
+        }
 
-            return eTag;
-        } catch (ServiceException e) {
-            throw e; // 重新抛出业务异常
-        } catch (Exception e) {
-            log.error("上传分片失败，uploadId: {}, partNumber: {}, 错误: {}", uploadId, partNumber, e.getMessage(), e);
-            throw new ServiceException("上传分片失败: " + e.getMessage());
+        OssClient storage = OssFactory.instance();
+        int retryCount = 0;
+        Exception lastException = null;
+
+        while (retryCount <= MAX_RETRY_TIMES) {
+            try {
+                // 调用OSS的分片上传API
+                String eTag = storage.uploadPart(fileName, uploadId, partNumber, partData);
+
+                // 在Redis中记录已上传的分片
+                String partsKey = UPLOAD_PARTS_PREFIX + uploadId;
+                RedisUtils.setCacheObject(partsKey + ":" + partNumber, eTag, Duration.ofHours(24));
+
+                log.info("分片上传成功，uploadId: {}, partNumber: {}, 重试次数: {}", uploadId, partNumber, retryCount);
+                return eTag;
+            } catch (ServiceException e) {
+                // 业务异常不重试，直接抛出
+                throw e;
+            } catch (Exception e) {
+                lastException = e;
+                retryCount++;
+                log.error("上传分片失败，uploadId: {}, partNumber: {}, 重试次数: {}, 错误: {}", 
+                    uploadId, partNumber, retryCount, e.getMessage(), e);
+
+                // 如果不是最后一次重试，则等待一段时间后重试
+                if (retryCount <= MAX_RETRY_TIMES) {
+                    try {
+                        Thread.sleep(RETRY_INTERVAL_MS * retryCount); // 指数退避
+                    } catch (InterruptedException ie) {
+                        Thread.currentThread().interrupt();
+                        throw new ServiceException("上传分片失败: 线程被中断");
+                    }
+                }
+            }
         }
+
+        // 所有重试都失败
+        throw new ServiceException("上传分片失败: " + lastException.getMessage());
     }
 
     /**