代码优化

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

@@ -0,0 +1,17 @@
+package com.volvo.ai.analytic.center.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+@Configuration
+public class ExecutorConfig {
+
+    @Bean("corpusProcessExecutor")
+    public ExecutorService corpusProcessExecutor() {
+        int poolSize = Runtime.getRuntime().availableProcessors() + 2;
+        return Executors.newFixedThreadPool(poolSize);
+    }
+}

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/CorpusFailJob.java

@@ -140,6 +140,7 @@ public class CorpusFailJob {
                                 aiAnalysisErrors.setRetryCount(aiAnalysisErrors.getRetryCount() + 1);
                                 updateAiAnalysisErrors(aiAnalysisErrors, oldAiAnalysisRequestLogs.getAiAnalysisRequestId());
                             }else {
+                                log.info("corpusFailTask Dcc失败重试:{}, {}", corpusReportDTO.getRecordId(),  oldAiAnalysisRequestLogs.getAiAnalysisRequestId());
                                 List<AicorpusTelephoneDTO>  dccDtoList =  tmTelephoneCorpusMapper.queryTelephoneCorpusBySourceIds( Arrays.asList(corpusReportDTO.getRecordId()));
                                 if(CollectionUtils.isNotEmpty(dccDtoList)){
                                     AicorpusTelephoneDTO dccDto = dccDtoList.get(0);

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

@@ -1,6 +1,7 @@
 
 package com.volvo.ai.analytic.center.mq;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.volvo.ai.analytic.center.constant.Constant;
 import com.volvo.ai.analytic.center.dto.corpus.AicorpusTelephoneDTO;
@@ -24,9 +25,9 @@ import org.springframework.web.bind.annotation.RestController;
 
 import javax.annotation.Resource;
 import java.time.LocalDateTime;
+import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.*;
-import java.util.concurrent.Executors;
 
 /**
  * @ClassName CorpusProcessKafkaConsumer
@@ -53,72 +54,100 @@ public class CorpusProcessKafkaProducer {
 	@Resource
 	private RocketMQTemplate rocketMqTemplate;
 
+	@Resource(name = "corpusProcessExecutor")
+	private ExecutorService executor;
+
 	@KafkaListener(topics = "${spring.kafka.topic}", groupId = "${spring.kafka.group}")
 	public void listen(List<ConsumerRecord<String, Object>> recordMessages) {
 		long startTime = System.currentTimeMillis();
-		log.info("CorpusProcessKafkaProducer Received message: {}", recordMessages);
+		if (CollectionUtils.isEmpty(recordMessages)) {
-		// 获取消息列表
+			log.debug("Received empty message batch.");
-		int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 2;
+			return;
-		log.info("获取的线程数：{}", optimalThreadPoolSize);
+		}
-		ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize);
+
+		List<CompletableFuture<Void>> futures = new ArrayList<>();
 		try {
-			if(CollectionUtils.isNotEmpty(recordMessages)){
+			log.info("Received {} messages. Start processing...", recordMessages.size());
-				log.info("CorpusProcessKafkaProducer List size: {}", recordMessages.size());
-				for (ConsumerRecord<String, Object> record : recordMessages) {
-					log.info("CorpusProcessKafkaProducer message: {}",record);
-					executor.submit(() -> {
-						try {
-							String message = (String) record.value();
-							AicorpusTelephoneDTO aicorpusTelephone = objectMapper.readValue(message, AicorpusTelephoneDTO.class);
-							log.info("aicorpusTelephone categoryCode:{}, display: {}", aicorpusTelephone.getCategoryCode(), aicorpusTelephone.getDisplay());
-							DisplayDTO display = objectMapper.readValue(aicorpusTelephone.getDisplay(), DisplayDTO.class);
-							log.info("aicorpusTelephone display getSegments: {}", display.getSegments());
-							TmTelephoneCorpus tmTelephoneCorpus = new TmTelephoneCorpus();
-							BeanUtils.copyProperties(aicorpusTelephone, tmTelephoneCorpus);
-							tmTelephoneCorpus.setCreateBy("kafka");
-							tmTelephoneCorpus.setCreateTime(LocalDateTime.now());
-							// 条件：只处理dcc的 10s通话时间以上
-							log.info("CorpusProcessKafkaProducer getCategoryCode: {}，audioFileId：{}，sourceId：{}", tmTelephoneCorpus.getCategoryCode(), aicorpusTelephone.getAudioFileId(), aicorpusTelephone.getSourceId());
-							if (Constant.CHANNEL_DCC.equals(tmTelephoneCorpus.getCategoryCode())) {
-								log.info(" dcc 语料开始处理");
-								tmTelephoneCorpusService.saveTelephoneCorpus(tmTelephoneCorpus);
-								long startTimeDify = System.currentTimeMillis();
 
-								rocketMqTemplate.asyncSend(dccMqTipic, MessageBuilder.withPayload(message).build(),
+			for (ConsumerRecord<String, Object> record : recordMessages) {
-										new SendCallback() {
+				// 使用 CompletableFuture 提交异步任务
-											@Override
+				futures.add(
-											public void onSuccess(SendResult sendResult) {
+						CompletableFuture.runAsync(() -> processSingleRecord(record), executor)
-												log.info("dcc发送MQ成功  消息体:{}", message);
+				);
-											}
-											@Override
-											public void onException(Throwable e) {
-												log.error("dcc 送MQ异常  消息体:{}, 异常：", message, e);
-											}
-										}, 10000);
-
-								log.info(" dify处理耗时：{}", System.currentTimeMillis() - startTimeDify);
-							}
-						} catch (Exception e) {
-							log.error("CorpusProcessKafkaProducer 电话语料 解析JSON出错: {}", e.getMessage());
-						}finally {
-							executor.shutdown();
-						}
-					});
-
-				}
-				executor.shutdown();
 			}
 
-			log.info("Kafka 消息处理完成，耗时：{}", System.currentTimeMillis() - startTime);
-			// 在这里可以添加对解析后的对象的进一步处理逻辑
 		} catch (Exception e) {
-			log.error("CorpusProcessKafkaProducer 电话语料 解析JSON出错: {}" , e.getMessage());
+			log.error("Failed to process message batch: {}", e.getMessage(), e);
-		}finally {
+		} finally {
+			// 2. 安全关闭线程池
 			executor.shutdown();
+			try {
+				if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
+					log.warn("Thread pool did not terminate in time. Forcing shutdown.");
+					executor.shutdownNow();
+				}
+			} catch (InterruptedException e) {
+				log.error("Thread pool termination interrupted: ", e);
+				Thread.currentThread().interrupt();
+			}
+			log.info("Total processing time: {} ms", System.currentTimeMillis() - startTime);
 		}
 	}
 
+	/**
+	 * 单条消息处理逻辑（解耦核心逻辑）
+	 */
+	private void processSingleRecord(ConsumerRecord<String, Object> record) {
+		try {
+			log.debug("Processing message: topic={}, partition={}, offset={}",
+					record.topic(), record.partition(), record.offset());
 
+			String message = (String) record.value();
+			AicorpusTelephoneDTO aicorpusTelephone = objectMapper.readValue(message, AicorpusTelephoneDTO.class);
+			DisplayDTO display = objectMapper.readValue(aicorpusTelephone.getDisplay(), DisplayDTO.class);
+
+			TmTelephoneCorpus tmTelephoneCorpus = new TmTelephoneCorpus();
+			BeanUtils.copyProperties(aicorpusTelephone, tmTelephoneCorpus);
+			tmTelephoneCorpus.setCreateBy("kafka");
+			tmTelephoneCorpus.setCreateTime(LocalDateTime.now());
+
+			// 3. 逻辑拆分：处理 DCC 语料
+			if (Constant.CHANNEL_DCC.equals(tmTelephoneCorpus.getCategoryCode())) {
+				handleDccCorpus(tmTelephoneCorpus, message);
+			}
+		} catch (JsonProcessingException e) {
+			log.error("JSON parsing failed for message: {}", record.value(), e);
+		} catch (Exception e) {
+			log.error("Unexpected error processing message: ", e);
+		}
+	}
+
+	/**
+	 * 处理 DCC 语料逻辑（异步发送 MQ）
+	 */
+	private void handleDccCorpus(TmTelephoneCorpus corpus, String rawMessage) {
+		log.info("handleDccCorpus: categoryCode={}, sourceId={}",
+				corpus.getCategoryCode(), corpus.getSourceId());
+
+		// 4. 保存语料
+		tmTelephoneCorpusService.saveTelephoneCorpus(corpus);
+
+		// 5. 异步发送 RocketMQ（带超时和异常回调）
+		long startTimeDify = System.currentTimeMillis();
+		rocketMqTemplate.asyncSend(dccMqTipic, MessageBuilder.withPayload(rawMessage).build(),
+				new SendCallback() {
+					@Override
+					public void onSuccess(SendResult sendResult) {
+						log.debug("DCC MQ sent successfully发送成功: {}", rawMessage);
+					}
+					@Override
+					public void onException(Throwable e) {
+						log.error("DCC MQ send failed发送失败: {}", rawMessage, e);
+					}
+				}, 10000);  // 超时时间配置化（建议提取到 @Value）
+
+		log.info("handleDccCorpus: {} ms", System.currentTimeMillis() - startTimeDify);
+	}
 
 }