kafka-生产者监听器（SpringBoot整合Kafka）

1、生产者监听器

1.1、创建生产者监听器

1. package com.atguigu.kafka.listener;
2. import org.apache.kafka.clients.producer.ProducerRecord;
3. import org.apache.kafka.clients.producer.RecordMetadata;
4. import org.springframework.kafka.support.ProducerListener;
5. @Component
6. public class MyKafkaProducerListener implements ProducerListener<String,String> {
8.     //生产者 ack 配置为 0 只要发送即成功
9.     //ack为 1  leader落盘  broker ack之后 才成功
10.     //ack为 -1 分区所有副本全部落盘  broker ack之后 才成功
11.     @Override
12.     public void onSuccess(ProducerRecord<String, String> producerRecord, RecordMetadata recordMetadata) {
13.         //ProducerListener.super.onSuccess(producerRecord, recordMetadata);
14.         System.out.println("MyKafkaProducerListener消息发送成功："+"topic="+producerRecord.topic()
15.         +"，partition = "+producerRecord.partition()
16.         +"，key = "+producerRecord.key()
17.         +"，value = "+producerRecord.value()
18.         +"，offset = "+recordMetadata.offset());
19.     }
21.     //消息发送失败的回调：监听器可以接收到发送失败的消息 可以记录失败的消息
22.     @Override
23.     public void onError(ProducerRecord<String, String> producerRecord, RecordMetadata recordMetadata, Exception exception) {
24.         System.out.println("MyKafkaProducerListener消息发送失败："+"topic="+producerRecord.topic()
25.                 +"，partition = "+producerRecord.partition()
26.                 +"，key = "+producerRecord.key()
27.                 +"，value = "+producerRecord.value()
28.                 +"，offset = "+recordMetadata.offset());
29.         System.out.println("异常信息：" + exception.getMessage());
30.     }
31. }

复制代码

1.2、创建生产者拦截器

1. package com.atguigu.kafka.interceptor;
2. import org.apache.kafka.clients.producer.ProducerInterceptor;
3. import org.apache.kafka.clients.producer.ProducerRecord;
4. import org.apache.kafka.clients.producer.RecordMetadata;
5. import org.springframework.stereotype.Component;
6. import java.util.Map;
7. //拦截器必须手动注册给kafka生产者（KafkaTemplate）
8. @Component
9. public class MyKafkaInterceptor implements ProducerInterceptor<String,String> {
10.     //kafka生产者发送消息前执行：拦截发送的消息预处理
11.     @Override
12.     public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
13.         System.out.println("生产者即将发送消息：topic = "+ producerRecord.topic()
14.         +"，partition:"+producerRecord.partition()
15.         +"，key = "+producerRecord.key()
16.         +"，value = "+producerRecord.value());
17.         return null;
18.     }
20.     //kafka broker 给出应答后执行
21.     @Override
22.     public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
23.         //exception为空表示消息发送成功
24.         if(e == null){
25.             System.out.println("消息发送成功：topic = "+ recordMetadata.topic()
26.                     +"，partition:"+recordMetadata.partition()
27.                     +"，offset="+recordMetadata.offset()
28.             +"，timestamp="+recordMetadata.timestamp());
29.         }
30.     }
32.     @Override
33.     public void close() {
35.     }
37.     @Override
38.     public void configure(Map<String, ?> map) {
40.     }
41. }

复制代码

1.3、发送消息测试

1. package com.atguigu.kafka.producer;
3. import com.atguigu.kafka.interceptor.MyKafkaInterceptor;
4. import jakarta.annotation.PostConstruct;
5. import jakarta.annotation.Resource;
6. import org.junit.jupiter.api.Test;
7. import org.springframework.boot.test.context.SpringBootTest;
8. import org.springframework.kafka.core.KafkaTemplate;
9. import java.io.IOException;
11. @SpringBootTest
12. class KafkaProducerApplicationTests {
14.     //装配kafka模板类： springboot启动时会自动根据配置文初始化kafka模板类对象注入到容器中
15.     @Resource
16.     KafkaTemplate kafkaTemplate;
18.     @Resource
19.     MyKafkaInterceptor myKafkaInterceptor;
21.     @PostConstruct
22.     public void init() {
23.         kafkaTemplate.setProducerInterceptor(myKafkaInterceptor);
24.     }
25.     @Test
26.     void contextLoads() throws IOException {
27.         kafkaTemplate.send("my_topic1", "spring-kafka-生产者监听器");
29.         //回调是等kafka，ack以后才执行，需要阻塞
30.         System.in.read();
31.     }
32. }

复制代码

1.4、利用Java代码创建主题分区副本

1. package com.atguigu.kafka.config;
2. import org.apache.kafka.clients.admin.NewTopic;
3. import org.springframework.context.annotation.Bean;
4. import org.springframework.kafka.config.TopicBuilder;
5. import org.springframework.stereotype.Component;
6. @Component
7. public class KafkaTopicConfig {
8.     @Bean
9.     public NewTopic myTopic1() {
10.         //相同名称的主题 只会创建一次，后面创建的主题名称相同配置不同可以做增量更新（分区、副本数）
11.         return TopicBuilder.name("my_topic1")//主题名称
12.                 .partitions(3)//主题分区
13.                 .replicas(3)//主题分区副本数
14.                 .build();//创建
15.     }
16. }

复制代码

1.5、application.yml配置----v1版

1. server:
2.   port: 8110
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.     producer: # producer 生产者
9.       retries: 0 # 重试次数 0表示不重试
10.       acks: -1 # 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、-1/all)
11.       batch-size: 16384 # 批次大小 单位byte
12.       buffer-memory: 33554432 # 生产者缓冲区大小 单位byte
13.       key-serializer: org.apache.kafka.common.serialization.StringSerializer # key的序列化器
14.       value-serializer: org.apache.kafka.common.serialization.StringSerializer # value的序列化器

复制代码

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.kafka</groupId>
15.     <artifactId>kafka-producer</artifactId>
16.     <version>0.0.1-SNAPSHOT</version>
17.     <name>kafka-producer</name>
18.     <description>kafka-producer</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>
35.         <dependency>
36.             <groupId>org.springframework.kafka</groupId>
37.             <artifactId>spring-kafka</artifactId>
38.         </dependency>
39.     </dependencies>
41.     <build>
42.         <plugins>
43.             <plugin>
44.                 <groupId>org.springframework.boot</groupId>
45.                 <artifactId>spring-boot-maven-plugin</artifactId>
46.             </plugin>
47.         </plugins>
48.     </build>
50. </project>

复制代码

1.8、控制台日志

1. 生产者即将发送消息：topic = my_topic1，partition:null，key = null，value = spring-kafka-生产者监听器
2. 消息发送成功：topic = my_topic1，partition:0，offset=0，timestamp=1717573749549
3. MyKafkaProducerListener消息发送成功：topic=my_topic1，partition = null，key = null，value = spring-kafka-生产者监听器，offset = 0

复制代码

1. [
2.   [
3.     {
4.       "partition": 0,
5.       "offset": 0,
6.       "msg": "spring-kafka-生产者监听器",
7.       "timespan": 1717573749549,
8.       "date": "2024-06-05 07:49:09"
9.     }
10.   ]
11. ]

复制代码

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