自助网站建设厦门网站制作,宣传片视频,网站建设网上商城,做产品封面的网站文章目录 1、生产者监听器1.1、创建生产者监听器1.2、创建生产者拦截器1.3、发送消息测试1.4、使用Java代码创建主题分区副本1.5、application.yml配置----v1版1.6、屏蔽 kafka debug 日志 logback.xml1.7、引入spring-kafka依赖1.8、控制台日志 1、生产者监听器
1.1、创建生产… 文章目录 1、生产者监听器1.1、创建生产者监听器1.2、创建生产者拦截器1.3、发送消息测试1.4、使用Java代码创建主题分区副本1.5、application.yml配置----v1版1.6、屏蔽 kafka debug 日志 logback.xml1.7、引入spring-kafka依赖1.8、控制台日志 1、生产者监听器
1.1、创建生产者监听器
package com.atguigu.kafka.listener;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.support.ProducerListener;
Component
public class MyKafkaProducerListener implements ProducerListenerString,String {//生产者 ack 配置为 0 只要发送即成功//ack为 1 leader落盘 broker ack之后 才成功//ack为 -1 分区所有副本全部落盘 broker ack之后 才成功Overridepublic void onSuccess(ProducerRecordString, String producerRecord, RecordMetadata recordMetadata) {//ProducerListener.super.onSuccess(producerRecord, recordMetadata);System.out.println(MyKafkaProducerListener消息发送成功topicproducerRecord.topic()partition producerRecord.partition()key producerRecord.key()value producerRecord.value()offset recordMetadata.offset());}//消息发送失败的回调监听器可以接收到发送失败的消息 可以记录失败的消息Overridepublic void onError(ProducerRecordString, String producerRecord, RecordMetadata recordMetadata, Exception exception) {System.out.println(MyKafkaProducerListener消息发送失败topicproducerRecord.topic()partition producerRecord.partition()key producerRecord.key()value producerRecord.value()offset recordMetadata.offset());System.out.println(异常信息 exception.getMessage());}
}
1.2、创建生产者拦截器
package com.atguigu.kafka.interceptor;
import 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 java.util.Map;
//拦截器必须手动注册给kafka生产者KafkaTemplate
Component
public class MyKafkaInterceptor implements ProducerInterceptorString,String {//kafka生产者发送消息前执行拦截发送的消息预处理Overridepublic ProducerRecordString, String onSend(ProducerRecordString, String producerRecord) {System.out.println(生产者即将发送消息topic producerRecord.topic()partition:producerRecord.partition()key producerRecord.key()value producerRecord.value());return null;}//kafka broker 给出应答后执行Overridepublic void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {//exception为空表示消息发送成功if(e null){System.out.println(消息发送成功topic recordMetadata.topic()partition:recordMetadata.partition()offsetrecordMetadata.offset()timestamprecordMetadata.timestamp());}}Overridepublic void close() {}Overridepublic void configure(MapString, ? map) {}
}
1.3、发送消息测试
package com.atguigu.kafka.producer;import com.atguigu.kafka.interceptor.MyKafkaInterceptor;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import java.io.IOException;SpringBootTest
class KafkaProducerApplicationTests {//装配kafka模板类 springboot启动时会自动根据配置文初始化kafka模板类对象注入到容器中ResourceKafkaTemplate kafkaTemplate;ResourceMyKafkaInterceptor myKafkaInterceptor;PostConstructpublic void init() {kafkaTemplate.setProducerInterceptor(myKafkaInterceptor);}Testvoid contextLoads() throws IOException {kafkaTemplate.send(my_topic1, spring-kafka-生产者监听器);//回调是等kafkaack以后才执行需要阻塞System.in.read();}
}
1.4、使用Java代码创建主题分区副本
package com.atguigu.kafka.config;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.stereotype.Component;
Component
public class KafkaTopicConfig {Beanpublic NewTopic myTopic1() {//相同名称的主题 只会创建一次后面创建的主题名称相同配置不同可以做增量更新分区、副本数return TopicBuilder.name(my_topic1)//主题名称.partitions(3)//主题分区.replicas(3)//主题分区副本数.build();//创建}
}
1.5、application.yml配置----v1版
server:port: 8110# v1
spring:kafka:bootstrap-servers: 192.168.74.148:9095,192.168.74.148:9096,192.168.74.148:9097producer: # producer 生产者retries: 0 # 重试次数 0表示不重试acks: -1 # 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、-1/all)batch-size: 16384 # 批次大小 单位bytebuffer-memory: 33554432 # 生产者缓冲区大小 单位bytekey-serializer: org.apache.kafka.common.serialization.StringSerializer # key的序列化器value-serializer: org.apache.kafka.common.serialization.StringSerializer # value的序列化器
1.6、屏蔽 kafka debug 日志 logback.xml
configuration !-- 如果觉得idea控制台日志太多src\main\resources目录下新建logback.xml
屏蔽kafka debug --logger nameorg.apache.kafka.clients leveldebug /
/configuration
1.7、引入spring-kafka依赖
?xml version1.0 encodingUTF-8?
project 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/ !-- lookup parent from repository --/parent!-- Generated by https://start.springboot.io --!-- 优质的 spring/boot/data/security/cloud 框架中文文档尽在 https://springdoc.cn --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.8、控制台日志
生产者即将发送消息topic my_topic1partition:nullkey nullvalue spring-kafka-生产者监听器
消息发送成功topic my_topic1partition:0offset0timestamp1717573749549
MyKafkaProducerListener消息发送成功topicmy_topic1partition nullkey nullvalue spring-kafka-生产者监听器offset 0[[{partition: 0,offset: 0,msg: spring-kafka-生产者监听器,timespan: 1717573749549,date: 2024-06-05 07:49:09}]
]
