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

提交验证kafka方法

Browse Source
This commit is contained in:
zren25
2025-03-31 21:15:13 +08:00
parent a21aa10251
commit 9fc695f93d
1 changed files with 3 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/TestKafkaListener.java
View File

@@ -1,5 +1,6 @@
package com.volvo.ai.analytic.center.mq; package com.volvo.ai.analytic.center.mq;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*; import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.TopicPartition;
@@ -18,6 +19,7 @@ import java.util.Map;
import java.util.Properties; import java.util.Properties;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@Slf4j
@Component @Component
@RestController @RestController
@RefreshScope @RefreshScope
@@ -61,7 +63,7 @@ public class TestKafkaListener {
while(keepOnReading){ while(keepOnReading){
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records){ for (ConsumerRecord<String, String> record : records){
System.out.println(" testKakfa Message received " + record.value() + ", partition " + record.partition() + ", offset=" + record.offset() + ", timestamp=" + record.timestamp()); log.info(" testKakfa Message received " + record.value() + ", partition " + record.partition() + ", offset=" + record.offset() + ", timestamp=" + record.timestamp());
} }
} }
} }