kafka EOS

àì夳堔傛蜴生んèń 2022-02-26 15:58 274阅读 0赞

Kafka 0.11.0.0引入了EOS（exactly once semantics，精确一次处理语义）的特性，这个特性包括kafka幂等性和kafka事务两个属性。

1）幂等producer 保证单个分区的只会发送一次，不会出现重复消息  
2）事务(transation)：保证原子性的写入多个分区，即写入到多个分区的消息要么全部成功，要么全部回滚  
启用方法：  
1）启用幂等producer:在producer程序中设置属性enabled.idempotence=true,但不要设置transational\_id.注意是不要设置，而不是设置为空字符串  
2）启用事务支持：在producer程序中设置属性transcational.id为一个指定字符串(你可以认为这是你的额事务名称,故最好七个有意义的名字),同时设置enable.idempotence=true

1、幂等性  
1.1 引入目的

生产者重复生产消息。生产者进行retry会产生重试时，会重复产生消息。有了幂等性之后，在进行retry重试时，只会生成一个消息。

1.2 幂等性实现  
首先，producer对象引入了一个新的字段：Producer ID(下称PID)，它唯一标识一个producer，当producer启动时Kafka会为每个producer分配一个PID（64位整数），因此PID的生成和分配对用户来说是完全透明的，用户无需考虑PID的事情，甚至都感受不到PID的存在。  
其次，0.11 Kafka重构了消息格式(有兴趣的参见Kafka 0.11消息设计)，引入了序列号字段(sequence number，下称seq number)来标识某个PID producer发送的消息。和consumer端的offset类似，seq number从0开始计数并严格单调增加。同时在broker端会为每个PID(即每个producer)保存该producer发送过来的消息batch的某些元信息，比如PID信息、消息batch的起始seq number及结束seq number等。这样每当该PID发送新的消息batch时，Kafka broker就会对比这些信息，如果发生冲突(比如起始seq number和结束seq number与当前缓存的相同)，那么broker就会拒绝这次写入请求。倘若没有冲突，那么broker端就会更新这部分缓存然后再开始写入消息。  
这就是Kafka实现幂等producer的设计思路：1. 为每个producer设置唯一的PID；2. 引入seq number以及broker端seq number缓存更新机制来去重。  
2.事务

2.1 事务属性

事务属性是2017年Kafka 0.11.0.0引入的新特性。类似于数据库事务，只是这里的数据源是Kafka，kafka事务属性是指一系列的生产者生产消息和消费者提交偏移量的操作在一个事务，或者说是是一个原子操作），同时成功或者失败。

注意：在理解消息的事务时，一直处于一个错误理解就是如下代码中，把操作db的业务逻辑跟操作消息当成是一个事务。其实这个是有问题的，操作DB数据库的数据源是DB，消息数据源是kfaka，这是完全不同两个数据，一种数据源（如mysql，kafka）对应一个事务，所以它们是两个独立的事务：kafka事务指kafka一系列 生产、消费消息等操作组成一个原子操作；db事务是指操作数据库的一系列增删改操作组成一个原子操作。  
2.2 引入目的

在事务属性之前先引入了生产者幂等性，它的作用为：  
•生产者多次发送消息可以封装成一个原子操作，要么都成功，要么失败  
•consumer-transform-producer模式下，因为消费者提交偏移量出现问题，导致在重复消费消息时，生产者重复生产消息。需要将这个模式下消费者提交偏移量操作和生成者一系列生成消息的操作封装成一个原子操作。

消费者提交偏移量导致重复消费消息的场景：消费者在消费消息完成提交便宜量o2之前挂掉了（假设它最近提交的偏移量是o1），此时执行再均衡时，其它消费者会重复消费消息(o1到o2之间的消息）。  
3 幂等性和事务性的关系

3.1 两者关系

事务属性实现前提是幂等性，即在配置事务属性transaction id时，必须还得配置幂等性；但是幂等性是可以独立使用的，不需要依赖事务属性。  
•幂等性引入了Porducer ID  
•事务属性引入了Transaction Id属性。、

使用场景  
•enable.idempotence = true，transactional.id不设置：只支持幂等性。  
•enable.idempotence = true，transactional.id设置：支持事务属性和幂等性  
•enable.idempotence = false，transactional.id不设置：没有事务属性和幂等性的kafka  
•enable.idempotence = false，transactional.id设置：无法获取到PID，此时会报错

3.2 tranaction id 、productid 和 epoch

一个app有一个tid，同一个应用的不同实例PID是一样的，只是epoch的值不同。如：  
![在这里插入图片描述][20190330141023541.png]

参考：  
[https://blog.csdn.net/alex\_xfboy/article/details/82988259][https_blog.csdn.net_alex_xfboy_article_details_82988259]  
[https://blog.csdn.net/qq\_35078688/article/details/86082858][https_blog.csdn.net_qq_35078688_article_details_86082858]

[20190330141023541.png]: /images/20220226/9ab031c64ee34d35a758103a9ec6e1a3.png
[https_blog.csdn.net_alex_xfboy_article_details_82988259]: https://blog.csdn.net/alex_xfboy/article/details/82988259
[https_blog.csdn.net_qq_35078688_article_details_86082858]: https://blog.csdn.net/qq_35078688/article/details/86082858