Kafka事务使用和编程示例/实例

以你之姓@ 2021-09-25 04:38 238阅读 0赞

一、概述

Kafka事务特性是指一系列的生产者生产消息和消费者提交偏移量的操作在一个事务中，或者说是一个原子操作，生产消息和提交偏移量同时成功或者失败。

*  注意：kafka事务和DB事务。
    
    在理解消息的事务时，一直处于一个错误理解是，把操作db的业务逻辑跟操作消息当成是一个事务，如下所示：
    
        void  kakfa_in_tranction(){
          // 1.kafa的操作：读取消息或生产消息
          kafkaOperation();
          // 2.db操作
          dbOperation();
        }
    
    操作DB数据库的数据源是DB，消息数据源是kfaka，这是完全不同两个数据。一种数据源（如mysql，kafka）对应一个事务，所以它们是两个独立的事务。kafka事务指kafka一系列 生产、消费消息等操作组成一个原子操作，db事务是指操作数据库的一系列增删改操作组成一个原子操作。

二、事务的使用

Kafka中的事务特性主要用于以下两种场景：

*  生产者发送多条消息可以封装在一个事务中，形成一个原子操作。多条消息要么都发送成功，要么都发送失败。
 *  read-process-write模式：将消息消费和生产封装在一个事务中，形成一个原子操作。在一个\*\*流式处理\*\*的应用中，常常一个服务需要从上游接收消息，然后经过处理后送达到下游，这就对应着消息的消费和生成。
    
        当事务中仅仅存在Consumer消费消息的操作时，它和Consumer手动提交Offset并没有区别。因此单纯的消费消息并不是Kafka引入事务机制的原因，单纯的消费消息也没有必要存在于一个事务中。

三、事务相关的API

*  API
    
        /**
             * 初始化事务
             */
            public void initTransactions();
         
            /**
             * 开启事务
             */
            public void beginTransaction() throws ProducerFencedException ;
         
            /**
             * 在事务内提交已经消费的偏移量
             */
            public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) ;
         
            /**
             * 提交事务
             */
            public void commitTransaction() throws ProducerFencedException;
         
            /**
             * 丢弃事务
             */
            public void abortTransaction() throws ProducerFencedException ;

*  事务配置
    
     *  Producer
        
            需要设置transactional.id属性。
            设置了transactional.id属性后，enable.idempotence属性会自动设置为true。

*  Consumer
    
        需要设置isolation.level = read_committed，这样Consumer只会读取已经提交了事务的消息。另外，需要设置enable.auto.commit = false来关闭自动提交Offset功能。
    
     *

四、事务使用示例

*  需求
    
        在Kafka的topic：ods_user中有一些用户数据，数据格式如下：
        
        姓名,性别,出生日期
        张三,1,1980-10-09
        李四,0,1985-11-01
        
        我们需要编写程序，将用户的性别转换为男、女（1-男，0-女），转换后将数据写入到topic：dwd_user中。要求使用事务保障，要么消费了数据同时写入数据到 topic，提交offset。要么全部失败。

*  控制台模拟数据
    
        # 创建名为ods_user和dwd_user的主题
        bin/kafka-topics.sh --create --zookeeper node-1:2181 --topic ods_user --partitions 3 --replication-factor 2
        bin/kafka-topics.sh --create --zookeeper node-1:2181 --topic dwd_user --partitions 3 --replication-factor 2
        # 生产数据到 ods_user
        bin/kafka-console-producer.sh --broker-list node-1:9092 --topic ods_user
        # 从dwd_user消费数据
        bin/kafka-console-consumer.sh --bootstrap-server node-1:9092 --topic dwd_user --from-beginning

*  详细代码
    
        public class TransUse {
            public static void main(String[] args) {
                Consumer<String, String> consumer = createConsumer();
                Producer<String, String> producer = createProduceer();
        
                // 初始化事务
                producer.initTransactions();
        
                while(true) {
                    try {
                        // 1. 开启事务
                        producer.beginTransaction();
                        // 2. 定义Map结构，用于保存分区对应的offset
                        Map<TopicPartition, OffsetAndMetadata> offsetCommits = new HashMap<>();
                        // 2. 拉取消息
                        ConsumerRecords<String, String> records = consumer.poll(2000);
                        for (ConsumerRecord<String, String> record : records) {
                            // 3. 保存偏移量
                            offsetCommits.put(new TopicPartition(record.topic(), record.partition()),
                                    new OffsetAndMetadata(record.offset() + 1));
                            // 4. 进行转换处理
                            String[] fields = record.value().split(",");
                            fields[1] = fields[1].equalsIgnoreCase("1") ? "男":"女";
                            String message = fields[0] + "," + fields[1] + "," + fields[2];
                            // 5. 生产消息到dwd_user
                            producer.send(new ProducerRecord<>("dwd_user", message));
                        }
                        // 6. 提交偏移量到事务
                        producer.sendOffsetsToTransaction(offsetCommits, "ods_user");
                        // 7. 提交事务
                        producer.commitTransaction();
                    } catch (Exception e) {
                        // 8. 放弃事务
                        producer.abortTransaction();
                    }
                }
            }
        
        
        
            // 1. 创建消费者
            public static Consumer<String, String> createConsumer() {
                // 1. 创建Kafka消费者配置
                Properties props = new Properties();
                props.setProperty("bootstrap.servers", "node-1:9092");
                props.setProperty("group.id", "ods_user");
                props.put("isolation.level","read_committed");
                props.setProperty("enable.auto.commit", "false");
                props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
                props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        
                // 2. 创建Kafka消费者
                KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        
                // 3. 订阅要消费的主题
                consumer.subscribe(Arrays.asList("ods_user"));
        
                return consumer;
            }
        
            // 2. 创建生产者
            public static Producer<String, String> createProduceer() {
                // 1. 创建生产者配置
                Properties props = new Properties();
                props.put("bootstrap.servers", "node-1:9092");
                props.put("transactional.id", "dwd_user");
                props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
                props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        
                // 2. 创建生产者
                Producer<String, String> producer = new KafkaProducer<>(props);
                return producer;
            }
        
        }

