1
// 智能文件选择逻辑
2
const handleFileChange = (event) => {
3
  const files = Array.from(event.target.files);
4
  const currentCount = fileList.value.length;
5
  const availableSlots = props.limit - currentCount;
6

7
  if (files.length > availableSlots) {
8
    if (availableSlots <= 0) {
9
      ElMessage.warning(`已达到文件数量上限${props.limit}个`);
10
      return;
11
    } else {
12
      // 智能截取：只选择前面的文件
13
      const originalCount = files.length;
14
      files.splice(availableSlots);
15
      ElMessage.info(
16
        `已自动选择前${availableSlots}个文件，忽略了${originalCount - availableSlots}个文件`
17
      );
18
    }
19
  }
20

21
  // 添加到文件列表
22
  files.forEach(file => {
23
    fileList.value.push({
24
      originalFile: file,
25
      name: file.name,
26
      size: file.size,
27
      status: 'ready',
28
      percentage: 0
29
    });
30
  });
31

32
  emit('change', fileList.value);
33
};

1
// 分批处理核心逻辑
2
const handleBatchUpload = async () => {
3
  // 设置上传状态，防止对话框关闭时清空文件列表
4
  batchUploading.value = true;
5

6
  try {
7
    // 保存总文件数
8
    currentProcessingTotalFiles.value = pendingFiles.value.length;
9

10
    // 计算批次数量
11
    const batchSize = 100;
12
    totalBatches.value = Math.ceil(pendingFiles.value.length / batchSize);
13

14
    ElMessage.success(`开始分批处理，共 ${totalBatches.value} 个批次`);
15

16
    // 关闭文件选择对话框，显示进度框
17
    open.value = false;
18
    showBatchProgress.value = true;
19

20
    // 逐批处理文件
21
    for (let i = 0; i < totalBatches.value; i++) {
22
      await processBatch(i);
23

24
      // 批次间延迟，确保系统稳定
25
      if (i < totalBatches.value - 1) {
26
        console.log(`批次 ${i + 1} 完成，等待3秒后开始下一批次...`);
27
        await new Promise(resolve => setTimeout(resolve, 3000));
28
      }
29
    }
30

31
    // 处理完成统计
32
    const successBatches = batchResults.value.filter(
33
      b => ['SUCCESS', 'PARTIAL_SUCCESS'].includes(b.status)
34
    ).length;
35

36
    if (successBatches === totalBatches.value) {
37
      ElMessage.success('所有批次处理完成！');
38
    } else {
39
      ElMessage.warning(`处理完成，成功批次: ${successBatches}`);
40
    }
41

42
  } catch (error) {
43
    ElMessage.error('批量处理失败: ' + error.message);
44
  } finally {
45
    batchUploading.value = false;
46
    // 清空文件列表
47
    pendingFiles.value = [];
48
  }
49
};

1
// 后台处理状态管理
2
const backgroundProcessing = ref(false);
3
const backgroundResults = ref({
4
  totalFiles: 0,
5
  successCount: 0,
6
  failedCount: 0,
7
  completedBatches: 0,
8
  totalBatches: 0
9
});
10

11
// 进度框关闭处理
12
const handleBatchProgressClose = (visible) => {
13
  if (!visible) {
14
    showBatchProgress.value = false;
15
    progressDialogClosedByUser.value = true;
16

17
    // 如果批次还在进行中，启用后台处理
18
    if (batchProgress.value &&
19
        ['SUBMITTED', 'PROCESSING'].includes(batchProgress.value.status)) {
20
      backgroundProcessing.value = true;
21

22
      backgroundResults.value = {
23
        totalFiles: currentProcessingTotalFiles.value,
24
        successCount: 0,
25
        failedCount: 0,
26
        completedBatches: batchResults.value.filter(
27
          b => ['SUCCESS', 'FAILED', 'PARTIAL_SUCCESS'].includes(b.status)
28
        ).length,
29
        totalBatches: totalBatches.value
30
      };
31

32
      ElMessage.info('批量处理已转入后台执行，完成后将通知您结果');
33
    }
34
  }
35
};
36

