修改kafka验证

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<Runnable> 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;
+    }
+}

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,
+					   @Header(KafkaHeaders.OFFSET) Long offset) throws InterruptedException {
-					   Consumer<?, ?> consumer    ) 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());
+		boolean isSubmit =  kafkaMessageScheduler.submit(() -> {
-		tmNameplateCorpus.getData().forEach(nameplate ->
-				CompletableFuture.runAsync(() -> {
 			try {
-						log.info("nameplateKafkaConsumerCustomerFlowId: {}",nameplate.getCustomerFlowId());
+				// 真正的业务逻辑在这里执行
-						tmNameplateCorpusService.processItem(nameplate);
+				tmNameplateCorpus.getData().forEach(nameplate -> 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());
+				log.error("消息处理失败", e);
-			Thread.currentThread().interrupt();
+			}
+		});
+		if (isSubmit){
+			log.info("nameplateKafkaConsumeracknowledge：{}");
+			ack.acknowledge(); // 手动提交 offset
 		}
 		log.info("nameplateKafkaConsumer消息处理完成，耗时：{}", System.currentTimeMillis() - startTime);