package com.tplink.cloud.kafka;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* spring-kafka 消费者
*/
@Slf4j
@Component
public class SpringKafkaSingleConsumer {
/**
* 因为有多个消费者，所以先用命令创建一个多分区的主题（防止出现消费者没有分区可以消费）
*/
public static final String SPRING_TEST_TOPIC = "SPRING_TEST_TOPIC";
public static final String SPRING_TEST_GROUP_A = "SPRING_TEST_GROUP_A";
public static final String SPRING_TEST_GROUP_A_CLIENT_B = "SPRING_TEST_GROUP_A_CLIENT_B";
@KafkaListener(topics = {SPRING_TEST_TOPIC}, groupId = "SingleConsumer", id = SPRING_TEST_GROUP_A_CLIENT_B)
public void consumeB(List<ConsumerRecord<String, String>> records, Acknowledgment ack) throws InterruptedException{
for (ConsumerRecord<String, String> record : records){
log.info("SingleConsumer：topic = {}, offset = {}, value = {}", record.topic(),record.offset(),record.value());
}
ack.acknowledge();
}
}
kafka code11
需积分: 0 82 浏览量
2022-09-19
12:33:30
上传
评论
收藏 20KB ZIP 举报
kafka.zip （21个子文件） 
kafka 
pom.xml 3KB target classes application.yml 767B com tplink cloud kafka SpringKafkaProducer.class 1KB KafkaRunner.class 1KB SpringKafkaConcurrencyConsumer.class 2KB SpringKafkaSingleConsumer.class 2KB DemoApplication.class 733B generated-sources annotations src test main resources application.yml 767B java com tplink cloud kafka SpringKafkaConcurrencyConsumer.java 1KB SpringKafkaSingleConsumer.java 1KB SpringKafkaProducer.java 614B KafkaRunner.java 652B DemoApplication.java 319B .idea uiDesigner.xml 9KB misc.xml 430B jarRepositories.xml 1KB encodings.xml 186B compiler.xml 688B workspace.xml 4KB .gitignore 176B .gitignore 62B
评论0