Merge remote-tracking branch 'origin/hotfix-20250525-mq' into feature_20250521_nameplate_difyResult

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/AsyncTaskExecutePool.java

@@ -1,68 +0,0 @@
-package com.volvo.ai.analytic.center.config;
-
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.context.annotation.Bean;
-import org.springframework.scheduling.annotation.AsyncConfigurer;
-import org.springframework.scheduling.annotation.EnableAsync;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
-import org.springframework.stereotype.Component;
-
-import java.util.concurrent.*;
-
-/**
- * 异步任务线程池装配类
- * @author gubin
- * @date 2022-04-14
- */
-@EnableAsync
-@Slf4j
-@Component
-public class AsyncTaskExecutePool implements AsyncConfigurer {
-
-    @Value("${task.pool.corePoolSize}")
-    private int corePoolSize;
-
-    @Value("${task.pool.maxPoolSize}")
-    private int maxPoolSize;
-
-    @Value("${task.pool.queueCapacity}")
-    private int queueCapacity;
-
-    @Value("${task.pool.keepAliveSeconds}")
-    private int keepAliveSeconds;
-
-
-    @Bean
-    @Override
-    public Executor getAsyncExecutor() {
-        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
-        //核心线程池大小
-        executor.setCorePoolSize(corePoolSize);
-        //最大线程数
-        executor.setMaxPoolSize(maxPoolSize);
-        //队列容量
-        executor.setQueueCapacity(queueCapacity);
-        //活跃时间
-        executor.setKeepAliveSeconds(keepAliveSeconds);
-        //线程名字前缀
-        executor.setThreadNamePrefix("async-task-");
-        // setRejectedExecutionHandler：当pool已经达到max size的时候，如何处理新任务
-        // CallerRunsPolicy：不在新线程中执行任务，而是由调用者所在的线程来执行
-        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
-        executor.initialize();
-        return executor;
-    }
-
-    @Override
-    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
-        return (throwable, method, objects) -> {
-            log.error("===="+throwable.getMessage()+"====", throwable);
-            log.error("exception method:"+method.getName());
-        };
-    }
-
-
-
-}

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java

@@ -56,12 +56,12 @@ public class CorpusProcessKafkaProducer {
 	@KafkaListener(topics = "${spring.kafka.topic}", groupId = "${spring.kafka.group}")
 	public void listen(List<ConsumerRecord<String, Object>> recordMessages) {
 		long startTime = System.currentTimeMillis();
+		log.info("CorpusProcessKafkaProducer Received message: {}", recordMessages);
+		// 获取消息列表
+		int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 2;
+		log.info("获取的线程数：{}", optimalThreadPoolSize);
+		ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize);
 		try {
-			log.info("CorpusProcessKafkaProducer Received message: {}", recordMessages);
-			// 获取消息列表
-			int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 2;
-			log.info("获取的线程数：{}", optimalThreadPoolSize);
-			ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize);
 			if(CollectionUtils.isNotEmpty(recordMessages)){
 				log.info("CorpusProcessKafkaProducer List size: {}", recordMessages.size());
 				for (ConsumerRecord<String, Object> record : recordMessages) {
@@ -100,6 +100,8 @@ public class CorpusProcessKafkaProducer {
 							}
 						} catch (Exception e) {
 							log.error("CorpusProcessKafkaProducer 电话语料 解析JSON出错: {}", e.getMessage());
+						}finally {
+							executor.shutdown();
 						}
 					});
 
@@ -111,6 +113,8 @@ public class CorpusProcessKafkaProducer {
 			// 在这里可以添加对解析后的对象的进一步处理逻辑
 		} catch (Exception e) {
 			log.error("CorpusProcessKafkaProducer 电话语料 解析JSON出错: {}" , e.getMessage());
+		}finally {
+			executor.shutdown();
 		}
 	}
 

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/MqMessageRecordServiceImpl.java

@@ -38,9 +38,6 @@ import org.springframework.util.CollectionUtils;
 import java.text.SimpleDateFormat;
 import java.time.LocalDateTime;
 import java.util.*;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Executor;
 import java.util.stream.Collectors;
 
 @Slf4j
@@ -92,8 +89,6 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
     @Autowired
     private AiAnalyticBusinessConfigMapper aiAnalyticBusinessConfigMapper;
 
