1.什么是Kafka事务

在Kafka0.11版本中，引入了Kafka事务这一概念。
Kafka事务是指，一系列Producer生产消息和Consumer消费消息在一个事务中进行，具备原子性，生产消息和Consumer偏移量同时成功或同时失败

2.Kafka事务支持的场景

  1. Producer生产消息封装在一个事务中，这些消息要么全部发送成功，要么全部发送失败

伪代码如下：

  1. public void sendInTransaction(){
  3.     //1.初始化事务
  4.     producer.initTransactions();
  6.     //2.开启事务
  7.     producer.beginTransaction();
  9.     //3.发送消息
  10.     producer.send("topic", "message");
  12.     //4.提交事务
  13.     producer.commitTransaction();   
  14. }
  1. Producer生产消息与Consumer消费消息封装在一个事务中，通常用于Kafka的流式处理

    3.Kafka的特性

    原子写

    确保同一个事务中的消息要么全部写入成功，要不全部写入失败。保证跨分区写操作的原子性
    这一点同样适用于生产消息与消费消息并提交偏移量在同一个事务中执行
    因为Consumer提交偏移量最终也是会提交到名为consumer_offsets的内置Topic中，本质也是发布消息

    读事务消息

    Consumer如果设置了isolation.level = read_committed，那么它只会读取已经提交了的消息。
    Producer提交事务后，该事务中相关消息的状态会从uncommitted变更为committed，从而实现Consumer可以拉取到已提交的消息

    拒绝僵尸进程

    本质上是基于Producer幂等性
    开启Producer幂等后，每个PID对应每个Topic的每个Partition发送的消息都会携带一个自增的Sequence Number，而Broker对于每个PID的每个Topic的每个Partition同样会维护一个自增的Sequence Number。
    可以通过对比Producer发送消息的Sequence Numer和Broker维护的Sequence Number，来确定消息的丢失，重复，乱序问题。具体可参考：Kafka一些常见问题
    而所有具有相同Transaction-id的Producer都会被分配相同的PID。因此，即使两个相同的Producer同时发布重复的消息，因为PID一样，Broker也只会持久化一条

    4.Kafka事务的原理

    Kafka为了支持事务特性，引入一个新的组件：Transaction Coordinator（TC：事务协调者）。
    主要负责分配PID，记录事务状态等操作
    详情参考：

    5.单数据源-Spring整合Kafka开启事务

    版本

    Kafka从0.11版本开始支持事务，对应Springboot版本应 >=2.1.X

    yml配置

    通过transaction-id-prefix配置开启事务机制 ```yaml

    服务器配置

    server: port: 8089

spring配置

spring: kafka: bootstrap-servers: 120.48.107.224:9092 producer: acks: -1 transaction-id-prefix: transaction-producer- key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer retries: 3 properties: enable.idempotence: true retry.backoff.ms: 100 consumer: group-id: transaction-group key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer auto-offset-reset: earliest enable-auto-commit: true

  1. <a name="Rtako"></a>
  2. ### Producer
  3. 如下，编写两个方法
  5. 1. 第一个方法正常发布消息
  6. 2. 第二个方法发布消息后代码出现运行时异常
  7. 3. Producer通过@Transactional(rollbackFor = RuntimeException.class)注解开启事务
  8. 4. 基于Spring的AOP原理，我们将发送逻辑与实际发送分开写，写在一个类里面Spring无法通过AOP来进行事务的相关控制
  9. ```java
  10. import com.ftc.transactionkafka.config.KafkaSendCallBackForString;
  11. import org.springframework.kafka.core.KafkaTemplate;
  12. import org.springframework.stereotype.Component;
  13. import org.springframework.transaction.annotation.Transactional;
  15. import javax.annotation.Resource;
  17. /**
  18.  * @author: 冯铁城 [17615007230@163.com]
  19.  * @date: 2022-09-19 18:07:06
  20.  * @describe: 生产者
  21.  */
  22. @Component
  23. @Transactional(rollbackFor = RuntimeException.class)
  24. public class Producer {
  26.     @Resource
  27.     private KafkaTemplate<String, String> kafkaTemplate;
  29.     @Resource
  30.     private KafkaSendCallBackForString kafkaSendCallBackForString;
  32.     public void sendMessageNoError() {
  33.         kafkaTemplate.send("transaction-topic", "transaction-message")
  34.                 .addCallback(kafkaSendCallBackForString);
  35.     }
  37.     public void sendMessageWithError() {
  39.         //1.发送消息
  40.         kafkaTemplate.send("transaction-topic", "transaction-message")
  41.                 .addCallback(kafkaSendCallBackForString);
  43.         //2.异常
  44.         int errorNumber = 1 / 0;
  45.     }
  46. }
  1. import com.ftc.transactionkafka.producer.Producer;
  2. import org.junit.jupiter.api.Test;
  3. import org.springframework.beans.factory.annotation.Autowired;
  4. import org.springframework.boot.test.context.SpringBootTest;
  6. @SpringBootTest
  7. class TransactionProducerTests {
  9.     @Autowired
  10.     private Producer producer;
  12.     @Test
  13.     void sendNoError() {
  14.         producer.sendMessageNoError();
  15.     }
  17.     @Test
  18.     void sendWithError() {
  19.         producer.sendMessageWithError();
  20.     }
  21. }