37
// 后台处理完成通知
38
const showBackgroundCompletionNotification = () => {
39
  const { totalFiles, successCount, failedCount } = backgroundResults.value;
40
  const successRate = totalFiles > 0 ?
41
    ((successCount / totalFiles) * 100).toFixed(1) : '0.0';
42

43
  ElNotification({
44
    title: '批量文件处理完成',
45
    message: `
46
      <div style="line-height: 1.5;">
47
        <p><strong>处理结果：</strong></p>
48
        <p>📁 总文件数：${totalFiles} 个</p>
49
        <p>✅ 成功处理：${successCount} 个</p>
50
        <p>❌ 处理失败：${failedCount} 个</p>
51
        <p>📊 成功率：${successRate}%</p>
52
      </div>
53
    `,
54
    dangerouslyUseHTMLString: true,
55
    type: successCount === totalFiles ? 'success' :
56
          (successCount > 0 ? 'warning' : 'error'),
57
    duration: 8000,
58
    position: 'top-right'
59
  });
60

61
  backgroundProcessing.value = false;
62
  getList(); // 刷新列表
63
};

1
@Service
2
@Slf4j
3
public class AsyncBatchProcessService {
4

5
    @Autowired
6
    private IThsBaseProjectService thsBaseProjectService;
7

8
    @Autowired
9
    private IThsParsingService thsParsingService;
10

11
    @Autowired
12
    private BatchProcessManager batchProcessManager;
13

14
    // 并发控制：最多同时处理30个文件
15
    private final Semaphore parsingSemaphore = new Semaphore(30);
16

17
    /**
18
     * 异步并行处理批量文件
19
     */
20
    @Async("threadPoolTaskExecutor")
21
    public void processBatchAsync(String batchId, TempFileInfo[] tempFiles,
22
                                  String areaId, String userId) {
23
        try {
24
            log.info("开始异步并行处理批次: {}, 用户ID: {}, 文件数量: {}",
25
                    batchId, userId, tempFiles.length);
26

27
            // 更新批次状态为处理中
28
            batchProcessManager.updateBatchStatus(batchId, "PROCESSING",
29
                    "正在并行处理文件...");
30

31
            // 获取批次进度信息
32
            BatchUploadParseVO batchProgress = batchProcessManager.getBatchProgress(batchId);
33
            if (batchProgress == null) {
34
                log.error("无法获取批次进度信息: {}", batchId);
35
                return;
36
            }
37

38
            // 🚀 核心优化：并行处理所有文件
39
            Arrays.stream(tempFiles)
40
                    .parallel()  // 启用并行流
41
                    .forEach(tempFileInfo -> {
42
                        try {
43
                            // 获取信号量，控制并发数
44
                            parsingSemaphore.acquire();
45

46
                            // 检查批次是否被取消
47
                            BatchUploadParseVO currentProgress =
48
                                    batchProcessManager.getBatchProgress(batchId);
49
                            if (currentProgress == null ||
50
                                "CANCELLED".equals(currentProgress.getStatus())) {
51
                                log.info("批次 {} 已被取消，跳过文件处理: {}",
52
                                        batchId, tempFileInfo.getOriginalFileName());
53
                                return;
54
                            }
55

56
                            // 跳过创建临时文件失败的文件
57
                            if (tempFileInfo == null) {
58
                                log.warn("跳过处理失败的文件: tempFileInfo为null");
59
                                return;
60
                            }
61

62
                            // 找到对应的文件结果对象
63
                            FileProcessResult fileResult =
64
                                    findFileResult(currentProgress, tempFileInfo);
65
                            if (fileResult != null) {
66
                                log.info("开始并行处理文件: {}",
67
                                        tempFileInfo.getOriginalFileName());
68
                                // 异步处理单个文件
69
                                processSingleFileAsync(batchId, tempFileInfo,
70
                                                     areaId, fileResult, userId);
71
                            }
72

73
                        } catch (InterruptedException e) {
74
                            Thread.currentThread().interrupt();
75
                            log.error("处理文件时被中断: {}",
76
                                    tempFileInfo.getOriginalFileName());
77
                        } catch (Exception e) {
78
                            log.error("处理文件异常: {}, 错误: {}",
79
                                    tempFileInfo.getOriginalFileName(), e.getMessage());
80
                        } finally {
81
                            parsingSemaphore.release();
82
                        }
83
                    });
84

85
            // 等待所有文件处理完成后检查最终状态
86
            checkAndUpdateFinalStatus(batchId);
87

88
        } catch (Exception e) {
89
            log.error("批量处理异常，批次ID: {}, 错误: {}", batchId, e.getMessage(), e);
90
            batchProcessManager.updateBatchStatus(batchId, "FAILED",
91
                    "批次处理异常: " + e.getMessage());
92
        }
93
    }
94

95
    /**
96
     * 检查并更新批次最终状态
97
     */
98
    private void checkAndUpdateFinalStatus(String batchId) {
99
        try {
100
            // 等待一段时间让所有异步任务完成
101
            Thread.sleep(3000);
102

103
            BatchUploadParseVO batchProgress =
104
                    batchProcessManager.getBatchProgress(batchId);
105
            if (batchProgress == null) {
106
                return;
107
            }
108

109
            // 统计处理结果
110
            long totalFiles = batchProgress.getTotalFiles();
111
            long successCount = batchProgress.getSuccessFiles().stream()
112
                    .filter(f -> "SUCCESS".equals(f.getStatus()))
113
                    .count();
114
            long failedCount = batchProgress.getSuccessFiles().stream()
115
                    .filter(f -> Arrays.asList("FAILED", "PARSE_FAILED", "PARSE_TIMEOUT")
116
                                       .contains(f.getStatus()))
117
                    .count();
118

119
            // 更新最终状态
120
            if (failedCount == 0 && successCount == totalFiles) {
121
                batchProcessManager.updateBatchStatus(batchId, "SUCCESS",
122
                        "所有文件处理成功");
123
            } else if (successCount > 0) {
124
                batchProcessManager.updateBatchStatus(batchId, "PARTIAL_SUCCESS",
125
                        String.format("部分文件处理成功，成功: %d, 失败: %d",
126
                                    successCount, failedCount));
127
            } else {
128
                batchProcessManager.updateBatchStatus(batchId, "FAILED",
129
                        "所有文件处理失败");
130
            }
131

132
        } catch (Exception e) {
133
            log.error("检查批次最终状态失败: {}", batchId, e);
134
        }
135
    }
136
}

