From f7bd2c2708e9b0d2f6f7103791bc0f659511f50e Mon Sep 17 00:00:00 2001 From: zren25 Date: Fri, 16 May 2025 14:31:45 +0800 Subject: [PATCH] sonar --- .../ai/analytic/center/mq/NameplateKafkaConsumer.java | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) 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 924e7b6..b9c6261 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 @@ -68,7 +68,7 @@ public class NameplateKafkaConsumer { log.info("nameplateKafkaConsumerParseType: {},size:{}", tmNameplateCorpus.getType(), tmNameplateCorpus.getData().size()); if(tmNameplateCorpus.getType().equals("INSERT")){ CountDownLatch latch = new CountDownLatch(tmNameplateCorpus.getData().size()); - tmNameplateCorpus.getData().forEach(nameplate -> { + tmNameplateCorpus.getData().forEach(nameplate -> CompletableFuture.runAsync(() -> { try { log.info("nameplateKafkaConsumerCustomerFlowId: {}",nameplate.getCustomerFlowId()); @@ -80,15 +80,14 @@ public class NameplateKafkaConsumer { log.info("nameplateKafkaConsumerack.acknowledge"); latch.countDown(); } - }, executor); - }); - boolean allDone = latch.await(2, TimeUnit.MINUTES); - log.info("nameplateKafkaConsumerack.acknowledge:{}",allDone); + }, executor)); try { ack.acknowledge(); // 确保无论成功与否都提交 offset } catch (IllegalStateException e) { log.warn("Offset 已提交，跳过重复提交"); } + boolean allDone = latch.await(2, TimeUnit.MINUTES); + log.info("nameplateKafkaConsumerack.acknowledge:{}",allDone); } }else { ack.acknowledge(); // 空消息直接跳过