SpringBoot+Kafka

---

一、依赖

<!-- spring-kafka（与kafka的版本一致） -->
<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.5.1.RELEASE</version>
</dependency>

二、配置文件

spring:kafka:# kafka地址，集群用逗号分隔（localhost:9092,localhost:9093）。缺省：localhost:9092bootstrap-servers: localhost:9092# 生产者#producer:# key的序列化方式，缺省：org.apache.kafka.common.serialization.StringSerializer#key-serializer: org.apache.kafka.common.serialization.StringSerializer# value的序列化方式，缺省：org.apache.kafka.common.serialization.StringSerializer#value-serializer: org.apache.kafka.common.serialization.StringSerializer# 消费者consumer:# 消费者组group-id: testGroup# 自动偏移量# earliest：当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时，从头开始消费# latest：当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时，消费新产生的该分区下的数据# none：topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset，则抛出异常auto-offset-reset: latest# key的序列化方式，缺省：org.apache.kafka.common.serialization.StringSerializer#key-deserializer: org.apache.kafka.common.serialization.StringDeserializer# value的序列化方式，缺省：org.apache.kafka.common.serialization.StringSerializer#value-deserializer: org.apache.kafka.common.serialization.StringDeserializer#listener:# SINGLE-单个消费；BATCH-批量消费。缺省SINGLE#type: BATCH# 消费者监听的主题不存在时，启动项目是否报错。缺省：false#missing-topics-fatal: false

三、API

1、生产者

/*** 生产消息** @author kimi* @date 2023/2/18*/
@Component
public class ProducerMsg {@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;/*** 生产消息** @param msg*/public void send(String topic, String msg) {kafkaTemplate.send(topic, msg);}/*** 生产消息+回调** @param topic* @param msg*/public void sendCallback(String topic, String msg) {kafkaTemplate.send(topic, msg).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {//成功的回调@Overridepublic void onSuccess(SendResult<String, String> stringStringSendResult) {RecordMetadata recordMetadata = stringStringSendResult.getRecordMetadata();//主题final String topic = recordMetadata.topic();//分区final int partition = recordMetadata.partition();//偏移量final long offset = recordMetadata.offset();System.err.println(String.format("生产消息成功：topic: %s，partition: %s，offset: %s", topic, partition, offset));}//失败的回调@Overridepublic void onFailure(Throwable throwable) {}});}}

2、消费者

/*** 消费者** @author kimi* @date 2023/2/18*/
@Component
public class ConsumeMsg {/*** 单个消费** @param consumer*/@KafkaListener(topics = {"USER", "LOG"})public void consumeSingle(ConsumerRecord<String, String> consumer) {System.err.println("监听到kafka消息: " + consumer);final String topic = consumer.topic();final String value = consumer.value();}/*** 批量消费* 需将配置文件中的listener.type设置成BATCH** @param consumers*///@KafkaListener(topics = {"USER", "LOG"})public void consumeBatch(List<ConsumerRecord<String, String>> consumers) {consumers.forEach(consumer -> {final String topic = consumer.topic();final String value = consumer.value();System.err.println(String.format("topic: %s，value: %s", topic, value));});}}