1、生产者拦截器1.1、创建生产者拦截器1.2、KafkaTemplate配置生产者拦截器1.3、使用Java代码创建主题分区副本1.4、application.yml配置----v1版1.5、屏蔽

kafka

logback.xml1.6、引入spring-kafka依赖1.7、控制台日志

1、生产者拦截器

org.apache.kafka.clients.producer.ProducerInterceptor;

import

org.apache.kafka.clients.producer.ProducerRecord;

import

org.apache.kafka.clients.producer.RecordMetadata;

import

org.springframework.stereotype.Component;

import

//拦截器必须手动注册给kafka生产者KafkaTemplate

Component

ProducerInterceptorString,String

{//kafka生产者发送消息前执行拦截发送的消息预处理Overridepublic

String

{System.out.println(生产者即将发送消息topic

producerRecord.topic()partition:producerRecord.partition()key

null;}//kafka

onAcknowledgement(RecordMetadata

recordMetadata,

null){System.out.println(消息发送成功topic

recordMetadata.topic()partition:recordMetadata.partition()offsetrecordMetadata.offset()timestamprecordMetadata.timestamp());}}Overridepublic

void

com.atguigu.kafka.producer;import

com.atguigu.kafka.interceptor.MyKafkaInterceptor;

import

jakarta.annotation.PostConstruct;

import

org.springframework.boot.test.context.SpringBootTest;

import

org.springframework.kafka.core.KafkaTemplate;

import

java.io.IOException;SpringBootTest

class

springboot启动时会自动根据配置文初始化kafka模板类对象注入到容器中ResourceKafkaTemplate

kafkaTemplate;ResourceMyKafkaInterceptor

myKafkaInterceptor;PostConstructpublic

void

{kafkaTemplate.setProducerInterceptor(myKafkaInterceptor);}Testvoid

contextLoads()

spring-kafka-生产者拦截器);//回调是等kafkaack以后才执行需要阻塞System.in.read();}

1.3、使用Java代码创建主题分区副本

org.apache.kafka.clients.admin.NewTopic;

import

org.springframework.context.annotation.Bean;

import

org.springframework.kafka.config.TopicBuilder;

import

org.springframework.stereotype.Component;

Component

只会创建一次后面创建的主题名称相同配置不同可以做增量更新分区、副本数return

TopicBuilder.name(my_topic1)//主题名称.partitions(3)//主题分区.replicas(3)//主题分区副本数.build();//创建}

server:port:

spring:kafka:bootstrap-servers:

192.168.74.148:9095,192.168.74.148:9096,192.168.74.148:9097producer:

producer

应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、-1/all)batch-size:

16384

org.apache.kafka.common.serialization.StringSerializer

org.apache.kafka.common.serialization.StringSerializer

1.5、屏蔽

如果觉得idea控制台日志太多src\main\resources目录下新建logback.xml

屏蔽kafka

xmlnshttp://maven.apache.org/POM/4.0.0

xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0

https://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersionparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion3.0.5/versionrelativePath/

!--

spring/boot/data/security/cloud

框架中文文档尽在

--groupIdcom.atguigu.kafka/groupIdartifactIdkafka-producer/artifactIdversion0.0.1-SNAPSHOT/versionnamekafka-producer/namedescriptionkafka-producer/descriptionpropertiesjava.version17/java.version/propertiesdependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependencydependencygroupIdorg.springframework.kafka/groupIdartifactIdspring-kafka/artifactId/dependency/dependenciesbuildpluginsplugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactId/plugin/plugins/build/project

1.7、控制台日志

my_topic1partition:0offset0timestamp1717490776329[[{partition:

0,offset: