文章目录
1、kafka数据传递语义 2、kafka生产者事务 3、事务消息发送 3.1、application.yml配置 3.2、创建生产者监听器 3.3、创建生产者拦截器 3.4、发送消息测试 3.5、使用Java代码创建主题分区副本 3.6、屏蔽 kafka debug 日志 logback.xml 3.7、引入spring-kafka依赖 3.8、控制台日志
kafka发送消息时是否需要重试
仅发送一次：生产者发送消息后不重试，只发送一次 可能丢失消息 效率最高至少一次：生产者发送消息后重试，可能重试多次 效率差精准一次发送：生产者发送消息后无论是否重复发送 发送了多少次，在 kafka broker 中只保存一次消息，通过幂等性 + 生产者事务来实现
kafak天然支持幂等性，每个消息头中带了一个唯一的标志 kafka broker 根据此标志判断消息是否已经发送过，生产者事务可以保证数据没有最终发送成功时，消费者不可以消费，如果生产者发送消息时出现异常会自动回滚（清除之前发送的事务中的消息）
kafka天然幂等性：但是指的是生产者事务 生产消息时的幂等性，发送消息时消息中带唯一标识、broker接收到消息时如果重复不再保存，事务没提交消费者不能消费改消息
保证消息生产的幂等性一组消息要么一起成功 被消费者消息 要么一起失败都不能被消费者消费
配置ack为-1 分区所有副本均落盘成功配置生产者重试（发送失败可以继续发送：需要保证发送失败后再次发送消息到kafka实现 精准一次发送）需要给事务分配事务id（区分一个事务中的多条消息）
server :
port : 8110
spring :
kafka :
bootstrap-servers : 192.168.74.148: 9095 , 192.168.74.148: 9096 , 192.168.74.148: 9097
producer :
retries : 1
acks : -1
transaction-id-prefix : tx_
batch-size : 16384
buffer-memory : 33554432
key-serializer : org.apache.kafka.common.serialization.StringSerializer
value-serializer : org.apache.kafka.common.serialization.StringSerializer
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 ProducerListener < String , String > {
@Override
public void onSuccess ( ProducerRecord < String , String > , RecordMetadata recordMetadata) {
System . out. println ( "MyKafkaProducerListener消息发送成功：" + "topic=" + producerRecord. topic ( )
+ "，partition = " + producerRecord. partition ( )
+ "，key = " + producerRecord. key ( )
+ "，value = " + producerRecord. value ( )
+ "，offset = " + recordMetadata. offset ( ) ) ;
}
@Override
public void onError ( ProducerRecord < String , String > , RecordMetadata recordMetadata, Exception exception) {
System . out. println ( "MyKafkaProducerListener消息发送失败：" + "topic=" + producerRecord. topic ( )
+ "，partition = " + producerRecord. partition ( )
+ "，key = " + producerRecord. key ( )
+ "，value = " + producerRecord. value ( )
+ "，offset = " + recordMetadata. offset ( ) ) ;
System . out. println ( "异常信息：" + exception. getMessage ( ) ) ;
}
}
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 ;
@Component
public class MyKafkaInterceptor implements ProducerInterceptor < String , String > {
@Override
public ProducerRecord < String , String > onSend ( ProducerRecord < String , String > ) {
System . out. println ( "生产者即将发送消息：topic = " + producerRecord. topic ( )
+ "，partition:" + producerRecord. partition ( )
+ "，key = " + producerRecord. key ( )
+ "，value = " + producerRecord. value ( ) ) ;
return null ;
}
@Override
public void onAcknowledgement ( RecordMetadata recordMetadata, Exception e) {
if ( e == null ) {
System . out. println ( "消息发送成功：topic = " + recordMetadata. topic ( )
+ "，partition:" + recordMetadata. partition ( )
+ "，offset=" + recordMetadata. offset ( )
+ "，timestamp=" + recordMetadata. timestamp ( ) ) ;
}
}
@Override
public void close ( ) {
}
@Override
public void configure ( Map < String , ? > ) {
}
}
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 {
@Resource
KafkaTemplate kafkaTemplate;
@Resource
MyKafkaInterceptor myKafkaInterceptor;
@PostConstruct
public void init ( ) {
kafkaTemplate. setProducerInterceptor ( myKafkaInterceptor) ;
}
@Test
void contextLoads ( ) throws IOException {
kafkaTemplate. send ( "my_topic1" , "spring-kafka-生产者监听器" ) ;
System . in. read ( ) ;
}
@Test
void testTransaction ( ) throws IOException {
kafkaTemplate. executeInTransaction ( ( var1) -> {
var1. send ( "my_topic1" , 0 , "" , "spring-kafka-事务1" ) ;
var1. send ( "my_topic1" , 0 , "" , "spring-kafka-事务2" ) ;
int i = 1 / 0 ;
var1. send ( "my_topic1" , 0 , "" , "spring-kafka-事务3" ) ;
return "发送消息失败" ;
} ) ;
System . in. read ( ) ;
}
}
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 {
@Bean
public NewTopic myTopic1 ( ) {
return TopicBuilder . name ( "my_topic1" )
. partitions ( 3 )
. replicas ( 3 )
. build ( ) ;
}
}
< configuration> < loggername = " org.apache.kafka.clients" level = " debug" /> </ configuration> <?xml version="1.0" encoding="UTF-8"?>
< projectxmlns = " http://maven.apache.org/POM/4.0.0" xmlns: xsi= " http://www.w3.org/2001/XMLSchema-instance" xsi: schemaLocation= " http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd" > < modelVersion> </ modelVersion> < parent> < groupId> </ groupId> < artifactId> </ artifactId> < version> </ version> < relativePath/> </ parent> < groupId> </ groupId> < artifactId> </ artifactId> < version> </ version> < name> </ name> < description> </ description> < properties> < java.version> </ java.version> </ properties> < dependencies> < dependency> < groupId> </ groupId> < artifactId> </ artifactId> </ dependency> < dependency> < groupId> </ groupId> < artifactId> </ artifactId> < scope> </ scope> </ dependency> < dependency> < groupId> </ groupId> < artifactId> </ artifactId> </ dependency> </ dependencies> < build> < plugins> < plugin> < groupId> </ groupId> < artifactId> </ artifactId> </ plugin> </ plugins> </ build> </ project> 生产者即将发送消息：topic = my_topic1，partition: 0 ，key = ，value = spring- kafka- 事务1
生产者即将发送消息：topic = my_topic1，partition: 0 ，key = ，value = spring- kafka- 事务2
MyKafkaProducerListener消息发送失败：topic= my_topic1，partition = 0 ，key = ，value = spring- kafka- 事务1 ，offset = - 1
异常信息：Failing batch since transaction was aborted
MyKafkaProducerListener消息发送失败：topic= my_topic1，partition = 0 ，key = ，value = spring- kafka- 事务2 ，offset = - 1
异常信息：Failing batch since transaction was aborted
java. lang. ArithmeticException: / by zero