1
@Component
2
@ConfigurationProperties("parsing-service")
3
public class ParsingServiceConfig {
4

5
    // 创建缓存，默认5分钟过期
6
    private TimedCache<String, String> timedCache;
7

8
    // 用于同步获取token的锁对象
9
    private final Object tokenLock = new Object();
10

11
    /**
12
     * 获取授权（优化并发访问）
13
     * 🚀 核心优化：双重检查锁定 + 缓存机制
14
     */
15
    public String getAuthToken() {
16
        // 第一次检查：快速路径，无锁检查缓存
17
        if (timedCache.containsKey(userId)) {
18
            String cachedToken = timedCache.get(userId, false);
19
            if (cachedToken != null) {
20
                log.debug("从缓存获取token成功");
21
                return cachedToken;
22
            }
23
        }
24

25
        // 使用同步块确保只有一个线程去获取token
26
        synchronized (tokenLock) {
27
            // 第二次检查：在锁内再次检查缓存
28
            if (timedCache.containsKey(userId)) {
29
                String cachedToken = timedCache.get(userId, false);
30
                if (cachedToken != null) {
31
                    log.debug("在同步块中从缓存获取token成功");
32
                    return cachedToken;
33
                }
34
            }
35

36
            // 缓存中确实没有token，开始获取新token
37
            log.info("缓存中无有效token，开始向第三方服务获取新token");
38
            TimeInterval timer = DateUtil.timer();
39

40
            try {
41
                // 获取加密后的密码
42
                String encryptedData = encryptionPassWord();
43

44
                // 构建请求参数
45
                Map<String, Object> params = new HashMap<>();
46
                params.put("userId", userId);
47
                params.put("password", encryptedData);
48

49
                // 发起 HTTP 请求
50
                String resultData = HttpUtil.post(apiUrl + authorizeApi, params);
51
                if (resultData == null || resultData.isEmpty()) {
52
                    log.error("认证请求返回空数据");
53
                    return null;
54
                }
55

56
                // 解析返回的 JSON 数据
57
                JSONObject resultJson = JSONUtil.parseObj(resultData);
58
                if (!resultJson.containsKey("code") ||
59
                    !resultJson.get("code").equals(200)) {
60
                    log.error("获取 token 失败，返回数据: {}", resultData);
61
                    return null;
62
                }
63

64
                // 获取 token 并缓存
65
                String token = resultJson.get("data").toString();
66
                if (ObjectUtil.isEmpty(token)) {
67
                    log.error("返回的 token 为空");
68
                    return null;
69
                }
70

71
                // 将 token 存入缓存
72
                timedCache.put(userId, token);
73
                log.info("成功获取并缓存新token，耗时：{} 毫秒", timer.intervalMs());
74
                return token;
75

76
            } catch (Exception e) {
77
                log.error("获取认证异常：{}", e.getMessage(), e);
78
                return null;
79
            }
80
        }
81
    }
82
}

