添加线程池

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

@@ -0,0 +1,68 @@
+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/service/impl/MqMessageRecordServiceImpl.java

@@ -38,6 +38,7 @@ 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
@@ -83,6 +84,9 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
     @Value("${rocketmq.producer.topic}")
     private String topic;
 
+    @Autowired
+    private Executor getAsyncExecutor;
+
     /**
      * 处理Mq消息
      *
@@ -147,7 +151,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
 
     private void processDify(DiFyReq diFyReq, JSONArray difyResult, String aiAnalysisRequestId) throws InterruptedException, ExecutionException {
         //舆情案件分析
-        CompletableFuture<JSONObject> caseWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(diFyReq, caseToken))
+        CompletableFuture<JSONObject> caseWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(diFyReq, caseToken),getAsyncExecutor)
                 .exceptionally(ex -> {
                     log.error("舆情案件分析失败: {}", ex.getMessage());
                     JSONObject errorResult = new JSONObject();
@@ -156,7 +160,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
                 });
 
         // 内容主题关键词打标
-        CompletableFuture<JSONObject> keywordWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(diFyReq, keywordToken))
+        CompletableFuture<JSONObject> keywordWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(diFyReq, keywordToken),getAsyncExecutor)
                 .exceptionally(ex -> {
                     log.error("内容主题关键词打标失败: {}", ex.getMessage());
                     JSONObject errorResult = new JSONObject();
@@ -165,7 +169,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
                 });
 
         // litecrm线索分析
-        CompletableFuture<JSONObject> clueAnalysisWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(diFyReq, clueAnalysisToken))
+        CompletableFuture<JSONObject> clueAnalysisWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(diFyReq, clueAnalysisToken),getAsyncExecutor)
                 .exceptionally(ex -> {
                     log.error("litecrm线索分析失败: {}", ex.getMessage());
                     JSONObject errorResult = new JSONObject();