diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/AsyncTaskExecutePool.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/AsyncTaskExecutePool.java deleted file mode 100644 index 51254a4..0000000 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/AsyncTaskExecutePool.java +++ /dev/null @@ -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()); - }; - } - - - -} diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java index dce3d71..047ce3e 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java +++ b/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> 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 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(); } } diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/MqMessageRecordServiceImpl.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/MqMessageRecordServiceImpl.java index 5e2ce92..2c92322 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/MqMessageRecordServiceImpl.java +++ b/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 caseWorkFlow = CompletableFuture.supplyAsync(() -> callCaseCommunityWorkFlow(difyCommunityTargetDTO,user, caseToken),getAsyncExecutor); + JSONObject caseResult = callCaseCommunityWorkFlow(difyCommunityTargetDTO,user, caseToken); // 内容主题关键词打标 - CompletableFuture keywordWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(difyCommunityTargetDTO,user, keywordToken),getAsyncExecutor); + JSONObject keywordResult = callCommunityWorkFlow(difyCommunityTargetDTO,user, keywordToken); // litecrm线索分析 - CompletableFuture clueAnalysisWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(difyCommunityTargetDTO,user, clueAnalysisToken),getAsyncExecutor); + JSONObject clueAnalysisResult = callCommunityWorkFlow(difyCommunityTargetDTO,user, clueAnalysisToken); - CompletableFuture 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); diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java index 0f2126e..da6b91c 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java +++ b/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 messageList = tmOdsVdqwMessagearchivingMapper.queryOdsVdqwMessageByData(statTime, endTime, offset, pageSize, retry); - // 处理查询到的数据 - // 使用 CompletableFuture 并行处理 - CompletableFuture[] futures = messageList.stream() - .map(item -> CompletableFuture.runAsync(() -> { - try { - processItem(item, finalStatTime, finalEndTime); - } catch (Exception e) { - log.error("处理企微语料失败: FromUserId={}, AcceptUserId={}, 异常: {}", - item.getFromUserId(), item.getAcceptUserId(), e.getMessage(), e); - } - }, executor)) - .toArray(CompletableFuture[]::new); - // 等待所有任务完成 - CompletableFuture.allOf(futures).join(); - } - // 关闭线程池 - executor.shutdown(); - log.info("企微数据跑批结束 耗时：{}",System.currentTimeMillis()-startTime); + try { + for (int i = 1; i <= totalPages; i++) { + int offset = (i - 1) * pageSize; + List messageList = tmOdsVdqwMessagearchivingMapper.queryOdsVdqwMessageByData(statTime, endTime, offset, pageSize, retry); + // 处理查询到的数据 + // 使用 CompletableFuture 并行处理 + CompletableFuture[] futures = messageList.stream() + .map(item -> CompletableFuture.runAsync(() -> { + try { + processItem(item, finalStatTime, finalEndTime); + } catch (Exception e) { + log.error("处理企微语料失败: FromUserId={}, AcceptUserId={}, 异常: {}", + item.getFromUserId(), item.getAcceptUserId(), e.getMessage(), e); + } + }, executor)) + .toArray(CompletableFuture[]::new); + // 等待所有任务完成 + CompletableFuture.allOf(futures).join(); + } + // 关闭线程池 + 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) {