Consumer

Consumer无特殊代码，正常监听Topic即可

  1. import org.apache.kafka.clients.consumer.ConsumerRecord;
  2. import org.springframework.kafka.annotation.KafkaListener;
  3. import org.springframework.stereotype.Component;
  5. /**
  6.  * @author: 冯铁城 [17615007230@163.com]
  7.  * @date: 2022-09-13 10:01:26
  8.  * @describe: 消费者demo
  9.  */
  10. @Component
  11. public class Consumer {
  13.     @KafkaListener(topics = {"transaction-topic"}, groupId = "${spring.kafka.consumer.group-id}")
  14.     public void listenerDemo(ConsumerRecord<String, String> record) {
  15.         System.out.println("receive message:" + record.value());
  16.     }
  17. }

验证事务

开启事务，无异常发布消息

Producer：如下图，消息发布成功
image.png
Consumer：如下图，消息消费成功
image.png

开启事务，有异常发布消息

Producer：因为开启事务，消息发布过程中出现异常，并且异常为事务回滚异常，消息发布失败
image.png
Consumer：如下图，并未接收到任何消息
image.png

关闭事务，有异常发布消息

为了进一步验证事务，我们关闭事务，再试一次
yml配置：如下，关闭事务机制

  1. #服务器配置
  2. server:
  3.   port: 8089
  5. #spring配置
  6. spring:
  7.   kafka:
  8.     bootstrap-servers: 120.48.107.224:9092
  9.     producer:
  10.       acks: -1
  11. #      transaction-id-prefix: transaction-producer-
  12.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
  13.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
  14.       retries: 3
  15.       properties:
  16.         enable.idempotence: true
  17.         retry.backoff.ms: 100
  18.     consumer:
  19.       group-id: transaction-group
  20.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  21.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  22.       auto-offset-reset: earliest
  23.       enable-auto-commit: true

Producer：如下图，发布过程依然出现异常，但是并不是事务异常
image.png
Consumer：如下图，消息接收成功，由此可以看出，开启事务可以保证，消息接收与发布的原子性
image.png

6.多数据源-Spring整合Kafka开启事务

版本

Kafka从0.11版本开始支持事务，对应Springboot版本应 >=2.1.X

yml配置

