kafka-消耗者组（SpringBoot整合Kafka）

1、消耗者组

1.1、利用 efak 创建 主题 my_topic1 并建立6个分区并给每个分区建立3个副本

1.2、创建生产者发送消息

1. package com.atguigu.spring.kafka.consumer;
3. import jakarta.annotation.Resource;
4. import org.junit.jupiter.api.Test;
5. import org.springframework.boot.test.context.SpringBootTest;
6. import org.springframework.kafka.core.KafkaTemplate;
8. @SpringBootTest
9. class SpringKafkaConsumerApplicationTests {
10.     @Resource
11.     KafkaTemplate kafkaTemplate;
13.     @Test
14.     void contextLoads() {
15.         for (int i = 0; i < 10; i++) {
16.             kafkaTemplate.send("my_topic1",i%6,"", "消费者组"+i);
17.         }
18.     }
19. }

复制代码

1.3、application.yml配置

1. server:
2.   port: 8120
4. # v1
5. spring:
6.   Kafka:
7.     bootstrap-servers: 192.168.74.148:9095,192.168.74.148:9096,192.168.74.148:9097
8.     consumer:
9.       # read-committed读事务已提交的消息 解决脏读问题
10.       isolation-level: read-committed # 消费者的事务隔离级别：read-uncommitted会导致脏读，可以读取生产者事务还未提交的消息
11.       # 消费者是否自动ack ：true自动ack 消费者获取到消息后kafka提交消费者偏移量
12.       enable-auto-commit: true
13.       # 消费者提交ack时多长时间批量提交一次
14.       auto-commit-interval: 1000
15.       # 消费者第一次消费主题消息时从哪个位置开始
16.       auto-offset-reset: earliest  #指定Offset消费:earliest | latest | none
17.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
18.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

复制代码

1.4、创建消耗者监听器

1. package com.atguigu.spring.kafka.consumer.listener;
2. import org.apache.kafka.clients.consumer.ConsumerRecord;
3. import org.springframework.kafka.annotation.KafkaListener;
4. import org.springframework.stereotype.Component;
5. @Component
6. public class MyKafkaListener {
8.     @KafkaListener(topics ={"my_topic1"},groupId = "my_group1")
9.     public void onMessage1(ConsumerRecord<String, String> record) {
10.         System.out.println("my_group1消费者1获取到消息：topic = "+ record.topic()
11.                 +"，partition:"+record.partition()
12.                 +"，offset = "+record.offset()
13.                 +"，key = "+record.key()
14.                 +"，value = "+record.value());
15.     }
16.     @KafkaListener(topics ={"my_topic1"},groupId = "my_group1")
17.     public void onMessage2(ConsumerRecord<String, String> record) {
18.         System.out.println("my_group1消费者2获取到消息：topic = "+ record.topic()
19.                 +"，partition:"+record.partition()
20.                 +"，offset = "+record.offset()
21.                 +"，key = "+record.key()
22.                 +"，value = "+record.value());
23.     }
24.     @KafkaListener(topics ={"my_topic1"},groupId = "my_group2")
25.     public void onMessage3(ConsumerRecord<String, String> record) {
26.         System.out.println("my_group2消费者获取到消息：topic = "+ record.topic()
27.                 +"，partition:"+record.partition()
28.                 +"，offset = "+record.offset()
29.                 +"，key = "+record.key()
30.                 +"，value = "+record.value());
31.     }
33. }

复制代码

1.5、创建SpringBoot启动类

1. package com.atguigu.spring.kafka.consumer;
3. import org.springframework.boot.SpringApplication;
4. import org.springframework.boot.autoconfigure.SpringBootApplication;
7. // Generated by https://start.springboot.io
8. // 优质的 spring/boot/data/security/cloud 框架中文文档尽在 => https://springdoc.cn
9. @SpringBootApplication
10. public class SpringKafkaConsumerApplication {
12.     public static void main(String[] args) {
13.         SpringApplication.run(SpringKafkaConsumerApplication.class, args);
14.     }
16. }

复制代码

1.6、屏蔽 kafka debug 日记 logback.xml

1. <configuration>　　　　　　
2.     <!-- 如果觉得idea控制台日志太多，src\main\resources目录下新建logback.xml
3. 屏蔽kafka debug -->
4.     <logger name="org.apache.kafka.clients" level="debug" />
5. </configuration>

复制代码

1.7、引入spring-kafka依靠