1
@Configuration
2
public class ThreadPoolConfig {
3

4
    @Value("${thread.pool.corePoolSize:50}")
5
    private int corePoolSize;
6

7
    @Value("${thread.pool.maxPoolSize:200}")
8
    private int maxPoolSize;
9

10
    @Value("${thread.pool.queueCapacity:1000}")
11
    private int queueCapacity;
12

13
    @Value("${thread.pool.keepAliveSeconds:300}")
14
    private int keepAliveSeconds;
15

16
    /**
17
     * 高性能线程池配置
18
     */
19
    @Bean(name = "threadPoolTaskExecutor")
20
    public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
21
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
22

23
        // 核心配置
24
        executor.setCorePoolSize(corePoolSize);           // 核心线程数：50
25
        executor.setMaxPoolSize(maxPoolSize);             // 最大线程数：200
26
        executor.setQueueCapacity(queueCapacity);         // 队列容量：1000
27
        executor.setKeepAliveSeconds(keepAliveSeconds);   // 空闲时间：5分钟
28

29
        // 优化配置
30
        executor.setThreadNamePrefix("BatchProcess-");    // 线程命名前缀
31
        executor.setRejectedExecutionHandler(            // 拒绝策略
32
                new ThreadPoolExecutor.CallerRunsPolicy());
33
        executor.setWaitForTasksToCompleteOnShutdown(true); // 优雅关闭
34
        executor.setAwaitTerminationSeconds(60);         // 等待时间
35

36
        executor.initialize();
37
        return executor;
38
    }
39
}

