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 new file mode 100644 index 0000000..047dadd --- /dev/null +++ b/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); + } +} diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/CorpusFailJob.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/CorpusFailJob.java index 70f1864..8608991 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/CorpusFailJob.java +++ b/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 dccDtoList = tmTelephoneCorpusMapper.queryTelephoneCorpusBySourceIds( Arrays.asList(corpusReportDTO.getRecordId())); if(CollectionUtils.isNotEmpty(dccDtoList)){ AicorpusTelephoneDTO dccDto = dccDtoList.get(0); diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java index 047ce3e..7bafbea 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java +++ b/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.Executors; +import java.util.concurrent.*; /** * @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> recordMessages) { long startTime = System.currentTimeMillis(); - log.info("CorpusProcessKafkaProducer Received message: {}", recordMessages); - // 获取消息列表 - int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 2; - log.info("获取的线程数：{}", optimalThreadPoolSize); - ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize); + if (CollectionUtils.isEmpty(recordMessages)) { + log.debug("Received empty message batch."); + return; + } + + List> futures = new ArrayList<>(); try { - if(CollectionUtils.isNotEmpty(recordMessages)){ - log.info("CorpusProcessKafkaProducer List size: {}", recordMessages.size()); - for (ConsumerRecord 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(); + log.info("Received {} messages. Start processing...", recordMessages.size()); - rocketMqTemplate.asyncSend(dccMqTipic, MessageBuilder.withPayload(message).build(), - new SendCallback() { - @Override - public void onSuccess(SendResult sendResult) { - 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(); + for (ConsumerRecord record : recordMessages) { + // 使用 CompletableFuture 提交异步任务 + futures.add( + CompletableFuture.runAsync(() -> processSingleRecord(record), executor) + ); } - log.info("Kafka 消息处理完成，耗时：{}", System.currentTimeMillis() - startTime); - // 在这里可以添加对解析后的对象的进一步处理逻辑 } catch (Exception e) { - log.error("CorpusProcessKafkaProducer 电话语料 解析JSON出错: {}" , e.getMessage()); - }finally { + log.error("Failed to process message batch: {}", e.getMessage(), e); + } 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 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); + } }