SpringBoot 集成Kafka的demo

Window环境下SpringBoot 集成Kafka

一、windows 环境下启动 Kafka

1.kafka前提

由于kafka是基于zookeeper运行的，在kafka2.x的版本中，自带了zk，本文介绍的是基于2.X版本的集成

2.kafka 本地启动

下载地址：https://kafka.apache.org/downloads

2.1 解压后修改/config两个配置文件：

kafka安装目录：D:\software\kafka\kafka_2.13-3.1.0

zookeeper.properties

dataDir=D:\\software\\kafka\\kafka_2.13-3.1.0\\tmp\\zookeeper 

server.properties

log.dirs=D:\\software\\kafka\\kafka_2.13-3.1.0\\tmp\\kafka 

2.2 启动zookeeper命令

zookeeper进入windows目录执行：

zookeeper-server-start.bat …/…/config/zookeeper.properties 

2.3 启动 kafka

kafka 进入windows目录执行：

kafka-server-start.bat …/…/config/server.properties 

2.4 第三个窗口用来执行 创建topic

PS：2.X以上的版本由原先的 –zookeeper localhost:2181 更改为现在的 –bootstrap-server localhost:9092

kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic demo 

查看是否创建成功

kafka-topics.bat --list --bootstrap-server localhost:9092 

二、springBoot集成Kafka

1.引入依赖

 <!-- kafka --> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> 

2.kafka配置文件

spring: kafka: bootstrap-servers: 127.0.0.1:9092 producer: # 发生错误后，消息重发的次数。 retries: 0 #当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。 batch-size: 16384 # 设置生产者内存缓冲区的大小。 buffer-memory:  # 键的序列化方式 key-serializer: org.apache.kafka.common.serialization.StringSerializer # 值的序列化方式 value-serializer: org.apache.kafka.common.serialization.StringSerializer # acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。 # acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。 # acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。 acks: 1 consumer: # 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D auto-commit-interval: 1S # 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理： # latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录） # earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录 auto-offset-reset: earliest # 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量 enable-auto-commit: false # 键的反序列化方式 key-deserializer: org.apache.kafka.common.serialization.StringDeserializer # 值的反序列化方式 value-deserializer: org.apache.kafka.common.serialization.StringDeserializer group-id: consumer01 listener: # 在侦听器容器中运行的线程数。 concurrency: 5 #listner负责ack，每调用一次，就立即commit ack-mode: manual_immediate missing-topics-fatal: false 

3.生产者Producer

package com.hy.demo.kafka; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.SendResult; import org.springframework.stereotype.Component; import org.springframework.util.concurrent.ListenableFutureCallback; / * 生产者 * @author hsq */ @Component public class KafkaProducer { 
    private static final Logger log= LoggerFactory.getLogger(KafkaProducer.class); @Autowired private KafkaTemplate<String, Object> kafkaTemplate; / * 生产者简单发送消息 * @param topic * @param msg */ public void send(String topic,String msg){ 
    System.out.println("发送信息内容："+msg); kafkaTemplate.send(topic,msg); } / * 回调发送信息 是否成功 * @param topic * @param msg */ public void CallBackSend01(String topic,String msg){ 
    System.out.println("发送信息内容："+msg); kafkaTemplate.send(topic, msg).addCallback(success -> { 
    // 消息发送到的topic String topics = success.getRecordMetadata().topic(); // 消息发送到的分区 int partition = success.getRecordMetadata().partition(); // 消息在分区内的offset long offset = success.getRecordMetadata().offset(); System.out.println("发送消息成功:" + topics + "-" + partition + "-" + offset); }, failure -> { 
    System.out.println("发送消息失败:" + failure.getMessage()); }); } / * 第二种方法 * 回调发送信息 是否成功 * @param topic * @param msg */ public void CallBackSend02(String topic,String msg){ 
    System.out.println("发送信息内容："+msg); kafkaTemplate.send(topic, msg).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() { 
    @Override public void onFailure(Throwable ex) { 
    System.out.println("发送消息失败："+ex.getMessage()); } @Override public void onSuccess(SendResult<String, Object> result) { 
    System.out.println("发送消息成功：" + result.getRecordMetadata().topic() + "-" + result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset()); } }); } } 

4.消费者Consumer

package com.hy.demo.kafka; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component; / * kafka消费者 * @author hsq */ @Component public class KafkaConsumer { 
    / * 消费者1监听 * @param record */ @KafkaListener(groupId ="groupId01",topics = { 
   "demo"}) public void onMessage1(ConsumerRecord<?, ?> record){ 
    // 消费的哪个topic、partition的消息,打印出消息内容 System.out.println("简单消费："+record.topic()+"-"+record.partition()+"-"+record.value()); } / * 消费者2监听 * @param record */ @KafkaListener(groupId ="groupId02",topics = { 
   "demo"}) public void onMessage2(ConsumerRecord<?, ?> record){ 
    // 消费的哪个topic、partition的消息,打印出消息内容 System.out.println("简单消费2："+record.topic()+"-"+record.partition()+"-"+record.value()); } } 

5.测试类

package com.hy.demo; import com.hy.demo.kafka.KafkaProducer; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; @SpringBootTest public class KafkaTest { 
    @Autowired private KafkaProducer kafkaProducer; @Test public void kafkaProducer(){ 
    kafkaProducer.send("demo","发送第一条信息msg"); } @Test public void kafkaProducerCallBack(){ 
    kafkaProducer.CallBackSend01("demo","hello world callBack"); } @Test public void kafkaProducerCallBack2(){ 
    kafkaProducer.CallBackSend02("demo","hello world callBack2"); } } 

消费者：