修改dcckafka处理

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

@@ -1,12 +1,16 @@
 package com.volvo.ai.analytic.center.config;
 
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
+import javax.annotation.PreDestroy;
 import java.util.concurrent.ThreadPoolExecutor;
 
+@Slf4j
 @Configuration
 public class ExecutorConfig {
 
@@ -22,7 +26,9 @@ public class ExecutorConfig {
     @Value("${task.pool.queueCapacity}")
     private int queueCapacity;
 
+    private ThreadPoolTaskExecutor executor;
     @Bean("threadPoolTaskExecutor")
+    @Primary
     public ThreadPoolTaskExecutor corpusProcessExecutor() {
         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
         executor.setCorePoolSize(corePoolSize);
@@ -32,7 +38,16 @@ public class ExecutorConfig {
         executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
         executor.setKeepAliveSeconds(keepAliveSeconds);
         executor.initialize();
-
+        this.executor = executor;
         return executor;
     }
+
+    @PreDestroy
+    public void destroy() {
+        if (executor != null) {
+            log.info("Shutting down thread pool: {}", executor);
+            ThreadPoolExecutor threadPoolExecutor = this.executor.getThreadPoolExecutor();
+            threadPoolExecutor.shutdownNow(); // 强制关闭所有正在执行的任务
+        }
+    }
 }

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/TestController.java

@@ -3,7 +3,9 @@ package com.volvo.ai.analytic.center.controller;
 
 import com.alibaba.fastjson.JSONObject;
 import com.volvo.ai.analytic.center.dto.corpus.AicorpusTelephoneDTO;
+import com.volvo.ai.analytic.center.entity.TmTelephoneCorpus;
 import com.volvo.ai.analytic.center.service.AiAnalysisRequestLogsService;
+import com.volvo.ai.analytic.center.service.TmTelephoneCorpusService;
 import com.volvo.common.core.util.ResultMsg;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
@@ -82,5 +84,13 @@ public class TestController {
         return dataSource.getClass().getName();
     }
 
+    @Autowired
+    public TmTelephoneCorpusService tmTelephoneCorpusService;
+
+    @PostMapping("/save")
+    public void save(@RequestBody TmTelephoneCorpus tmTelephoneCorpus) {
+          tmTelephoneCorpusService.saveTelephoneCorpus(tmTelephoneCorpus);
+    }
+
 }
 

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

@@ -29,8 +29,6 @@ import javax.annotation.Resource;
 import java.time.LocalDateTime;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
 
 /**
  * @ClassName CorpusProcessKafkaConsumer
@@ -61,8 +59,9 @@ public class CorpusProcessKafkaProducer {
 	@Resource(name = "threadPoolTaskExecutor")
 	private ThreadPoolTaskExecutor executor;
 
+
 	@KafkaListener(topics = "${spring.kafka.topic}", groupId = "${spring.kafka.group}")
-	public void listen( @Payload List<String> messageList,
+	public void listen(@Payload  List<Object> messageList,
 						@Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
 						@Header(KafkaHeaders.OFFSET) List<Long> offsets
 	)  {
@@ -74,7 +73,7 @@ public class CorpusProcessKafkaProducer {
 		}
 		try {
 			for (int i = 0; i < messageList.size(); i++) {
-				String message = messageList.get(i);
+				String message =String.valueOf( messageList.get(i));
 				log.debug("corpusProcessKafkaProducer message:{}", message);
 				int partition = partitions.get(i);
 				long offset = offsets.get(i);
@@ -82,26 +81,21 @@ public class CorpusProcessKafkaProducer {
 				log.debug("corpusProcessKafkaProducer message: topic={}, partition={}, offset={}", "your-topic",partition, offset);
 
 				// 提交异步任务
-				CompletableFuture.runAsync(() -> processSingleRecord(message), executor);
+				CompletableFuture.runAsync(() -> {
+					try {
+						processSingleRecord(message);
+					} catch (Exception e) {
+						log.error("corpusProcessKafkaProducer异步任务执行失败", e);
+					}
+				}, executor);
 			}
 
 		} catch (Exception e) {
-			log.error("Failed to process message batch: {}", e.getMessage(), e);
+			log.error("corpusProcessKafkaProducer Failed to process message batch: {}", e.getMessage(), e);
 		} finally {
-			// 2. 安全关闭线程池
+
-			executor.shutdown();
+			log.info("corpusProcessKafkaProducer Total processing time: {} ms", System.currentTimeMillis() - startTime);
-			try {
+ 		}
-				ThreadPoolExecutor threadPoolExecutor = this.executor.getThreadPoolExecutor();
-				if (!threadPoolExecutor.awaitTermination(60, TimeUnit.SECONDS)) {
-					log.warn("Thread pool did not terminate in time. Forcing shutdown.");
-					threadPoolExecutor.shutdownNow();
-				}
-			} catch (InterruptedException e) {
-				log.error("Thread pool termination interrupted: ", e);
-				Thread.currentThread().interrupt();
-			}
-			log.info("Total processing time: {} ms", System.currentTimeMillis() - startTime);
-		}
 	}
 
 	/**
@@ -109,8 +103,8 @@ public class CorpusProcessKafkaProducer {
 	 */
 	private void processSingleRecord(String record) {
 		try {
-			log.info("processSingleRecord message:{}",  record);
+			log.info("corpusProcessKafkaProducer processSingleRecord :{}",  record);
-			AicorpusTelephoneDTO aicorpusTelephone = objectMapper.readValue(record, AicorpusTelephoneDTO.class);
+			AicorpusTelephoneDTO aicorpusTelephone = objectMapper. readValue(record, AicorpusTelephoneDTO.class);
 			TmTelephoneCorpus tmTelephoneCorpus = new TmTelephoneCorpus();
 			BeanUtils.copyProperties(aicorpusTelephone, tmTelephoneCorpus);
 			tmTelephoneCorpus.setCreateBy("kafka");