SpringBoot集成Kafka

SpringBoot集成Kafka

知识索引

• SpringBoot集成Kafka工程搭建
• SpringBoot集成Kafka配置
• SpringBoot集成Kafka生产消息
• SpringBoot集成Kafka消费消息

1 生产者

SpringBoot

集成

Kafka

，无非就是生产者和消费者，但首先得实现

SpringBoot

集成，流程如下：

1:pom.xml引入依赖包

2:创建启动类

3:配置核心配置文件application.yml

1.1 pom.xml

在

pom.xml

中引入

spring-kafka

，代码如下：

<?xml version="1.0" encoding="UTF-8"?><projectxmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.itmentu</groupId><artifactId>kafka-producer-01</artifactId><version>1.0-SNAPSHOT</version><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.3.8.RELEASE</version></parent><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.62</version></dependency></dependencies></project>

1.2 启动类

创建启动类

com.itmentu.KafkaProducerApplication

，代码如下：

@SpringBootApplicationpublicclassKafkaProducerApplication{publicstaticvoidmain(String[] args){SpringApplication.run(KafkaProducerApplication.class,args);}}

1.3 核心配置文件

创建

application.yml

核心配置文件，代码如下：

server:port:18081spring:kafka:bootstrap-servers: 192.168.211.130:9092,192.168.211.130:9093,192.168.211.130:9094producer:# producer 生产者retries:0# 重试次数acks:1# 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)batch-size:16384# 批量大小buffer-memory:33554432# 生产端缓冲区大小key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

参数说明：

bootstrap-servers:Kafka服务地址
retries:发送失败后，重试次数，0表示不重试
key-serializer:向Kafka发送数据，key采用的序列化方式
value-serializer:向Kafka发送数据，数据采用的序列化方式

1.4 生产者消息发送

消息发送使用

KafkaTemplate

实现消息发送，创建

com.itmentu.controller.SendController

，实现消息发送，代码如下：

@RestController@RequestMapping(value ="/producer")publicclassSendController{@AutowiredprivateKafkaTemplate kafkaTemplate;/***
     * 发送消息
     * topic:要发送的队列
     * msg:发送的消息
     */@GetMapping(value ="/send/{topic}/{msg}")publicStringsend(@PathVariable(value ="topic")String topic,@PathVariable(value ="msg")String msg){//消息发送
        kafkaTemplate.send(topic,msg);return"SUCCESS";}}

发送消息的时候，使用的是

kafkaTemplate.send(topic,msg)

，第1个参数是队列名字或者叫主题名字，第2个参数是发送的消息。

我们可以请求

http://localhost:18081/producer/send/itmentu/hello

测试，

eagle

打开，可以看到效果如下：

2 消费者

SpringBoot

集成

Kafka

实现消费者消费消息，流程如下：

1:pom.xml引入依赖包

2:创建启动类

3:配置核心配置文件application.yml

2.1 pom.xml

和生产者一样，都需要引入

spring-kafka

依赖，配置如下：

<?xml version="1.0" encoding="UTF-8"?><projectxmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.itmentu</groupId><artifactId>kafka-consumer-01</artifactId><version>1.0-SNAPSHOT</version><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.3.8.RELEASE</version></parent><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.62</version></dependency></dependencies></project>

2.2 启动类

创建启动类

com.itmentu.KafkaConsumerApplication

，代码如下：

@SpringBootApplicationpublicclassKafkaConsumerApplication{publicstaticvoidmain(String[] args){SpringApplication.run(KafkaConsumerApplication.class,args);}}

2.3 核心配置文件

创建

application.yml

，配置如下：

server:port:18082spring:kafka:bootstrap-servers: 192.168.211.130:9092,192.168.211.130:9093,192.168.211.130:9094consumer:# consumer消费者group-id: mentugroup # 默认的消费组IDenable-auto-commit:true# 是否自动提交offsetauto-commit-interval:100# 提交offset延时(接收到消息后多久提交offset)# earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费# latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据# none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常auto-offset-reset: latest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

参数说明：

bootstrap-servers:Kafka服务地址
group-id:消费者组默认名字
enable-auto-commit:每次消费数据后，需要提交数据的偏移量，这里true表示自动提交，false表示手动提交
key-deserializer:读Kafka数据，key采用的反序列化方式
value-deserializer:读Kafka数据，数据采用的反序列化方式

2.4 消费消息

Kafka

消费消息采用

@KafkaListener(topics="{}",groupId="")

方式消费数据，我们可以创建一个消息监听类实现消息消费，创建

com.itmentu.listener.MessageListener

，代码如下：

@ComponentpublicclassMessageListener{@KafkaListener(topics ={"itmentu"},groupId ="itmentuGroup")publicvoidlistener(ConsumerRecord<String,String>record){//获取消息String message =record.value();//消息偏移量long offset =record.offset();System.out.println("读取的消息："+message+"\n当前偏移量："+offset);}}

测试效果如下：