-    @Autowired
-    private Executor getAsyncExecutor;
 
     /**
      * 处理Mq消息
@@ -156,23 +151,16 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
         return true;
     }
 
-    private void processDify(DifyCommunityTargetDTO difyCommunityTargetDTO ,String user, JSONArray difyResult, String aiAnalysisRequestId) throws InterruptedException, ExecutionException {
+    private void processDify(DifyCommunityTargetDTO difyCommunityTargetDTO ,String user, JSONArray difyResult, String aiAnalysisRequestId) {
         //舆情案件分析
-        CompletableFuture<JSONObject> caseWorkFlow = CompletableFuture.supplyAsync(() -> callCaseCommunityWorkFlow(difyCommunityTargetDTO,user, caseToken),getAsyncExecutor);
+        JSONObject caseResult =  callCaseCommunityWorkFlow(difyCommunityTargetDTO,user, caseToken);
 
         // 内容主题关键词打标
-        CompletableFuture<JSONObject> keywordWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(difyCommunityTargetDTO,user, keywordToken),getAsyncExecutor);
+        JSONObject keywordResult =   callCommunityWorkFlow(difyCommunityTargetDTO,user, keywordToken);
 
         // litecrm线索分析
-        CompletableFuture<JSONObject> clueAnalysisWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(difyCommunityTargetDTO,user, clueAnalysisToken),getAsyncExecutor);
+        JSONObject clueAnalysisResult = callCommunityWorkFlow(difyCommunityTargetDTO,user, clueAnalysisToken);
 
-        CompletableFuture<Void> allFutures = CompletableFuture.allOf(caseWorkFlow, keywordWorkFlow, clueAnalysisWorkFlow);
-        // 等待所有API调用完成
-        allFutures.get();
-        // 获取各个API的结果
-        JSONObject caseResult = caseWorkFlow.get();
-        JSONObject keywordResult = keywordWorkFlow.get();
-        JSONObject clueAnalysisResult = clueAnalysisWorkFlow.get();
         difyResult.add(caseResult);
         difyResult.add(keywordResult);
         difyResult.add(clueAnalysisResult);

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java

@@ -124,27 +124,34 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
 		// 创建线程池
 		ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize); // 根据需求调整线程池大小
 
-		for (int i = 1; i <= totalPages; i++) {
+        try {
-			int offset = (i - 1) * pageSize;
+            for (int i = 1; i <= totalPages; i++) {
-			List<OdsVdqwMessageOTD> messageList = tmOdsVdqwMessagearchivingMapper.queryOdsVdqwMessageByData(statTime, endTime, offset, pageSize, retry);
+                int offset = (i - 1) * pageSize;
-			// 处理查询到的数据
+                List<OdsVdqwMessageOTD> messageList = tmOdsVdqwMessagearchivingMapper.queryOdsVdqwMessageByData(statTime, endTime, offset, pageSize, retry);
-			// 使用 CompletableFuture 并行处理
+                // 处理查询到的数据
-			CompletableFuture<?>[] futures = messageList.stream()
+                // 使用 CompletableFuture 并行处理
-					.map(item -> CompletableFuture.runAsync(() -> {
+                CompletableFuture<?>[] futures = messageList.stream()
-						try {
+                        .map(item -> CompletableFuture.runAsync(() -> {
-							processItem(item, finalStatTime, finalEndTime);
+                            try {
-						} catch (Exception e) {
+                                processItem(item, finalStatTime, finalEndTime);
-							log.error("处理企微语料失败: FromUserId={}, AcceptUserId={}, 异常: {}",
+                            } catch (Exception e) {
-									item.getFromUserId(), item.getAcceptUserId(), e.getMessage(), e);
+                                log.error("处理企微语料失败: FromUserId={}, AcceptUserId={}, 异常: {}",
-						}
+                                        item.getFromUserId(), item.getAcceptUserId(), e.getMessage(), e);
-					}, executor))
+                            }
-					.toArray(CompletableFuture[]::new);
+                        }, executor))
-			// 等待所有任务完成
+                        .toArray(CompletableFuture[]::new);
-			CompletableFuture.allOf(futures).join();
+                // 等待所有任务完成
-		}
+                CompletableFuture.allOf(futures).join();
-		// 关闭线程池
+            }
-		executor.shutdown();
+            // 关闭线程池
-		log.info("企微数据跑批结束 耗时：{}",System.currentTimeMillis()-startTime);
+            executor.shutdown();
+        } catch (Exception e) {
+			log.error("企微数据跑批异常",e);
+        } finally {
+			// 关闭线程池
+			executor.shutdown();
+        }
+        log.info("企微数据跑批结束 耗时：{}",System.currentTimeMillis()-startTime);
 	}
 
 	private void processItem(OdsVdqwMessageOTD item, String statTime, String endTime) {