*  异常的模拟
    
        // 3. 保存偏移量
        offsetCommits.put(new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1));
        // 4. 进行转换处理
        String[] fields = record.value().split(",");
        fields[1] = fields[1].equalsIgnoreCase("1") ? "男":"女";
        String message = fields[0] + "," + fields[1] + "," + fields[2];
        
        // 模拟异常
        int i = 1/0;
        
        // 5. 生产消息到dwd_user
        producer.send(new ProducerRecord<>("dwd_user", message));
    
    我们发现，可以消费到消息，但如果中间出现异常的话，offset是不会被提交的，除非消费、生产消息都成功，才会提交事务。

转自：

[https://blog.csdn.net/qq\_36382679/article/details/108466254][https_blog.csdn.net_qq_36382679_article_details_108466254]

\-------------------------------------------

## 一、事务场景 ##

1.  最简单的需求是producer发的多条消息组成一个事务这些消息需要对consumer同时可见或者同时不可见 。
2.  producer可能会给多个topic，多个partition发消息，这些消息也需要能放在一个事务里面，这就形成了一个典型的分布式事务。
3.  **kafka的应用场景经常是应用先消费一个topic，然后做处理再发到另一个topic，这个consume-transform-produce过程需要放到一个事务里面，比如在消息处理或者发送的过程中如果失败了，消费位点也不能提交。**
4.  producer或者producer所在的应用可能会挂掉，新的producer启动以后需要知道怎么处理之前未完成的事务 。
5.  流式处理的拓扑可能会比较深，如果下游只有等上游消息事务提交以后才能读到，可能会导致rt非常长吞吐量也随之下降很多，所以需要实现read committed和read uncommitted两种事务隔离级别。

## 二、几个关键概念和推导 ##

1.因为producer发送消息可能是分布式事务，所以引入了常用的2PC，所以有事务协调者(Transaction Coordinator)。Transaction Coordinator和之前为了解决脑裂和惊群问题引入的Group Coordinator在选举和failover上面类似。

2.事务管理中事务日志是必不可少的，kafka使用一个内部topic来保存事务日志，这个设计和之前使用内部topic保存位点的设计保持一致。事务日志是Transaction Coordinator管理的状态的持久化，因为不需要回溯事务的历史状态，所以事务日志只用保存最近的事务状态。  
3.因为事务存在commit和abort两种操作，而客户端又有read committed和read uncommitted两种隔离级别，所以消息队列必须能标识事务状态，这个被称作Control Message。  
4.producer挂掉重启或者漂移到其它机器需要能关联的之前的未完成事务所以需要有一个唯一标识符来进行关联，这个就是TransactionalId，一个producer挂了，另一个有相同TransactionalId的producer能够接着处理这个事务未完成的状态。注意不要把TransactionalId和数据库事务中常见的transaction id搞混了，kafka目前没有引入全局序，所以也没有transaction id，这个TransactionalId是用户提前配置的。  
5. TransactionalId能关联producer，也需要避免两个使用相同TransactionalId的producer同时存在，所以引入了producer epoch来保证对应一个TransactionalId只有一个活跃的producer epoch

## 三、事务语义 ##

### 2.1.  多分区原子写入 ###

事务能够保证Kafka topic下每个分区的原子写入。事务中所有的消息都将被成功写入或者丢弃。例如，处理过程中发生了异常并导致事务终止，这种情况下，事务中的消息都不会被Consumer读取。现在我们来看下Kafka是如何实现原子的“读取-处理-写入”过程的。

首先，我们来考虑一下原子“读取-处理-写入”周期是什么意思。简而言之，这意味着如果某个应用程序在某个topic tp0的偏移量X处读取到了消息A，并且在对消息A进行了一些处理（如B = F（A））之后将消息B写入topic tp1，则只有当消息A和B被认为被成功地消费并一起发布，或者完全不发布时，整个读取过程写入操作是原子的。

**现在，只有当消息A的偏移量X被标记为消耗时，消息A才被认为是从topic tp0消耗的，消费到的数据偏移量（record offset）将被标记为提交偏移量（Committing offset）。在Kafka中，我们通过写入一个名为**\_\_consumer\_offsets **topic的内部Kafka topic来记录offset commit。消息仅在其offset被提交给**\_\_consumer\_offsets **topic时才被认为成功消费。**

由于offset commit只是对Kafkatopic的另一次写入，并且由于消息仅在提交偏移量时被视为成功消费，所以跨多个主题和分区的原子写入也启用原子“读取-处理-写入”循环：提交偏移量X到offset topic和消息B到tp1的写入将是单个事务的一部分，所以整个步骤都是原子的。

[https://www.cnblogs.com/wangzhuxing/p/10125437.html][https_www.cnblogs.com_wangzhuxing_p_10125437.html]

[https_blog.csdn.net_qq_36382679_article_details_108466254]: https://blog.csdn.net/qq_36382679/article/details/108466254
[https_www.cnblogs.com_wangzhuxing_p_10125437.html]: https://www.cnblogs.com/wangzhuxing/p/10125437.html