修改企微画像的线程池处理

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

@@ -8,7 +8,10 @@ import org.springframework.context.annotation.Primary;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 import javax.annotation.PreDestroy;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 @Slf4j
 @Configuration
@@ -27,26 +30,100 @@ public class ExecutorConfig {
     private int queueCapacity;
 
     private ThreadPoolTaskExecutor executor;
+    private ThreadPoolExecutor blockingExecutor;
+
     @Bean("threadPoolTaskExecutor")
     @Primary
     public ThreadPoolTaskExecutor corpusProcessExecutor() {
         executor = new ThreadPoolTaskExecutor();
+
+        log.info("Thread-process-pool-Initializing: corePoolSize: {}, maxPoolSize: {}, queueCapacity: {}, keepAliveSeconds: {}",corePoolSize, maxPoolSize, queueCapacity, keepAliveSeconds);
+        // 设置核心线程数
         executor.setCorePoolSize(corePoolSize);
+        // 设置最大线程数为5
         executor.setMaxPoolSize(maxPoolSize);
+        // 设置队列容量为10
         executor.setQueueCapacity(queueCapacity);
-        executor.setThreadNamePrefix("Thread-process-pool-");
+        executor.setThreadNamePrefix("Thread-pool-lizh");
+        // 使用阻塞策略：当队列满时，新任务会阻塞等待
         executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
         executor.setKeepAliveSeconds(keepAliveSeconds);
+        // 允许核心线程超时，提高资源利用率
+        executor.setAllowCoreThreadTimeOut(true);
         executor.initialize();
         return executor;
     }
 
+    /**
+     * 使用Java自带的ThreadPoolExecutor实现真正的阻塞功能
+     * 功能：
+     * 1. 最大线程数是50个
+     * 2. 超过数量时排队等待，排队队列里最多100个对象，超过数量时阻塞
+     */
+    @Bean("blockingThreadPoolExecutor")
+    public ThreadPoolExecutor blockingThreadPoolExecutor() {
+        // 创建容量为100的阻塞队列
+        BlockingQueue<Runnable> blockingQueue = new LinkedBlockingQueue<>(100);
+        
+        // 创建线程池执行器
+        blockingExecutor = new ThreadPoolExecutor(
+                10,                    // 核心线程数
+                50,                    // 最大线程数
+                60L,                   // 线程空闲时间
+                TimeUnit.SECONDS,      // 时间单位
+                blockingQueue,         // 阻塞队列，容量100
+                r -> {                 // 线程工厂
+                    Thread t = new Thread(r, "Blocking-Thread-Pool-" + System.currentTimeMillis());
+                    t.setDaemon(false);
+                    return t;
+                },
+                new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略：调用者运行
+        );
+        
+        // 允许核心线程超时
+        blockingExecutor.allowCoreThreadTimeOut(true);
+        
+        log.info("Java自带阻塞式线程池初始化完成 - 核心线程数: 10, 最大线程数: 50, 队列容量: 100");
+        return blockingExecutor;
+    }
+
     @PreDestroy
     public void destroy() {
+        // 关闭ThreadPoolTaskExecutor
         if (executor != null) {
             log.info("Thread-process-pool-ShuttingDown: {}", executor);
             ThreadPoolExecutor threadPoolExecutor = this.executor.getThreadPoolExecutor();
-            threadPoolExecutor.shutdownNow(); // 强制关闭所有正在执行的任务
+            threadPoolExecutor.shutdown();
+            
+            try {
+                // 等待所有任务完成，最多等待30秒
+                if (!threadPoolExecutor.awaitTermination(30, TimeUnit.SECONDS)) {
+                    log.warn("线程池未能在30秒内正常关闭，强制关闭");
+                    threadPoolExecutor.shutdownNow();
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                log.error("等待线程池关闭时被中断", e);
+                threadPoolExecutor.shutdownNow();
+            }
+        }
+        
+        // 关闭Java自带的阻塞式线程池
+        if (blockingExecutor != null) {
+            log.info("Java自带阻塞式线程池正在关闭: {}", blockingExecutor);
+            blockingExecutor.shutdown();
+            
+            try {
+                // 等待所有任务完成，最多等待30秒
+                if (!blockingExecutor.awaitTermination(30, TimeUnit.SECONDS)) {
+                    log.warn("Java自带线程池未能在30秒内正常关闭，强制关闭");
+                    blockingExecutor.shutdownNow();
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                log.error("等待Java自带线程池关闭时被中断", e);
+                blockingExecutor.shutdownNow();
+            }
         }
     }
 }

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

@@ -2,6 +2,7 @@
 package com.volvo.ai.analytic.center.mq;
 
 import com.alibaba.fastjson.JSON;
+import com.volvo.ai.analytic.center.config.ExecutorConfig;
 import com.volvo.ai.analytic.center.dto.corpus.NameplateTableKafkaDTO;
 import com.volvo.ai.analytic.center.service.TmNameplateCorpusService;
 import lombok.extern.slf4j.Slf4j;
@@ -13,9 +14,13 @@ import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Component;
 import org.springframework.web.bind.annotation.RestController;
 
+import javax.annotation.Resource;
+import java.util.concurrent.CompletableFuture;
+
 /**
  * @ClassName NameplateKafkaConsumer
  * @Description  消费tm_nameplate_corpus表binlog的Kafka消息
@@ -33,6 +38,10 @@ public class NameplateKafkaConsumer {
 	@Autowired
 	private TmNameplateCorpusService tmNameplateCorpusService;
 
+	@Autowired
+	@Resource(name = "threadPoolTaskExecutor")
+	private ThreadPoolTaskExecutor executor;
+
 
 	@KafkaListener(topics = "${analyticCenterKafka.consumer.topic}",  // = smart_assistant_nameplate_topic
 			groupId = "${analyticCenterKafka.consumer.group}" , //smart_assistant_nameplate_topic_group
@@ -43,7 +52,7 @@ public class NameplateKafkaConsumer {
 					   @Header(KafkaHeaders.OFFSET) Long offset) {
 		long startTime = System.currentTimeMillis();
 		log.info("nameplateKafkaConsumer 当前线程: {}, 线程ID: {},计数：{}", Thread.currentThread().getName(), Thread.currentThread().getId());
-		log.info("nameplateKafkaConsumerMessage: {}", recordMessages);
+		log.info("nameplateKafkaConsumerMessage总数：{}，消息：{}", recordMessages);
 		// 初始化绑定 Consumer
 		if (StringUtils.isEmpty(recordMessages)) {
 			ack.acknowledge();
@@ -55,7 +64,15 @@ public class NameplateKafkaConsumer {
 				ack.acknowledge();
 			}
 
-			tmNameplateCorpus.getData().forEach(nameplate -> tmNameplateCorpusService.processItem(nameplate));
+			tmNameplateCorpus.getData().forEach(nameplate ->
+					CompletableFuture.runAsync(() -> {
+						try {
+							tmNameplateCorpusService.processItem(nameplate);
+						} catch (Exception e) {
+							log.error("corpusPortrait画像铭牌异步任务执行失败", e);
+						}
+					}, executor));
+
 			log.info("nameplateKafkaConsumeracknowledge：{}",partitionId, offset);
 			 // 手动提交 offset
 		} catch (Exception e) {