diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java index 01beff9..0624b4b 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java @@ -14,6 +14,8 @@ import org.springframework.kafka.listener.ContainerProperties; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; @Configuration @RefreshScope @@ -68,7 +70,7 @@ public class KafkaConfig { props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 45000); // 会话1分钟 props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 600000); // 可选：允许更长的消费间隔 props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 15000); - /** props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 20); // 单次最多拉取的消息数 **/ + props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1); // 单次最多拉取的消息数 **/ ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props)); @@ -77,4 +79,10 @@ public class KafkaConfig { return factory; } + @Bean("kafkaTaskExecutor") + public ExecutorService kafkaTaskExecutor() { + int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 1; + return Executors.newFixedThreadPool(optimalThreadPoolSize); + } + } \ No newline at end of file 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 0eccb7c..ff37246 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 @@ -100,7 +100,7 @@ public class CorpusFailJob { queryWrapper.eq(AiAnalysisRequestLogs::getAiAnalysisRequestId, aiAnalysisErrors.getAiAnalysisRequestId()); AiAnalysisRequestLogs oldAiAnalysisRequestLogs = aiAnalysisRequestLogsMapper.selectOne(queryWrapper); - if (null != oldAiAnalysisRequestLogs && StringUtils.isNotBlank(oldAiAnalysisRequestLogs.getBusinessResponse())) { + if (null != oldAiAnalysisRequestLogs && StringUtils.isBlank(oldAiAnalysisRequestLogs.getBusinessResponse())) { DiFyReq diFyReq = JSONObject.parseObject(oldAiAnalysisRequestLogs.getDifyRequest(), DiFyReq.class); CorpusReportDTO corpusReportDTO = JSONObject.parseObject(oldAiAnalysisRequestLogs.getBusinessRequest(), CorpusReportDTO.class); Map ltoMap = new HashMap<>(); @@ -141,7 +141,7 @@ public class CorpusFailJob { } } else if(null != corpusReportDTO && corpusReportDTO.getAnalysisScene() == 3){ - log.info(" corpusFailTask 铭牌数据重试：{}", corpusReportDTO.getAnalysisScene()); + log.info(" corpusFailTask 铭牌数据重试：{},aiId:{}", corpusReportDTO.getCustomerFlowId(), oldAiAnalysisRequestLogs.getAiAnalysisRequestId()); JSONObject execDifyFlow = diFyService.executeDifyFlow(diFyReq, oldAiAnalysisRequestLogs.getAiAnalysisRequestType(), JSONObject.toJSONString(corpusReportDTO),oldAiAnalysisRequestLogs.getAiAnalysisRequestId()); List nameplateCorpusList = tmNameplateCorpusService.queryTelephoneCorpusByCustomerFlowId( Arrays.asList(corpusReportDTO.getCustomerFlowId())); if(CollectionUtils.isNotEmpty(nameplateCorpusList)) { diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/ConsumerStateManager.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/ConsumerStateManager.java new file mode 100644 index 0000000..d4e31d6 --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/ConsumerStateManager.java @@ -0,0 +1,54 @@ +package com.volvo.ai.analytic.center.mq; + +import org.apache.kafka.clients.consumer.Consumer; +import org.springframework.stereotype.Component; + +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +@Component +public class ConsumerStateManager { + // 使用 ThreadLocal 存储每个线程的 Consumer 实例 + private final ThreadLocal> consumerThreadLocal = new ThreadLocal<>(); + private final AtomicInteger messageCount = new AtomicInteger(0); + private volatile boolean paused = false; + private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(); + + // 绑定当前线程的 Consumer + public void bindConsumer(Consumer consumer) { + consumerThreadLocal.set(consumer); + } + + // 获取当前线程绑定的 Consumer + public Consumer getKafkaConsumer() { + return consumerThreadLocal.get(); + } + + // 更新计数器并检查是否需要暂停 + public void checkAndPause() { + if (messageCount.incrementAndGet() >= 30 && !paused) { + paused = true; + Consumer consumer = getKafkaConsumer(); + if (consumer != null) { + consumer.pause(consumer.assignment()); + scheduler.schedule(this::resume, 2, TimeUnit.MINUTES); + } + } + } + + // 恢复消费 + private void resume() { + paused = false; + messageCount.set(0); + Consumer consumer = getKafkaConsumer(); + if (consumer != null) { + consumer.resume(consumer.assignment()); + } + } + + public boolean isPaused() { + return paused; + } +} \ No newline at end of file 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 9f50859..ebc5aa0 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 @@ -6,11 +6,14 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.volvo.ai.analytic.center.dto.corpus.NameplateTableKafkaDTO; import com.volvo.ai.analytic.center.service.TmNameplateCorpusService; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.collections.CollectionUtils; import org.apache.commons.lang3.StringUtils; +import org.apache.kafka.clients.consumer.Consumer; import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Value; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.cloud.context.config.annotation.RefreshScope; +import org.springframework.context.annotation.Bean; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.KafkaHeaders; @@ -19,9 +22,8 @@ import org.springframework.stereotype.Component; import org.springframework.web.bind.annotation.RestController; import javax.annotation.Resource; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; /** * @ClassName NameplateKafkaConsumer @@ -42,56 +44,84 @@ public class NameplateKafkaConsumer { private final ObjectMapper objectMapper = new ObjectMapper(); - @Value("${rocketmq.producer.corpus.dcctopic}") - private String dccMqTipic; +// @Value("${analyticCenterKafka.consumer.threshold}") +// private String threshold; @Resource private RocketMQTemplate rocketMqTemplate; - private int messageCount = 0; - @KafkaListener(topics = "${analyticCenterKafka.consumer.topic}", + private final AtomicInteger messageCount = new AtomicInteger(0); + // 批次阈值（例如每30条提交一次offset） + private static final int THRESHOLD = 20; + + // 最大等待时间（2分钟） + private static final long MAX_WAIT_TIME = TimeUnit.MINUTES.toMillis(2); + @Autowired + @Qualifier("kafkaTaskExecutor") + public ExecutorService kafkaTaskExecutor; + + @Autowired + public ConsumerStateManager consumerStateManager; + @KafkaListener(topics = "${analyticCenterKafka.consumer.topic}", groupId = "${analyticCenterKafka.consumer.group}" , containerFactory = "analyticCenterConsumerFactory", concurrency = "3") public void listen(String recordMessages, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partitionId, - @Header(KafkaHeaders.OFFSET) Long offset) { + @Header(KafkaHeaders.OFFSET) Long offset, + Consumer consumer ) throws InterruptedException { long startTime = System.currentTimeMillis(); - - messageCount++; - log.info("nameplateKafkaConsumer 当前线程: {}, 线程ID: {},计数：{}", Thread.currentThread().getName(), Thread.currentThread().getId(),messageCount); + log.info("nameplateKafkaConsumer 当前线程: {}, 线程ID: {},计数：{}", Thread.currentThread().getName(), Thread.currentThread().getId()); log.info("nameplateKafkaConsumerMessage: {}", recordMessages); - if(StringUtils.isNotEmpty(recordMessages)){ + // 初始化绑定 Consumer - int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 1; - log.info("获取的线程数：{}",optimalThreadPoolSize); - // 创建线程池 - ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize); // 根据需求调整线程池大小 - NameplateTableKafkaDTO tmNameplateCorpus = JSON.parseObject(recordMessages, NameplateTableKafkaDTO.class); - log.info("nameplateKafkaConsumerParseType: {},size:{}", tmNameplateCorpus.getType(), tmNameplateCorpus.getData().size()); - if(tmNameplateCorpus.getType().equals("INSERT")){ - tmNameplateCorpus.getData().forEach(nameplate -> - CompletableFuture.runAsync(() -> { - try { - log.info("nameplateKafkaConsumerCustomerFlowId: {}",nameplate.getCustomerFlowId()); - tmNameplateCorpusService.processItem(nameplate); - } catch (Exception e) { - log.error("nameplateKafkaConsumer铭牌解析处理出错 {}:{}，ex:{}", - partitionId, offset, e.getMessage()); - } - }, executor)); - try { - log.info("nameplateKafkaConsumerack.acknowledge:{}"); - ack.acknowledge(); // 确保无论成功与否都提交 offset - } catch (IllegalStateException e) { - log.warn("Offset 已提交，跳过重复提交"); - } - } - }else { - ack.acknowledge(); // 空消息直接跳过 + consumerStateManager.bindConsumer(consumer); + + // 检查是否处于暂停状态 + if (consumerStateManager.isPaused()) { + log.info("当前处于暂停状态，忽略消息: partition={}, offset={}", partitionId, offset); + return; // 不提交偏移量，等待恢复后重新拉取 } + if (StringUtils.isEmpty(recordMessages)) { + ack.acknowledge(); + return; + } + + NameplateTableKafkaDTO tmNameplateCorpus = JSON.parseObject(recordMessages, NameplateTableKafkaDTO.class); + if (!"INSERT".equals(tmNameplateCorpus.getType()) || CollectionUtils.isEmpty(tmNameplateCorpus.getData())) { + ack.acknowledge(); + return; + } + + messageCount.addAndGet(1); + if (messageCount.get() >= THRESHOLD) { + log.info("nameplateKafkaConsumer 消费数量：{}", messageCount.get()); + Thread.sleep(MAX_WAIT_TIME); // 让线程休眠指定的时间 + messageCount.set(0); // 重置计数器 + } + + log.info("nameplateKafkaConsumerParseType: {},size:{}", tmNameplateCorpus.getType(), tmNameplateCorpus.getData().size()); + tmNameplateCorpus.getData().forEach(nameplate -> + CompletableFuture.runAsync(() -> { + try { + log.info("nameplateKafkaConsumerCustomerFlowId: {}",nameplate.getCustomerFlowId()); + tmNameplateCorpusService.processItem(nameplate); + } catch (Exception e) { + log.error("nameplateKafkaConsumer铭牌解析处理出错 {}:{}，ex:{}", + partitionId, offset, e.getMessage()); + } + }, kafkaTaskExecutor)); + + try { + ack.acknowledge(); // 确保无论成功与否都提交 offset + consumerStateManager.checkAndPause(); // 更新计数器并检查暂停 + } catch (Exception e) { + log.error("任务执行异常: {}", e.getMessage()); + Thread.currentThread().interrupt(); + } log.info("nameplateKafkaConsumer消息处理完成，耗时：{}", System.currentTimeMillis() - startTime); + } }