docs: 添加项目异步文件上传功能设计文档

2026-03-05 09:11:36 +08:00
parent 014fd8a35c
commit de45854c0f
1 changed files with 560 additions and 0 deletions
--- a/doc/design/2026-03-05-async-file-upload-design.md
+++ b/doc/design/2026-03-05-async-file-upload-design.md
@@ -0,0 +1,560 @@
+# 项目异步文件上传功能 - 设计文档
+
+## 文档信息
+- **创建日期**: 2026-03-05
+- **版本**: v1.0
+- **作者**: Claude
+- **状态**: 已批准
+
+## 1. 概述
+
+### 1.1 功能描述
+实现项目流水文件的异步批量上传功能，支持文件上传到流水分析平台、轮询解析状态、获取解析结果、保存流水数据到本地数据库的完整流程。
+
+### 1.2 核心需求
+- 批量上传流水文件（最多100个文件）
+- 异步处理每个文件的上传→解析→存储流程
+- 线程池容量100，超载时等待30秒重试
+- 实时跟踪文件处理状态
+- 生成独立的批次日志文件便于维护
+
+### 1.3 技术栈
+- Spring @Async 异步处理
+- ThreadPoolTaskExecutor 线程池
+- MyBatis Plus 批量操作
+- Logback 自定义日志
+- Vue + Element UI 前端
+
+## 2. 数据库设计
+
+### 2.1 文件上传记录表
+
+```sql
+CREATE TABLE `ccdi_file_upload_record` (
+  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键ID',
+  `project_id` bigint(20) NOT NULL COMMENT '项目ID',
+  `lsfx_project_id` int(11) DEFAULT NULL COMMENT '流水分析平台项目ID',
+  `log_id` int(11) DEFAULT NULL COMMENT '流水分析平台返回的logId',
+  `file_name` varchar(255) NOT NULL COMMENT '文件名称',
+  `file_size` bigint(20) DEFAULT NULL COMMENT '文件大小（字节）',
+  `file_status` varchar(20) NOT NULL COMMENT '文件状态：uploading-上传中，parsing-解析中，parsed_success-解析成功，parsed_failed-解析失败',
+  `enterprise_names` text COMMENT '主体名称（多个用逗号分隔）',
+  `account_nos` text COMMENT '主体账号（多个用逗号分隔）',
+  `error_message` text COMMENT '错误信息（解析失败时记录）',
+  `upload_time` datetime NOT NULL COMMENT '上传时间',
+  `upload_user` varchar(64) NOT NULL COMMENT '上传人',
+  PRIMARY KEY (`id`),
+  KEY `idx_project_id` (`project_id`),
+  KEY `idx_log_id` (`log_id`),
+  KEY `idx_file_status` (`file_status`),
+  KEY `idx_upload_time` (`upload_time`),
+  KEY `idx_project_status` (`project_id`, `file_status`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='项目文件上传记录表';
+```
+
+### 2.2 字段说明
+
+| 字段 | 类型 | 说明 | 备注 |
+|------|------|------|------|
+| id | bigint | 主键ID | 自增 |
+| project_id | bigint | 项目ID | 外键关联 ccdi_project |
+| lsfx_project_id | int | 流水分析平台项目ID | 用于调用流水分析接口 |
+| log_id | int | 流水分析平台返回的logId | 关键字段，用于查询解析状态和流水数据 |
+| file_name | varchar(255) | 文件名称 | 原始文件名 |
+| file_size | bigint | 文件大小 | 字节数 |
+| file_status | varchar(20) | 文件状态 | uploading/parsing/parsed_success/parsed_failed |
+| enterprise_names | text | 主体名称 | 解析成功后存储，多个用逗号分隔 |
+| account_nos | text | 主体账号 | 解析成功后存储，多个用逗号分隔 |
+| error_message | text | 错误信息 | 解析失败时记录原因 |
+| upload_time | datetime | 上传时间 | 记录创建时间 |
+| upload_user | varchar(64) | 上传人 | 操作用户 |
+
+## 3. 后端架构设计
+
+### 3.1 模块结构
+
+```
+ccdi-project/src/main/java/com/ruoyi/ccdi/project/
+├── controller/
+│   └── CcdiFileUploadController.java          # 文件上传接口
+├── service/
+│   ├── ICcdiFileUploadService.java            # 文件上传服务接口
+│   └── impl/
+│       └── CcdiFileUploadServiceImpl.java     # 文件上传服务实现
+├── mapper/
+│   └── CcdiFileUploadRecordMapper.java        # 文件上传记录Mapper
+├── domain/
+│   ├── entity/
+│   │   └── CcdiFileUploadRecord.java          # 文件上传记录实体
+│   ├── dto/
+│   │   └── CcdiFileUploadQueryDTO.java        # 查询DTO
+│   └── vo/
+│       ├── CcdiFileUploadVO.java              # 文件上传响应VO
+│       └── CcdiFileUploadStatisticsVO.java    # 统计VO
+├── config/
+│   └── AsyncThreadPoolConfig.java             # 异步线程池配置
+└── log/
+    └── FileUploadLogAppender.java             # 自定义日志Appender
+
+ccdi-project/src/main/resources/
+└── mapper/ccdi/project/
+    └── CcdiFileUploadRecordMapper.xml         # Mapper XML映射文件
+```
+
+### 3.2 Controller 接口设计
+
+| 接口路径 | 方法 | 功能 | 参数 | 返回值 |
+|---------|------|------|------|--------|
+| `/ccdi/file-upload/batch` | POST | 批量上传文件 | projectId, files[] | batchId |
+| `/ccdi/file-upload/list` | GET | 查询上传记录列表 | projectId, fileStatus, pageNum, pageSize | 分页列表 |
+| `/ccdi/file-upload/statistics/{projectId}` | GET | 查询上传统计 | projectId | 各状态数量 |
+| `/ccdi/file-upload/detail/{id}` | GET | 查询记录详情 | id | 完整信息 |
+| `/ccdi/file-upload/thread-pool/status` | GET | 查询线程池状态 | - | 线程池状态信息 |
+
+### 3.3 Service 核心方法
+
+#### ICcdiFileUploadService 接口
+
+```java
+public interface ICcdiFileUploadService {
+    /**
+     * 批量上传文件
+     * @param projectId 项目ID
+     * @param files 文件数组
+     * @param username 上传人
+     * @return 批次ID
+     */
+    String batchUploadFiles(Long projectId, MultipartFile[] files, String username);
+
+    /**
+     * 异步处理单个文件
+     * @Async("fileUploadExecutor")
+     */
+    void processFileAsync(Long projectId, Integer lsfxProjectId, MultipartFile file,
+                          Long recordId, String batchId, CcdiFileUploadRecord record);
+
+    /**
+     * 查询上传记录列表
+     */
+    Page<CcdiFileUploadRecord> selectPage(Page<CcdiFileUploadRecord> page,
+                                          CcdiFileUploadQueryDTO queryDTO);
+
+    /**
+     * 统计各状态文件数量
+     */
+    Map<String, Long> countByStatus(Long projectId);
+}
+```
+
+#### 核心处理流程
+
+```java
+// 1. batchUploadFiles - 主入口
+String batchUploadFiles(Long projectId, MultipartFile[] files, String username) {
+    // 1.1 生成批次ID
+    String batchId = UUID.randomUUID().toString().replace("-", "");
+
+    // 1.2 获取项目的 lsfxProjectId
+    Integer lsfxProjectId = project.getLsfxProjectId();
+
+    // 1.3 批量插入文件记录（status=uploading）
+    List<CcdiFileUploadRecord> records = createRecords(projectId, lsfxProjectId, files, username);
+    recordMapper.insertBatch(records);
+
+    // 1.4 异步启动调度线程提交任务
+    CompletableFuture.runAsync(() -> {
+        submitTasksAsync(projectId, lsfxProjectId, files, records, batchId);
+    });
+
+    // 1.5 立即返回 batchId
+    return batchId;
+}
+
+// 2. submitTasksAsync - 调度线程
+void submitTasksAsync(Long projectId, Integer lsfxProjectId, MultipartFile[] files,
+                      List<CcdiFileUploadRecord> records, String batchId) {
+    // 2.1 创建批次日志文件
+    FileUploadLogAppender.createBatchLogFile(projectId, batchId);
+
+    // 2.2 循环提交任务，支持重试
+    for (int i = 0; i < files.length; i++) {
+        boolean submitted = false;
+        int retryCount = 0;
+
+        while (!submitted && retryCount < 2) {
+            try {
+                // 提交异步任务到线程池
+                CompletableFuture.runAsync(
+                    () -> processFileAsync(projectId, lsfxProjectId, files[i],
+                                           records.get(i).getId(), batchId, records.get(i)),
+                    fileUploadExecutor
+                );
+                submitted = true;
+            } catch (RejectedExecutionException e) {
+                retryCount++;
+                if (retryCount == 1) {
+                    Thread.sleep(30000); // 等待30秒
+                } else {
+                    // 重试失败，更新记录状态
+                    updateRecordStatus(records.get(i).getId(), "parsed_failed", "系统繁忙");
+                }
+            }
+        }
+    }
+}
+
+// 3. processFileAsync - 文件处理线程
+@Async("fileUploadExecutor")
+void processFileAsync(Long projectId, Integer lsfxProjectId, MultipartFile file,
+                      Long recordId, String batchId, CcdiFileUploadRecord record) {
+    try {
+        // 3.1 上传文件到流水分析平台
+        UploadFileResponse uploadResponse = lsfxClient.uploadFile(lsfxProjectId, file);
+        Integer logId = uploadResponse.getData().getLogId();
+
+        // 3.2 更新状态为 parsing
+        record.setLogId(logId);
+        record.setFileStatus("parsing");
+        recordMapper.updateById(record);
+
+        // 3.3 轮询解析状态（最多300次，间隔2秒）
+        boolean parsingComplete = waitForParsingComplete(lsfxProjectId, logId.toString());
+
+        // 3.4 获取文件上传状态
+        GetFileUploadStatusResponse statusResponse = lsfxClient.getFileUploadStatus(...);
+
+        // 3.5 判断解析结果
+        if (status == -5 && desc == "data.wait.confirm.newaccount") {
+            // 解析成功
+            record.setFileStatus("parsed_success");
+            record.setEnterpriseNames(...);
+            record.setAccountNos(...);
+            recordMapper.updateById(record);
+
+            // 3.6 获取流水数据并批量保存
+            fetchAndSaveBankStatements(projectId, lsfxProjectId, logId, totalCount);
+        } else {
+            // 解析失败
+            record.setFileStatus("parsed_failed");
+            record.setErrorMessage(...);
+            recordMapper.updateById(record);
+        }
+    } catch (Exception e) {
+        updateRecordStatus(recordId, "parsed_failed", e.getMessage());
+    }
+}
+```
+
+## 4. 线程池配置
+
+### 4.1 配置类
+
+```java
+@Configuration
+@EnableAsync
+public class AsyncThreadPoolConfig {
+
+    @Bean("fileUploadExecutor")
+    public Executor fileUploadExecutor() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(100);              // 核心线程数
+        executor.setMaxPoolSize(100);               // 最大线程数
+        executor.setQueueCapacity(0);               // 队列容量（0表示不使用队列）
+        executor.setThreadNamePrefix("file-upload-"); // 线程名称前缀
+        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy()); // 拒绝策略
+        executor.setKeepAliveSeconds(60);           // 空闲线程存活时间
+        executor.setWaitForTasksToCompleteOnShutdown(true); // 等待任务完成再关闭
+        executor.setAwaitTerminationSeconds(60);    // 最长等待时间
+        executor.initialize();
+        return executor;
+    }
+}
+```
+
+### 4.2 拒绝策略
+
+- **策略**: AbortPolicy
+- **行为**: 抛出 RejectedExecutionException
+- **处理**: 调度线程捕获异常，等待30秒后重试1次
+- **重试失败**: 更新记录状态为 `parsed_failed`，错误信息"系统繁忙"
+
+## 5. 日志管理
+
+### 5.1 日志文件组织
+
+- **路径格式**: `logs/file-upload/{projectId}/{timestamp}.log`
+- **示例**: `logs/file-upload/123/20260305-103025.log`
+- **特点**: 每个批次生成独立的日志文件
+
+### 5.2 Logback 配置
+
+```xml
+<!-- logback-fileupload.xml -->
+<appender name="FILE_UPLOAD" class="com.ruoyi.ccdi.project.log.FileUploadLogAppender">
+    <layout class="ch.qos.logback.classic.PatternLayout">
+        <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n</pattern>
+    </layout>
+    <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+        <fileNamePattern>logs/file-upload/%d{yyyy-MM-dd}/%d{HH}.log</fileNamePattern>
+        <maxHistory>30</maxHistory>
+        <maxFileSize>100MB</maxFileSize>
+    </rollingPolicy>
+</appender>
+
+<logger name="com.ruoyi.ccdi.project.service.impl.CcdiFileUploadServiceImpl"
+        level="INFO" additivity="false">
+    <appender-ref ref="FILE_UPLOAD"/>
+</logger>
+```
+
+### 5.3 自定义 Appender
+
+```java
+public class FileUploadLogAppender extends UnsynchronizedAppenderBase<ILoggingEvent> {
+
+    private static final ThreadLocal<FileAppender<ILoggingEvent>> currentAppender =
+        new ThreadLocal<>();
+
+    /**
+     * 为指定批次创建独立的日志文件
+     */
+    public static void createBatchLogFile(Long projectId, String batchId) {
+        String timestamp = new SimpleDateFormat("yyyyMMdd-HHmmss").format(new Date());
+        String logPath = String.format("logs/file-upload/%d/%s.log", projectId, timestamp);
+
+        FileAppender<ILoggingEvent> appender = new FileAppender<>();
+        appender.setFile(logPath);
+        appender.setLayout(...);
+        appender.start();
+
+        currentAppender.set(appender);
+    }
+
+    @Override
+    protected void append(ILoggingEvent event) {
+        FileAppender<ILoggingEvent> appender = currentAppender.get();
+        if (appender != null) {
+            appender.doAppend(event);
+        }
+    }
+}
+```
+
+## 6. 前端交互设计
+
+### 6.1 上传流程
+
+```
+用户选择文件 → 确认上传 → 显示loading
+    ↓
+调用 batchUploadFiles() API
+    ↓
+后端立即返回 batchId
+    ↓
+前端提示"上传任务已提交"
+    ↓
+跳转到上传记录列表页
+    ↓
+每5秒自动刷新列表（可关闭）
+```
+
+### 6.2 列表页展示
+
+**统计卡片:**
+- 上传中: 2
+- 解析中: 3
+- 解析成功: 15
+- 解析失败: 1
+
+**文件列表:**
+
+| 文件名 | 大小 | 状态 | 主体名称 | 上传时间 | 操作 |
+|--------|------|------|----------|----------|------|
+| 流水1.xlsx | 2.5MB | 🔄 解析中 | - | 10:30:25 | - |
+| 流水2.xlsx | 1.8MB | ✅ 解析成功 | 张三,李四 | 10:28:15 | 查看流水 |
+| 流水3.xlsx | 3.2MB | ❌ 解析失败 | - | 10:25:30 | 查看错误 |
+
+### 6.3 API 接口
+
+```javascript
+// 批量上传文件
+POST /ccdi/file-upload/batch
+参数: FormData(projectId, files[])
+返回: { code: 200, msg: "上传任务已提交", data: batchId }
+
+// 查询上传记录列表
+GET /ccdi/file-upload/list
+参数: { projectId, fileStatus, pageNum, pageSize }
+返回: { rows: [], total: 100 }
+
+// 查询上传统计
+GET /ccdi/file-upload/statistics/{projectId}
+返回: { uploading: 2, parsing: 3, parsed_success: 15, parsed_failed: 1 }
+```
+
+## 7. 异常处理
+
+### 7.1 Controller 层异常
+
+| 异常类型 | 处理方式 | 返回信息 |
+|---------|---------|---------|
+| 参数为空 | 参数校验 | "项目ID不能为空" |
+| 文件数量超限 | 参数校验 | "单次最多上传100个文件" |
+| 文件大小超限 | 参数校验 | "文件超过50MB限制" |
+| 文件格式错误 | 参数校验 | "仅支持Excel文件" |
+| 项目不存在 | 业务校验 | "项目不存在" |
+
+### 7.2 Service 层异常
+
+| 异常类型 | 处理方式 | 记录状态 |
+|---------|---------|---------|
+| 流水分析平台接口异常 | 捕获并记录 | parsed_failed |
+| 轮询超时（>300次） | 捕获并记录 | parsed_failed |
+| 文件解析失败 | 捕获并记录 | parsed_failed |
+| 线程池满且重试失败 | 捕获并记录 | parsed_failed |
+| 其他未知异常 | 捕获并记录 | parsed_failed |
+
+### 7.3 异常处理代码示例
+
+```java
+try {
+    // 处理文件
+    processFileInternal(projectId, lsfxProjectId, file, record);
+} catch (LsfxApiException e) {
+    log.error("流水分析平台接口异常", e);
+    updateRecordStatus(recordId, "parsed_failed", "流水分析平台接口异常：" + e.getMessage());
+} catch (InterruptedException e) {
+    Thread.currentThread().interrupt();
+    log.error("处理被中断", e);
+    updateRecordStatus(recordId, "parsed_failed", "处理被中断");
+} catch (Exception e) {
+    log.error("处理失败（未知异常）", e);
+    updateRecordStatus(recordId, "parsed_failed", "处理失败：" + e.getMessage());
+}
+```
+
+## 8. 性能优化
+
+### 8.1 数据库优化
+
+**索引建议:**
+```sql
+-- 组合索引提升查询性能
+ALTER TABLE ccdi_file_upload_record
+ADD INDEX idx_project_status (project_id, file_status);
+
+ALTER TABLE ccdi_bank_statement
+ADD INDEX idx_project_log (project_id, batch_id);
+```
+
+**批量插入:**
+- 使用 MyBatis Plus 的 `saveBatch(statements, 500)`
+- 每批500条，避免单次插入过多数据
+
+### 8.2 轮询优化
+
+**动态间隔策略:**
+- 前10次：1秒间隔
+- 11-50次：2秒间隔
+- 51次后：5秒间隔
+
+### 8.3 线程池监控
+
+```java
+@GetMapping("/thread-pool/status")
+public AjaxResult getThreadPoolStatus() {
+    ThreadPoolExecutor pool = fileUploadExecutor.getThreadPoolExecutor();
+
+    Map<String, Object> status = new HashMap<>();
+    status.put("activeCount", pool.getActiveCount());
+    status.put("corePoolSize", pool.getCorePoolSize());
+    status.put("queueSize", pool.getQueue().size());
+    status.put("completedTaskCount", pool.getCompletedTaskCount());
+
+    return AjaxResult.success(status);
+}
+```
+
+## 9. 测试场景
+
+### 9.1 功能测试
+
+| 场景 | 输入 | 预期结果 |
+|------|------|---------|
+| 正常上传 | 10个Excel文件，每个5MB | 所有文件处理成功 |
+| 大文件上传 | 1个50MB文件 | 处理成功 |
+| 文件数量超限 | 101个文件 | 返回错误提示 |
+| 文件格式错误 | 上传PDF文件 | 返回错误提示 |
+| 解析失败 | 格式错误的Excel | 状态更新为parsed_failed |
+
+### 9.2 压力测试
+
+| 场景 | 并发数 | 预期结果 |
+|------|--------|---------|
+| 正常并发 | 100个线程同时上传 | 所有任务正常处理 |
+| 超载测试 | 150个文件同时上传 | 超过100的文件等待30秒重试 |
+| 持续运行 | 1000次循环上传 | 无内存泄漏，无线程死锁 |
+
+### 9.3 边界测试
+
+| 场景 | 操作 | 预期结果 |
+|------|------|---------|
+| 项目被删除 | 上传中删除项目 | 任务取消，状态更新为失败 |
+| 重复上传 | 同一文件上传2次 | 生成2条独立记录和logId |
+| 网络中断 | 轮询时网络断开 | 捕获异常，状态更新为失败 |
+
+## 10. 部署注意事项
+
+### 10.1 配置检查清单
+
+- [ ] 线程池容量配置（默认100）
+- [ ] 文件上传大小限制（默认50MB）
+- [ ] 日志文件路径权限
+- [ ] 数据库索引创建
+- [ ] 流水分析平台地址配置
+- [ ] 应用认证信息配置
+
+### 10.2 监控指标
+
+- 线程池活跃线程数
+- 文件上传成功率（parsed_success / total）
+- 平均处理时长
+- 线程池拒绝次数
+- 日志文件大小和数量
+
+### 10.3 运维建议
+
+- 定期清理30天前的日志文件
+- 监控线程池状态，必要时调整容量
+- 关注数据库连接池使用情况
+- 流水分析平台接口调用成功率监控
+
+## 11. 附录
+
+### 11.1 状态机转换
+
+```
+uploading (初始状态)
+    ↓
+parsing (上传成功，轮询中)
+    ↓
+parsed_success (解析成功) 或 parsed_failed (解析失败)
+```
+
+### 11.2 关键时序
+
+- 文件上传：2-5秒（取决于文件大小）
+- 轮询解析：最多10分钟（300次 × 2秒）
+- 获取流水数据：1-3分钟（取决于流水数量）
+- 总处理时长：约3-15分钟/文件
+
+### 11.3 数据量估算
+
+- 单个Excel文件：平均5000条流水
+- 100个文件：约50万条流水
+- 数据库存储：约200MB
+- 日志文件：约5-10MB/批次
+
+---
+
+**文档结束**