admin/vlv-ai-analytic-center
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

取消修改的数据ack

Browse Source
This commit is contained in:
ZLI263
2025-09-19 14:34:34 +08:00
parent 3fddbe1d79
commit 99179c33dc
1 changed files with 4 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaConsumer.java
View File

@@ -42,8 +42,8 @@ public class NameplateKafkaConsumer {
private ThreadPoolTaskExecutor executor; private ThreadPoolTaskExecutor executor;
private Semaphore semaphore = new Semaphore(2); // 限制并发数 private Semaphore semaphore = new Semaphore(2); // 限制并发数
@KafkaListener(topics = "${analyticCenterKafka.consumer.topic}", // = smart_assistant_nameplate_topic @KafkaListener(topics = "${analyticCenterKafka.consumer.topic}",// = smart_assistant_nameplate_topic
groupId = "${analyticCenterKafka.consumer.group}", //smart_assistant_nameplate_topic_group groupId = "${analyticCenterKafka.consumer.group}" ,//smart_assistant_nameplate_topic_group
containerFactory = "analyticCenterConsumerFactory", containerFactory = "analyticCenterConsumerFactory",
concurrency = "3") concurrency = "3")
public void listen(String recordMessages, Acknowledgment ack, public void listen(String recordMessages, Acknowledgment ack,
@@ -53,13 +53,14 @@ public class NameplateKafkaConsumer {
log.info("nameplateKafkaConsumerMessage，消息：{}", recordMessages); log.info("nameplateKafkaConsumerMessage，消息：{}", recordMessages);
if (StringUtils.isEmpty(recordMessages)) { if (StringUtils.isEmpty(recordMessages)) {
ack.acknowledge(); ack.acknowledge();
log.info("nameplateKafkaConsumerMessage，消息为空");
return; return;
} }
try { try {
NameplateTableKafkaDTO tmNameplateCorpus = JSON.parseObject(recordMessages, NameplateTableKafkaDTO.class); NameplateTableKafkaDTO tmNameplateCorpus = JSON.parseObject(recordMessages, NameplateTableKafkaDTO.class);
if (!"INSERT".equals(tmNameplateCorpus.getType()) || CollectionUtils.isEmpty(tmNameplateCorpus.getData())) { if (!"INSERT".equals(tmNameplateCorpus.getType()) || CollectionUtils.isEmpty(tmNameplateCorpus.getData())) {
ack.acknowledge(); log.info("nameplateKafkaConsumerMessage，不满足条件");
} }
tmNameplateCorpus.getData().forEach(nameplate -> tmNameplateCorpus.getData().forEach(nameplate ->