1. <?xml version="1.0" encoding="UTF-8"?>
2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
3.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
4.     <modelVersion>4.0.0</modelVersion>
5.     <parent>
6.         <groupId>org.springframework.boot</groupId>
7.         <artifactId>spring-boot-starter-parent</artifactId>
8.         <version>3.0.5</version>
9.         <relativePath/> <!-- lookup parent from repository -->
10.     </parent>
12.     <!-- Generated by https://start.springboot.io -->
13.     <!-- 优质的 spring/boot/data/security/cloud 框架中文文档尽在 => https://springdoc.cn -->
14.     <groupId>com.atguigu</groupId>
15.     <artifactId>spring-kafka-consumer</artifactId>
16.     <version>0.0.1-SNAPSHOT</version>
17.     <name>spring-kafka-consumer</name>
18.     <description>spring-kafka-consumer</description>
19.     <properties>
20.         <java.version>17</java.version>
21.     </properties>
22.     <dependencies>
23.         <dependency>
24.             <groupId>org.springframework.boot</groupId>
25.             <artifactId>spring-boot-starter</artifactId>
26.         </dependency>
28.         <dependency>
29.             <groupId>org.springframework.boot</groupId>
30.             <artifactId>spring-boot-starter-test</artifactId>
31.             <scope>test</scope>
32.         </dependency>
34.         <dependency>
35.             <groupId>org.springframework.boot</groupId>
36.             <artifactId>spring-boot-starter-web</artifactId>
37.         </dependency>
38.         <dependency>
39.             <groupId>org.springframework.kafka</groupId>
40.             <artifactId>spring-kafka</artifactId>
41.         </dependency>
43.     </dependencies>
45.     <build>
46.         <plugins>
47.             <plugin>
48.                 <groupId>org.springframework.boot</groupId>
49.                 <artifactId>spring-boot-maven-plugin</artifactId>
50.             </plugin>
51.         </plugins>
52.     </build>
54. </project>

复制代码

1.8、消耗者控制台：

1.   .   ____          _            __ _ _
2. /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
3. ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
4. \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
5.   '  |____| .__|_| |_|_| |_\__, | / / / /
6. =========|_|==============|___/=/_/_/_/
7. :: Spring Boot ::                (v3.0.5)
9. my_group2消费者获取到消息：topic = my_topic1，partition:2，offset = 0，key = ，value = 消费者组2
10. my_group2消费者获取到消息：topic = my_topic1，partition:2，offset = 1，key = ，value = 消费者组8
11. my_group2消费者获取到消息：topic = my_topic1，partition:0，offset = 0，key = ，value = 消费者组0
12. my_group2消费者获取到消息：topic = my_topic1，partition:0，offset = 1，key = ，value = 消费者组6
13. my_group2消费者获取到消息：topic = my_topic1，partition:5，offset = 0，key = ，value = 消费者组5
14. my_group2消费者获取到消息：topic = my_topic1，partition:3，offset = 0，key = ，value = 消费者组3
15. my_group2消费者获取到消息：topic = my_topic1，partition:3，offset = 1，key = ，value = 消费者组9
16. my_group2消费者获取到消息：topic = my_topic1，partition:1，offset = 0，key = ，value = 消费者组1
17. my_group2消费者获取到消息：topic = my_topic1，partition:1，offset = 1，key = ，value = 消费者组7
18. my_group2消费者获取到消息：topic = my_topic1，partition:4，offset = 0，key = ，value = 消费者组4
19. my_group1消费者2获取到消息：topic = my_topic1，partition:2，offset = 0，key = ，value = 消费者组2
20. my_group1消费者2获取到消息：topic = my_topic1，partition:2，offset = 1，key = ，value = 消费者组8
21. my_group1消费者1获取到消息：topic = my_topic1，partition:5，offset = 0，key = ，value = 消费者组5
22. my_group1消费者1获取到消息：topic = my_topic1，partition:4，offset = 0，key = ，value = 消费者组4
23. my_group1消费者1获取到消息：topic = my_topic1，partition:3，offset = 0，key = ，value = 消费者组3
24. my_group1消费者1获取到消息：topic = my_topic1，partition:3，offset = 1，key = ，value = 消费者组9
25. my_group1消费者2获取到消息：topic = my_topic1，partition:1，offset = 0，key = ，value = 消费者组1
26. my_group1消费者2获取到消息：topic = my_topic1，partition:1，offset = 1，key = ，value = 消费者组7
27. my_group1消费者2获取到消息：topic = my_topic1，partition:0，offset = 0，key = ，value = 消费者组0
28. my_group1消费者2获取到消息：topic = my_topic1，partition:0，offset = 1，key = ，value = 消费者组6

复制代码

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