Kafka事务消息-整合SpringBoot

Kafka事务消息-整合SpringBoot

Kafka 从 0.11 版本开始引入了事务支持

        事务可以保证对多个分区写入 *** 作的原子性
         *** 作的原子性是指多个 *** 作要么全部成功，要么全部失败，不存在部分成功、部分失败的可能

        关于多条消息发送需要原子性的事务 *** 作，需要配置spring.kafka.producer.transaction-id-prefix来开启事务(否则即使抛出异常消息还是会被发出去)，但需要注意的是，如果开启事务，那么之后就必须使用事务方式kafkaTemplate.executeInTransaction(声明式事务)或@Transactional(注解事务)，否则会报错。

增加配置:

spring.kafka.producer.transaction-id-prefix = wnn-tran-

注解方式的事务：

    @GetMapping("/api/v1/tran1")
    @Transactional(rollbackFor = RuntimeException.class)
    public void sendMessage2(int num){

        kafkaTemplate.send(TOPIC_NAME,"这个是事务消息 1 i="+num);

        if(num == 0){
            throw new RuntimeException();
        }
        kafkaTemplate.send(TOPIC_NAME,"这个是事务消息 2 i="+num);

    }

 http://localhost:8080/api/v1/tran1?num=0

当num输入0时，报异常。事务消息一条都不会发送。

 

声明式事务：

    
    @GetMapping("/api/v1/tran2")
    public void sendMessage3( int num){

        kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback() {
            @Override
            public Object doInOperations(KafkaOperations kafkaOperations) {

                kafkaOperations.send(TOPIC_NAME,"这个是事务消息 1 i="+num);

                if(num == 0){
                    throw new RuntimeException();
                }
                kafkaOperations.send(TOPIC_NAME,"这个是事务消息 2 i="+num);
                return true;
            }
        });
    }

http://localhost:8080/api/v1/tran2?num=0

当num输入0时，报异常。事务消息一条都不会发送。