From abc35c0b68a5516f1bd8de70f3ad6ddd4b32ae17 Mon Sep 17 00:00:00 2001 From: zren25 Date: Fri, 16 May 2025 17:49:22 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9kafka=E9=AA=8C=E8=AF=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../center/mq/KafkaMessageScheduler.java | 38 +++++++++++++ .../center/mq/NameplateKafkaConsumer.java | 56 +++++-------------- 2 files changed, 51 insertions(+), 43 deletions(-) create mode 100644 ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/KafkaMessageScheduler.java diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/KafkaMessageScheduler.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/KafkaMessageScheduler.java new file mode 100644 index 0000000..774e47a --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/KafkaMessageScheduler.java @@ -0,0 +1,38 @@ +package com.volvo.ai.analytic.center.mq; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + +import javax.annotation.PostConstruct; +import java.util.concurrent.*; +@Slf4j +@Component +public class KafkaMessageScheduler { + + private final BlockingQueue messageQueue = new LinkedBlockingQueue<>(); + + @PostConstruct + public void startScheduler() { + ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); + scheduler.scheduleAtFixedRate(() -> { + try { + Runnable task = messageQueue.poll(1, TimeUnit.SECONDS); + if (task != null) { + task.run(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }, 0, 4, TimeUnit.SECONDS); // 每隔 4 秒执行一次 + } + + public boolean submit(Runnable task) { + if (messageQueue.remainingCapacity() == 0) { + log.warn("Kafka message queue is full, task rejected."); + return false; + } + log.info("messageQueueSize: {}",messageQueue.size()); + messageQueue.offer(task); + return true; + } +} 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 ebc5aa0..2c78ae7 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 @@ -43,45 +43,27 @@ public class NameplateKafkaConsumer { private TmNameplateCorpusService tmNameplateCorpusService; private final ObjectMapper objectMapper = new ObjectMapper(); - -// @Value("${analyticCenterKafka.consumer.threshold}") -// private String threshold; - @Resource private RocketMQTemplate rocketMqTemplate; - 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; + public KafkaMessageScheduler kafkaMessageScheduler; @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, - Consumer consumer ) throws InterruptedException { + @Header(KafkaHeaders.OFFSET) Long offset) throws InterruptedException { long startTime = System.currentTimeMillis(); log.info("nameplateKafkaConsumer 当前线程: {}, 线程ID: {},计数：{}", Thread.currentThread().getName(), Thread.currentThread().getId()); log.info("nameplateKafkaConsumerMessage: {}", recordMessages); // 初始化绑定 Consumer - consumerStateManager.bindConsumer(consumer); - - // 检查是否处于暂停状态 - if (consumerStateManager.isPaused()) { - log.info("当前处于暂停状态，忽略消息: partition={}, offset={}", partitionId, offset); - return; // 不提交偏移量，等待恢复后重新拉取 - } if (StringUtils.isEmpty(recordMessages)) { ack.acknowledge(); @@ -94,31 +76,19 @@ public class NameplateKafkaConsumer { 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)); + boolean isSubmit = kafkaMessageScheduler.submit(() -> { + try { + // 真正的业务逻辑在这里执行 + tmNameplateCorpus.getData().forEach(nameplate -> tmNameplateCorpusService.processItem(nameplate)); - try { - ack.acknowledge(); // 确保无论成功与否都提交 offset - consumerStateManager.checkAndPause(); // 更新计数器并检查暂停 - } catch (Exception e) { - log.error("任务执行异常: {}", e.getMessage()); - Thread.currentThread().interrupt(); + } catch (Exception e) { + log.error("消息处理失败", e); + } + }); + if (isSubmit){ + log.info("nameplateKafkaConsumeracknowledge：{}"); + ack.acknowledge(); // 手动提交 offset } log.info("nameplateKafkaConsumer消息处理完成，耗时：{}", System.currentTimeMillis() - startTime);