全部标签
写点什么
OpenHarmony算法元宇宙MySQL移动开发学习方法Web3.0高效工作数据库Python音视频前端AI大数据团队管理程序员运维深度思考低代码redisgolang微服务架构flutter 查看更多

原来 kafka 也有事务啊，再也不担心消息不一致了

作者：做梦都在改BUG
  • 2023-06-06
    湖南

  • 本文字数：3384 字

    阅读完需：约 11 分钟

现在假定这么一个业务场景，从kafka中的topic获取消息数据，经过一定加工处理后，发送到另外一个topic中，要求整个过程消息不能丢失，也不能重复发送，即实现端到端的Exactly-Once精确一次消息投递。这该如何实现呢？

kafka 事务介绍

针对上面的业务场景，kafka 已经替我们想到了，在 kafka 0.11 版本以后，引入了一个重大的特性：幂等性和事务。

幂等性

这里提到幂等性的原因，主要是因为事务的启用必须要先开启幂等性，那么什么是幂等性呢？


幂等性是指生产者无论向kafka broker发送多少次重复的数据，broker 端只会持久化一条，保证数据不会重复。


幂等性通过生产者配置项enable.idempotence=true开启，默认情况下为 true。


幂等性实现原理

  1. 每条消息都有一个主键，这个主键由 <PID, Partition, SeqNumber>组成。

  • PID：ProducerID，每个生产者启动时，Kafka 都会给它分配一个 ID，ProducerID 是生产者的唯一标识，需要注意的是，Kafka 重启也会重新分配 PID。

  • Partition：消息需要发往的分区号。

  • SeqNumber：生产者，他会记录自己所发送的消息，给他们分配一个自增的 ID，这个 ID 就是 SeqNumber，是该消息的唯一标识，每发送一条消息，序列号加 1。


  1. 对于主键相同的数据，kafka 是不会重复持久化的，它只会接收一条。


幂等性缺点


根据幂等性的原理，我们发现它存在下面的缺点:

  • 只能保证单分区、单会话内的数据不重复

  • kafka 挂掉，重新给生产者分配了 PID，还是有可能产生重复的数据


那么如何实现跨分区、kafka broker 重启也能保证不重复呢？这就要使用事务了。

事务

所谓事务，就是要求保证原子性，要么全部成功，要么全部失败。那么具体该如何开启呢？

  1. kafka要想开启事务必须要启用幂等性，即生产者配置enable.idempotence=true

  2. kafka生产者需要配置唯一的事务 idtransactional.id, 最好为其设置一个有意义的名字。

  3. kafka消费端也有一个配置项isolation.level和事务有很大关系。

  • read_uncommitted：默认值，消费端应用可以看到（消费到）未提交的事务，当然对于已提交的事务也是可见的。

  • read_committed：消费端应用只能消费到提交的事务内的消息。

kafka 事务 API

现在我们用 java 的 api 来实现一下前面这个“消费-处理-生产“的例子吧。

  1. 引入依赖

<dependency>    <groupId>org.apache.kafka</groupId>    <artifactId>kafka-clients</artifactId>    <version>3.4.0</version></dependency>
复制代码

  1. 创建事务的生产者

Properties prodcuerProps = new Properties();// kafka地址prodcuerProps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");// key序列化prodcuerProps.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// value序列化prodcuerProps.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());// 启用幂等性producerProps.put("enable.idempotence", "true");// 设置事务idproducerProps.put("transactional.id", "prod-1");KafkaProducer<String, String> producer = new KafkaProducer(prodcuerProps);
复制代码

  • enable.idempotence配置项目为 true

  • 设置transactional.id

  1. 创建事务的消费者

Properties consumerProps = new Properties();consumerProps.put("bootstrap.servers", "localhost:9092");consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());consumerProps.put("group.id", "my-group-id");// 设置consumer手动提交consumerProps.put("enable.auto.commit", "false");// 设置隔离级别，读取事务已提交的消息consumerProps.put("isolation.level", "read_committed");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);//订阅主题consumer.subscribe(Collections.singletonList("topic1"));
复制代码

  • enable.auto.commit=false,设置手动提交消费者offset

  • 设置isolation.level=read_committed，消费事务已提交的消息

  1. 核心逻辑

