介绍
kafka中的事务，相当于 关系型数据库（mysql、oracle）中的事务管理
当发送消息 有事务要求时，比如：假设第一条消息发送后，在发第二条消息前出现了异常，那么第一条已经发送的消息也会回滚。当所有消息发送成功才算成功
默认情况下，Spring-kafka自动生成的 KafkaTemplate 实例，没有事务管理，需要修改配置文件激活
注意： 事务激活后，所有的消息发送 只能在事务管理的方法内执行，否则会抛没有事务管理的异常，如下：
java.lang.IllegalStateException: No transaction is in process; possible solutions: run the template operation within the scope of a template.executeInTransaction() operation, start a transaction with @Transactional before invoking the template method, run in a transaction started by a listener container when consuming a record
本文代码
需要先实现 kafka教程-springboot实现简单生产者 文中的代码
本文代码在文中的 生产者工程 做修改
配置文件
下面配置文件中有2处重要：
spring.kafka.producer.acks=-1开启事务，此处必须是-1，否则报错spring.kafka.producer.transaction-id-prefix=kafka_tx.事务支持
完整配置如下：
server.port=8081
###########【Kafka集群】###########
spring.kafka.bootstrap-servers=hadoop1:9092,hadoop2:9092,hadoop3:9092
###########【初始化生产者配置】###########
# 重试次数
spring.kafka.producer.retries=3
# 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
#spring.kafka.producer.acks=1
spring.kafka.producer.acks=-1 ################### 开启事务，此处必须是-1########################
# 批量大小
spring.kafka.producer.batch-size=16384
# 提交延时
spring.kafka.producer.properties.linger.ms=0
# 当生产端积累的消息达到batch-size或接收到消息linger.ms后,生产者就会将消息提交给kafka
# linger.ms为0表示每接收到一条消息就提交给kafka,这时候batch-size其实就没用了
# 生产端缓冲区大小
spring.kafka.producer.buffer-memory = 33554432
# Kafka提供的序列化和反序列化类
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.transaction-id-prefix=kafka_tx.
方式一：注解方式
@GetMapping("/kafka3/{msg}")
@Transactional(rollbackFor = Exception.class)
public String sendMessage3(@PathVariable("msg") String msg){
//事务的支持
kafkaTemplate.send("userlog",msg);
//当msg是是error时，故意报错，演示事务回滚是否生效
if("error".equals(msg)){
int i=1/0;
}
kafkaTemplate.send("userlog",msg+"--2");
return "消息已发送";
}
测试
访问下面链接测试：
当发送消息是 error 时，将会执行抛错代码，虽然已经发了一个消息，但是后面报错，由于有事务管理，之前的消息也不会发出
方式二：代码方式
通过 executeInTransaction() 方法实现
@GetMapping("/kafka4/{message}")
public String sendMessage4(@PathVariable("message") final String msg) {
// 声明事务：后面报错消息不会发出去
kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback(){
@Override
public Object doInOperations(KafkaOperations operations) {
operations.send("userlog", msg);
if("error".equals(msg)){
int i=1/0;
}
operations.send("userlog",msg+"--2");
return null;
}
});
return "消息已发送";
}
注意：启用事务后，必须通过事务发消息
启用事务后，发送消息 必须有事务管理。如果没有事务，直接发送消息，会报错
代码
/**
* 启用事务后，如果没有事务，直接发送消息，会报错
* @param msg
* @return
*/
@GetMapping("/kafka5/{message}")
public String sendMessage4(@PathVariable("message") String msg) {
kafkaTemplate.send("userlog", msg);
return "消息已发送";
}
报错如下：
java.lang.IllegalStateException: No transaction is in process; possible solutions: run the template operation within the scope of a template.executeInTransaction() operation, start a transaction with @Transactional before invoking the template method, run in a transaction started by a listener container when consuming a record