kafka教程-springboot发送消息（事务）

作者：马育民 • • 阅读：10233

介绍

kafka中的事务，相当于 关系型数据库（mysql、oracle）中的事务管理

当发送消息 有事务要求时，比如：假设第一条消息发送后，在发第二条消息前出现了异常，那么第一条已经发送的消息也会回滚。当所有消息发送成功才算成功

默认情况下，Spring-kafka自动生成的 KafkaTemplate 实例，没有事务管理，需要修改配置文件激活

注意： 事务激活后，所有的消息发送 只能在事务管理的方法内执行，否则会抛没有事务管理的异常

本文代码

需要先实现 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=localhost: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.

代码

方式一

通过 executeInTransaction() 方法实现

@GetMapping("/kafka3/{message}")
    public String sendMessage3(@PathVariable("message") 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;
                }
                kafkaTemplate.send("userlog",msg+"--2");
                return null;
            }
        });
        return "消息已发送";
    }

方式二

注解方式实现

@GetMapping("/send5/{msg}")
@Transactional(rollbackFor = Exception.class)
public String send5(@PathVariable("msg") String msg){
    //事务的支持
    kafkaTemplate.send("userlog",msg);
    if("error".equals(msg)){
        int i=1/0;
    }
    kafkaTemplate.send("userlog",msg+"--2");
    return "消息已发送";
}

测试

访问下面链接测试：

当发送消息是 error 时，将会执行抛错代码，虽然已经发了一个消息，但是后面报错，由于有事务管理，之前的消息也不会发出

注意

事务激活后，所有的消息发送 只能在事务管理的方法内执行，否则会抛没有事务管理的异常

原文出处：/show_1IXM5TfLJVZ.html

吉ICP备15003043号