1
@Component
2
@Slf4j
3
public class BatchProcessWebSocketHandler {
4

5
    @Autowired
6
    private SimpMessagingTemplate messagingTemplate;
7

8
    /**
9
     * 推送批次进度更新
10
     */
11
    public void sendBatchProgressUpdate(String userId, BatchUploadParseVO batchProgress) {
12
        try {
13
            String destination = "/user/" + userId + "/queue/batch-progress";
14
            messagingTemplate.convertAndSend(destination, batchProgress);
15
            log.debug("推送批次进度更新到用户: {}, 批次ID: {}", userId, batchProgress.getBatchId());
16
        } catch (Exception e) {
17
            log.error("推送批次进度失败: {}", e.getMessage(), e);
18
        }
19
    }
20

21
    /**
22
     * 推送文件处理进度
23
     */
24
    public void sendFileProgressUpdate(String userId, FileProgressUpdate fileUpdate) {
25
        try {
26
            String destination = "/user/" + userId + "/queue/file-progress";
27
            messagingTemplate.convertAndSend(destination, fileUpdate);
28
            log.debug("推送文件进度更新到用户: {}, 文件: {}",
29
                     userId, fileUpdate.getFileResult().getFileName());
30
        } catch (Exception e) {
31
            log.error("推送文件进度失败: {}", e.getMessage(), e);
32
        }
33
    }
34

35
    /**
36
     * 推送批次完成通知
37
     */
38
    public void sendBatchCompletionNotification(String userId,
39
                                               BatchCompletionNotification notification) {
40
        try {
41
            String destination = "/user/" + userId + "/queue/batch-completion";
42
            messagingTemplate.convertAndSend(destination, notification);
43
            log.info("推送批次完成通知到用户: {}, 成功: {}, 失败: {}",
44
                    userId, notification.getSuccessCount(), notification.getFailedCount());
45
        } catch (Exception e) {
46
            log.error("推送批次完成通知失败: {}", e.getMessage(), e);
47
        }
48
    }
49
}
50

51
// WebSocket配置
52
@Configuration
53
@EnableWebSocketMessageBroker
54
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
55

56
    @Override
57
    public void configureMessageBroker(MessageBrokerRegistry config) {
58
        // 启用简单消息代理
59
        config.enableSimpleBroker("/queue", "/topic");
60
        // 设置应用程序目标前缀
61
        config.setApplicationDestinationPrefixes("/app");
62
        // 设置用户目标前缀
63
        config.setUserDestinationPrefix("/user");
64
    }
65

66
    @Override
67
    public void registerStompEndpoints(StompEndpointRegistry registry) {
68
        // 注册STOMP端点
69
        registry.addEndpoint("/ws")
70
                .setAllowedOriginPatterns("*")
71
                .withSockJS();
72
    }
73
}

1
@Service
2
public class BatchProcessManager {
3

4
    @Autowired
5
    private RedisCache redisCache;
6

7
    private static final String BATCH_PROGRESS_KEY = "batch:progress:";
8

9
    /**
10
     * 保存批次进度到Redis
11
     */
12
    public void saveBatchProgress(String batchId, BatchUploadParseVO batchProgress) {
13
        try {
14
            String key = BATCH_PROGRESS_KEY + batchId;
15
            redisCache.setCacheObject(key, batchProgress, 24, TimeUnit.HOURS);
16
            log.info("批次进度已保存到Redis，批次ID: {}", batchId);
17
        } catch (Exception e) {
18
            log.error("保存批次进度到Redis失败: {}", e.getMessage(), e);
19
            throw new RuntimeException("保存批次进度失败", e);
20
        }
21
    }
22

23
    /**
24
     * 从Redis获取批次进度
25
     */
26
    public BatchUploadParseVO getBatchProgress(String batchId) {
27
        try {
28
            String key = BATCH_PROGRESS_KEY + batchId;
29
            return redisCache.getCacheObject(key);
30
        } catch (Exception e) {
31
            log.error("从Redis获取批次进度失败: {}", e.getMessage(), e);
32
            return null;
33
        }
34
    }
35

36
    /**
37
     * 更新批次状态
38
     */
39
    public void updateBatchStatus(String batchId, String status, String message) {
40
        BatchUploadParseVO batchProgress = getBatchProgress(batchId);
41
        if (batchProgress != null) {
42
            batchProgress.setStatus(status);
43
            batchProgress.setMessage(message);
44
            batchProgress.setUpdateTime(new Date());
45

46
            // 保存到Redis
47
            saveBatchProgress(batchId, batchProgress);
48

49
            // 推送WebSocket消息
50
            webSocketHandler.sendBatchProgressUpdate(
51
                    batchProgress.getUserId(), batchProgress);
52
        }
53
    }
54
}

1
// 1. 并行流处理
2
Arrays.stream(tempFiles)
3
        .parallel()  // 🚀 启用并行处理
4
        .forEach(tempFileInfo -> {
5
            // 处理逻辑
6
        });
7