主要通过transaction-id-prefix开启Kafka事务

  1. 主数据源开启Kafka事务
  2. 备数据源不开启Kafka事务 ```yaml

    服务器配置

    server: port: 8082

spring配置

kafka:

主数据源配置

primary: producer: bootstrap-servers: 120.48.107.224:9092 ack: “-1” key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer retries: 3 idempotence: true transaction-id-prefix: primary-transaction-producer- consumer: bootstrap-servers: 120.48.107.224:9092 group-id: multi-kafka-group-primary enable-auto-commit: true key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer auto-offset-reset: earliest poll-time: 3000

从数据源配置

secondary: producer: bootstrap-servers: 120.48.107.224:9092 ack: “-1” key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer retries: 3 idempotence: true consumer: bootstrap-servers: 120.48.107.224:9092 group-id: multi-kafka-group-secondary enable-auto-commit: true key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer auto-offset-reset: earliest poll-time: 3000

  1. <a name="TksF3"></a>
  2. ### 配置类
  3. 参考：[Springboot整合Kafka](https://www.yuque.com/u27809381/tsycgx/vgn8z2?view=doc_embed&inner=vv96V)
  4. <a name="hfqwF"></a>
  5. ### Producer
  6. 注意点如下
  8. 1. 主数据源发送消息到0分区，并且开启事务
  9. 2. 备数据源发送消息到1分区，不开启事务
  10. 3. 基于Spring的AOP原理，我们将发送逻辑与实际发送分开写，写在一个类里面Spring无法通过AOP来进行事务的相关控制
  11. ```java
  12. import com.ftc.multi.config.KafkaSendCallBackForString;
  13. import org.springframework.beans.factory.annotation.Qualifier;
  14. import org.springframework.kafka.core.KafkaTemplate;
  15. import org.springframework.stereotype.Component;
  16. import org.springframework.transaction.annotation.Transactional;
  18. import javax.annotation.Resource;
  20. /**
  21.  * @author: 冯铁城 [17615007230@163.com]
  22.  * @date: 2022-09-19 19:06:26
  23.  * @describe: 事务生产者
  24.  */
  25. @Component
  26. public class TransactionProducer {
  28.     @Resource
  29.     @Qualifier("primaryProducerTemplate")
  30.     private KafkaTemplate<String, String> primaryTemplate;
  32.     @Resource
  33.     @Qualifier("secondaryProducerTemplate")
  34.     private KafkaTemplate<String, String> secondaryTemplate;
  36.     @Resource
  37.     private KafkaSendCallBackForString kafkaSendCallBackForString;
  39.     @Transactional(rollbackFor = RuntimeException.class, transactionManager = "primaryTransactionManager")
  40.     public void sendMessageWithErrorPrimary() {
  42.         //1.发送消息
  43.         primaryTemplate.send("multi-kafka-topic", 0, "key", "primary-transaction-message")
  44.                 .addCallback(kafkaSendCallBackForString);
  46.         //2.出现异常
  47.         int errorNumber = 1 / 0;
  48.     }
  50.     public void sendMessageWithErrorSecondary() {
  52.         //1.发送消息
  53.         secondaryTemplate.send("multi-kafka-topic", 1, "key", "secondary-transaction-message")
  54.                 .addCallback(kafkaSendCallBackForString);
  56.         //2.出现异常
  57.         int errorNumber = 1 / 0;
  58.     }
  59. }
  1. package com.ftc.multi;
  3. import com.ftc.multi.producer.TransactionProducer;
  4. import lombok.SneakyThrows;
  5. import org.junit.jupiter.api.AfterEach;
  6. import org.junit.jupiter.api.Test;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.boot.test.context.SpringBootTest;
  10. import java.util.concurrent.TimeUnit;
  12. @SpringBootTest
  13. class ProducerTests {
  15.     @Autowired
  16.     private TransactionProducer transactionProducer;
  18.     @AfterEach
  19.     @SneakyThrows({InterruptedException.class})
  20.     void stopSeconds() {
  21.         TimeUnit.MILLISECONDS.sleep(500);
  22.     }
  24.     @Test
  25.     void sendWithErrorPrimary() {
  26.         transactionProducer.sendMessageWithErrorPrimary();
  27.     }
  29.     @Test
  30.     void sendWithErrorSecondary() {
  31.         transactionProducer.sendMessageWithErrorSecondary();
  32.     }
  33. }

Consumer

注意点如下

  1. 第一个消费者监听主数据源
  2. 第二个消费者监听备数据源 ```java import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.annotation.TopicPartition; import org.springframework.stereotype.Component;

/**

  • @author: 冯铁城 [17615007230@163.com]
  • @date: 2022-09-08 16:06:48

  • @describe: 多数据源监听 */ @Component public class Consumer {

    @KafkaListener(

    1.      topics = {"multi-kafka-topic"},
    2.      containerFactory = "primaryConsumerFactory",
    3.      topicPartitions = {@TopicPartition(topic = "multi-kafka-topic", partitions = {"0"})}

    ) public void primaryListener(ConsumerRecord record) {

    1.  System.out.println("主数据源_监听获取数据:" + record.value());

    }

    @KafkaListener(

    1.      topics = {"multi-kafka-topic"},
    2.      containerFactory = "secondaryConsumerFactory",
    3.      topicPartitions = {@TopicPartition(topic = "multi-kafka-topic", partitions = {"1"})}

    ) public void secondaryListener(ConsumerRecord record) {

    1.  System.out.println("备数据源_监听获取数据:" + record.value());

    } } ```

    验证事务

    主数据源发送异常

    Producer：如下图，消息发送异常，并且异常为事务回滚异常
    image.png
    Consumer：如下图，Consumer并未收到任何消息，事务成功
    image.png

    备数据源发送异常

    Producer：如下图，数据发送异常，异常为非事务异常
    image.png
    Consumer：如下图，因为备数据源未开启事务，即使数据发布过程中出现异常，Consumer依然可以接收到消息。由此可以看出，开启事务可以保证消息发布与接收的原子性
    image.png