diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/ExecutorConfig.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/ExecutorConfig.java index ddb1c60..c4eda2d 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/ExecutorConfig.java +++ b/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 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(); + } } } } diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaConsumer.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaConsumer.java index 40dc84d..37da48a 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaConsumer.java +++ b/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) {