SpingBoot集成kafka-发送读取消息示例

kafka的几个常见概念

1、springboot和kafka对应版本（重要）

https://spring.io/projects/spring-kafka

2、创建springboot项目，引入kafka依靠

2.1、生产者EventProducer

1. package com.power.producer;
3. import org.springframework.kafka.core.KafkaTemplate;
4. import org.springframework.stereotype.Component;
6. import javax.annotation.Resource;
8. @Component
9. public class EventProducer {
11.     @Resource
12.     private KafkaTemplate<String,String> kafkaTemplate;
14.     public void sendEvent(){
15.         kafkaTemplate.send("hello-topic","hello kafka");
16.     }
17. }

复制代码

2.2、消耗者EventConsumer

1. package com.power.consumer;
3. import org.springframework.kafka.annotation.KafkaListener;
4. import org.springframework.stereotype.Component;
6. @Component
7. public class EventConsumer {
9.     //采用监听的方式接收事件（消息，数据）
10.     @KafkaListener(topics = {"hello-topic"},groupId="hello-group")
11.     public void onEvent(String event){
12.         System.out.println("读取到的事件："+event);
13.     }
14. }

复制代码

2.3、启动生产者的方法SpringBoot01KafkaBaseApplication

实行一次该方法，会调用一次生产者发送一次消息。
即每实行一次，会调用EventProducer类下的sendEvent方法一次。

1. package com.power;
3. import com.power.producer.EventProducer;
4. import org.junit.jupiter.api.Test;
5. import org.springframework.boot.test.context.SpringBootTest;
7. import javax.annotation.Resource;
9. @SpringBootTest
10. public class SpringBoot01KafkaBaseApplication {
12.     @Resource
13.     private EventProducer eventProducer;
15.     @Test
16.     void test01(){
17.         eventProducer.sendEvent();
18.     }
19. }

复制代码

2.4、application.yml

1. spring:
2.   application:
3.     #应用名称
4.     name: spring-boot-01-kafka-base
6.   #kafka连接地址（ip+port）
7.   kafka:
8.     bootstrap-servers: <你的服务器ip>:9092
9.     #配置生产者(有24个配置)
10.     #producer:
11.     #配置消费者(有24个配置)
12.     #consumer:

复制代码

2.5、pom.xml

1. <?xml version="1.0" encoding="UTF-8"?>
2. <project xmlns="http://maven.apache.org/POM/4.0.0"
3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5.     <modelVersion>4.0.0</modelVersion>
7.     <parent>
8.         <groupId>org.springframework.boot</groupId>
9.         <artifactId>spring-boot-starter-parent</artifactId>
10.         <version>2.7.2</version>
11.         <relativePath />
12.     </parent>
14.     <groupId>org.powernode</groupId>
15.     <artifactId>spring-boot-01-kafka-base</artifactId>
16.     <version>0.0.1-SNAPSHOT</version>
17.     <name>kafkaSpringBootProject</name>
18.     <description>kafka project for Spring Boot</description>
20.     <properties>
21.         <java.version>8</java.version>
22.     </properties>
25.     <repositories>
26.         <repository>
27.             <id>central</id>
28.             <name>aliyun maven</name>
29.             <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
30.             <layout>default</layout>
31.             <!-- 是否开启发布版构件下载 -->
32.             <releases>
33.                 <enabled>true</enabled>
34.             </releases>
35.             <!-- 是否开启快照版构件下载 -->
36.             <snapshots>
37.                 <enabled>false</enabled>
38.             </snapshots>
39.         </repository>
40.     </repositories>
42.     <dependencies>
43.         <dependency>
44.             <groupId>org.springframework.boot</groupId>
45.             <artifactId>spring-boot-starter</artifactId>
46.         </dependency>
47.         <!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
48.         <dependency>
49.             <groupId>org.springframework.kafka</groupId>
50.             <artifactId>spring-kafka</artifactId>
51.             <version>2.8.0</version>
52.         </dependency>
53.         <dependency>
54.             <groupId>org.springframework.boot</groupId>
55.             <artifactId>spring-boot-devtools</artifactId>
56.             <scope>runtime</scope>
57.             <optional>true</optional>
58.         </dependency>
59.         <dependency>
60.             <groupId>org.springframework.boot</groupId>
61.             <artifactId>spring-boot-starter-test</artifactId>
62.             <scope>test</scope>
63.         </dependency>
64.         <!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka-test -->
65.         <dependency>
66.             <groupId>org.springframework.kafka</groupId>
67.             <artifactId>spring-kafka-test</artifactId>
68.             <version>2.8.0</version>
69.             <scope>test</scope>
70.         </dependency>
71.     </dependencies>
73.     <build>
74.         <plugins>
75.             <plugin>
76.                 <groupId>org.springframework.boot</groupId>
77.                 <artifactId>spring-boot-maven-plugin</artifactId>
78.                 <configuration>
79.                     <excludes>
80.                         <exclude>
81.                             <groupId>org.projectlombok</groupId>
82.                             <artifactId>lombok</artifactId>
83.                         </exclude>
84.                     </excludes>
85.                 </configuration>
86.             </plugin>
87.         </plugins>
88.     </build>
90. </project>

复制代码

2.6、启动springboot项目标启动类（Application）报错

项目启动类

1. package com.power;
3. import org.springframework.boot.SpringApplication;
4. import org.springframework.boot.autoconfigure.SpringBootApplication;
6. @SpringBootApplication
7. public class Application {
9.         public static void main(String[] args) {
10.                 SpringApplication.run(Application.class, args);
11.                 System.out.println("启动成功--------------------------");
12.         }
13. }

复制代码

启动服务后发现报错：

修改server.properties设置文件：

修改前：

修改后：

3、springboot集成kafka读取最早的消息

已经被消耗者读取/消耗的消息，无法被新启动的消耗组消息的，那么新启动的消耗组该如何读取最早的消息呢，可以通过设置消耗者auto-offset-reset: earliest去实现。

3.1、如何设置消耗者auto-offset-reset: earliest

1、修改application.yml

3.2、设置消耗者auto-offset-reset: earliest后存在的题目

3.2.1、修改消耗组ID

原消耗组ID

修改后的消耗组ID

4、新的消耗组ID乐成读取到之前的消息

3.2.2、手动重置偏移量

3.2.2.1、手动将偏移量设置为最早

1. #示例：./kafka-consumer-groups.sh --bootstrap-server <your-kafka-bootstrap-servers> --group <your-consumer-group> --topic <your-topic> --reset-offsets --to-earliest --execute

复制代码

来到kafka安装目录下：

实行如下下令：

1. ./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --group hello-group --topic hello-topic --reset-offsets --to-earliest --execute

复制代码

实行后报错

需要先停掉服务，在去手动重置偏移量，此时重置偏移量乐成，偏移量为0

3.2.2.2、手动将偏移量设置为最新

1. #示例：./kafka-consumer-groups.sh --bootstrap-server <your-kafka-bootstrap-servers> --group <your-consumer-group> --topic <your-topic> --reset-offsets --to-latest --execute

复制代码

1. ./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --group hello-group --topic hello-topic --reset-offsets --to-latest --execute

复制代码

设置乐成，此时偏移量已为最新：

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。