Spring Boot 快速集成Kafka

1. 添加依赖

       
            org.springframework.kafka
            spring-kafka
        

2.生产者配置

# 用于建立初始连接的broker地址
spring.kafka.bootstrap-servers=kafka ip:9092

# producer用到的key和value的序列化类
spring.kafka.producer.keyserializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.valueserializer=org.apache.kafka.common.serialization.StringSerializer
# 默认的批处理记录数
spring.kafka.producer.batch-size=16384
# 32MB的总发送缓存
spring.kafka.producer.buffer-memory=33554432
spring.kafka.producer.retries=3

3.发送消息

service 中添加

    @Autowired
    private KafkaTemplate kafkaTemplate;

kafkaTemplate.send("testTopic", msg);

4.消费者配置

## consumer用到的key和value的反序列化类
spring.kafka.consumer.keydeserializer=org.apache.kafka.common.serialization.IntegerDeserializer
spring.kafka.consumer.valuedeserializer=org.apache.kafka.common.serialization.StringDeserializer
# consumer的消费组id
spring.kafka.consumer.group-id=test_group
# 是否自动提交消费者偏移量
spring.kafka.consumer.enable-auto-commit=true
# 每隔100ms向broker提交一次偏移量
spring.kafka.consumer.auto-commit-interval=100
# 如果该消费者的偏移量不存在，则自动设置为最早的偏移量
spring.kafka.consumer.auto-offset-reset=earliest

5.消息监听 

新增消息监听类

@Component
public class MessageListener {

    @KafkaListener(topics = {"testTopic"},groupId = "test_group")
    public void listener(ConsumerRecord record){
        //获取消息
        String message = record.value();
        //消息偏移量
        long offset = record.offset();
        System.out.println("读取的消息："+message+"n当前偏移量："+offset);
    }
}