8
// 2. 信号量控制并发数
9
private final Semaphore parsingSemaphore = new Semaphore(30);
10

11
// 3. 异步方法配置
12
@Async("threadPoolTaskExecutor")
13
public void processBatchAsync(...) {
14
    // 异步处理逻辑
15
}

1
public String getAuthToken() {
2
    // 第一次检查：无锁快速路径
3
    if (timedCache.containsKey(userId)) {
4
        return timedCache.get(userId, false);
5
    }
6

7
    // 同步块：确保只有一个线程获取token
8
    synchronized (tokenLock) {
9
        // 第二次检查：防止重复获取
10
        if (timedCache.containsKey(userId)) {
11
            return timedCache.get(userId, false);
12
        }
13

14
        // 真正的token获取逻辑
15
        String token = fetchTokenFromThirdParty();
16
        timedCache.put(userId, token);
17
        return token;
18
    }
19
}

1
// 1. 及时释放大文件引用
2
private void cleanupTempFiles(TempFileInfo[] tempFiles) {
3
    for (TempFileInfo tempFile : tempFiles) {
4
        if (tempFile != null && tempFile.getTempFile() != null) {
5
            try {
6
                Files.deleteIfExists(tempFile.getTempFile().toPath());
7
            } catch (IOException e) {
8
                log.warn("删除临时文件失败: {}", tempFile.getOriginalFileName());
9
            }
10
        }
11
    }
12
}
13

14
// 2. Redis缓存自动过期
15
redisCache.setCacheObject(key, batchProgress, 24, TimeUnit.HOURS);
16

17
// 3. 线程池合理配置
18
executor.setCorePoolSize(50);    // 核心线程数
19
executor.setMaxPoolSize(200);    // 最大线程数
20
executor.setQueueCapacity(1000); // 队列容量

1
# Dockerfile
2
FROM openjdk:17-jdk-slim
3

4
# 设置工作目录
5
WORKDIR /app
6

7
# 复制jar包
8
COPY target/cost-db-management-*.jar app.jar
9

10
# 设置JVM参数
11
ENV JAVA_OPTS="-Xms2g -Xmx8g -XX:+UseG1GC -XX:MaxGCPauseMillis=200"
12

13
# 暴露端口
14
EXPOSE 8080
15

16
# 启动应用
17
ENTRYPOINT ["sh", "-c", "java $JAVA_OPTS -jar app.jar"]

1
@Component
2
public class BatchProcessMonitor {
3

4
    private final MeterRegistry meterRegistry;
5
    private final Counter batchCounter;
6
    private final Timer batchTimer;
7
    private final Gauge activeThreads;
8

9
    public BatchProcessMonitor(MeterRegistry meterRegistry) {
10
        this.meterRegistry = meterRegistry;
11

12
        // 批次处理计数器
13
        this.batchCounter = Counter.builder("batch.process.count")
14
                .description("批量处理计数")
15
                .register(meterRegistry);
16

17
        // 批次处理耗时
18
        this.batchTimer = Timer.builder("batch.process.duration")
19
                .description("批量处理耗时")
20
                .register(meterRegistry);
21

22
        // 活跃线程数
23
        this.activeThreads = Gauge.builder("batch.process.active.threads")
24
                .description("活跃处理线程数")
25
                .register(meterRegistry, this, BatchProcessMonitor::getActiveThreadCount);
26
    }
27

28
    public void recordBatchStart(String batchId) {
29
        batchCounter.increment();
30
        log.info("批次开始监控: {}", batchId);
31
    }
32

33
    public void recordBatchComplete(String batchId, Duration duration) {
34
        batchTimer.record(duration);
35
        log.info("批次完成监控: {}, 耗时: {}ms", batchId, duration.toMillis());
36
    }
37

38
    private double getActiveThreadCount() {
39
        ThreadPoolTaskExecutor executor =
40
                (ThreadPoolTaskExecutor) applicationContext.getBean("threadPoolTaskExecutor");
41
        return executor.getActiveCount();
42
    }
43
}