// 初始化事务 producer.initTransactions();while(true) {	// 拉取消息 	ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000L));    if(!records.isEmpty()){        // 准备一个 hashmap 来记录："分区-消费位移" 键值对        HashMap<TopicPartition, OffsetAndMetadata> offsetsMap = new HashMap<>();        // 开启事务         producer.beginTransaction();        try {            // 获取本批消息中所有的分区            Set<TopicPartition> partitions = records.partitions();            // 遍历每个分区            for (TopicPartition partition : partitions) {                // 获取该分区的消息                List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);                // 遍历每条消息                for (ConsumerRecord<String, String> record : partitionRecords) {                    // 执行数据的业务处理逻辑                    ProducerRecord<String, String> outRecord = new ProducerRecord<>("topic2", record.key(), record.value().toUpperCase());                    // 将处理结果写入 kafka                    producer.send(outRecord);                }
                // 将处理完的本分区对应的消费位移记录到 hashmap 中                long offset = partitionRecords.get(partitionRecords.size() - 1).offset();                // 事务提交的是即将到来的偏移量，这意味着我们需要加 1                offsetsMap.put(partition,new OffsetAndMetadata(offset+1));            }            // 向事务管理器提交消费位移             producer.sendOffsetsToTransaction(offsetsMap,"groupid");            // 提交事务             producer.commitTransaction();        } catch(Exeception e) {            e.printStackTrace();            // 终止事务             producer.abortTransaction();        }    }}
复制代码

  • initTransactions(): 初始化事务

  • beginTransaction(): 开启事务

  • sendOffsetsToTransaction(): 在事务内提交已经消费的偏移量（主要用于消费者）

  • commitTransaction(): 提交事务

  • abortTransaction(): 放弃事务

kafka 事务实现原理

kafka 事务的实现引入了事务协调器，如下图所示：

  1. 生产者使用事务必须配置事务 id, kafka 根据事务 id 计算分配事务协调器

  2. 事务协调器返回 pid，前面的幂等性中需要

  3. 开始发送消息到 topic 中，不过这些消息与普通的消息不同，它们带着一个字段标识自己是事务消息

  4. 当生产者事务内的消息发送完毕，会向事务协调器发送 commit 或 abort 请求，等待 kafka 响应

  5. 事务协调器收到请求后先持久化到内置事务主题__transaction_state中，__transaction_state默认有 50 个分区，每个分区负责一部分事务。事务划分是根据transactional.id的hashcode值%50，计算出该事务属于哪个分区。 该分区Leader副本所在的 broker 节点即为这个transactional.id对应的Transaction Coordinator节点，这也是上面第一步中的计算逻辑。

  6. 事务协调器后台会跟 topic 通信，告诉它们事务是成功还是失败的。

  • 如果是成功，topic 会汇报自己已经收到消息，协调者收到主题的回应便确认了事务完成，并持久化这一结果。

  • 如果是失败的，主题会把这个事务内的消息丢弃，并汇报给协调者，协调者收到所有结果后再持久化这一信息，事务结束。

  1. 持久化第 6 步中的事务成功或者失败的信息, 如果kafka broker配置max.transaction.timeout.ms之前既不提交也不中止事务, kafka broker将中止事务本身。 此属性的默认值为 15 分钟。

总结

本文讲解了通过 kafka 事务可以实现端到端的精确一次的消息语义，通过事务机制，KAFKA 实现了对多个 topic 的多个 partition 的原子性的写入，通过一个例子了解了一下如何使用事物。同时也简单介绍了事务实现的原理，它底层必须要依赖 kafka 的幂等性机制，同时通过类似“二段提交”的方式保证事务的原子性。


作者：JAVA 旭阳

链接：https://juejin.cn/post/7241184271318122557

来源：稀土掘金

划线
评论
复制
发布于: 刚刚阅读数: 5
Javakafka事务
用户头像

做梦都在改BUG

关注

还未添加个人签名 2021-07-28 加入

公众号：该用户快成仙了

评论

发布
暂无评论
Copyright © 2023, Geekbang Technology Ltd. All rights reserved. 极客邦控股（北京）有限公司 | 京 ICP 备 16027448 号 - 5京公网安备京公网安备 11010502039052号 | 产品资质
原来kafka也有事务啊，再也不担心消息不一致了_Java_做梦都在改BUG_InfoQ写作社区