Kafka--原理--幂等与事务

深藏阁楼爱情的钟 2022-10-13 12:55 187阅读 0赞

原文网址：[Kafka--原理--幂等与事务\_IT利刃出鞘的博客-CSDN博客\_kafka 事务原理][Kafka--_--_IT_-CSDN_kafka]

# 网址 #

[kafka的幂等性和事务性\_lianchaozhao的博客-CSDN博客][kafka_lianchaozhao_-CSDN]  
[kafka系列之幂等生产者(11)\_【CSDN官方推荐】-CSDN博客][kafka_11_CSDN_-CSDN]  
[kafka--事务和幂等(二)\_atarik@163.com-CSDN博客][kafka--_atarik_163.com-CSDN]

# 简介 #

**说明**

> 一般而言，消息中间件的消息传输保障有3个层级，分别如下。
> 
>  *  at most once：至多一次。消息可能会丢失，但绝对不会重复传输。
>  *  at least once：最少一次。消息绝不会丢失，但可能会重复传输。
>  *  exactly once：恰好一次。每条消息肯定会被传输一次且仅传输一次。
> 
> 通常情况下，为保证消息不丢失，kafka 的producer会有重试机制，但重试可能会导致消息的重复。消息重复可以在consumer端处理（幂等性处理）。Kafka 从 0.11.0.0 版本开始引入了幂等和事务这两个特性，在producer到broker这个链路就支持了幂等（exactly once）。

**Kafka在0.11.0.0新加的幂等与事务**

> 官网地址：[https://kafka.apache.org/documentation/\#upgrade\_11\_exactly\_once\_semantics][https_kafka.apache.org_documentation_upgrade_11_exactly_once_semantics]
> 
> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaXlpbmcwY2FuZ2xhbmc_size_16_color_FFFFFF_t_70][]

# 幂等 #

## 引入幂等之前 ##

在正常情况下，produce向Broker投递消息，broker将消息追加写到对应的流（即某一个topic的某一partition）中，并向Producer返回ACK信号，表示确认收到。

引入幂等性之前

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDgwOTYyNw_size_16_color_FFFFFF_t_70][]

上图的实现流程是一种理想状态下的消息发送情况，但是实际情况中，会出现各种不确定的因素，比如在Producer在发送给Broker的时候出现网络异常。比如以下这种异常情况的出现：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDgwOTYyNw_size_16_color_FFFFFF_t_70 1][]

上图这种情况，当Producer第一次发送消息给Broker时，Broker将消息(x2,y2)追加到了消息流中，但是在返回Ack信号给Producer时失败了（比如网络异常） 。此时，Producer端触发重试机制，将消息(x2,y2)重新发送给Broker，Broker接收到消息后，再次将该消息追加到消息流中，然后成功返回Ack信号给Producer。这样下来，消息流中就被重复追加了两条相同的(x2,y2)的消息。

## 引入幂等之后 ##

kafka 为实现幂等性，在底层引入了ProducerID和SequenceNumber。

*  ProducerID：在每一个新的Producer初始化时，或被分配一个唯一的ProducerID，这个ProducerID对客户端使用者是不可见的。
 *  SequenceNumber：对于每个producerID，Producer发送数据的每个Topic和Partition都对应一个从0开始递增的SequenceNumber值。

数据发送到kafka中会对数据增加Pid 和SequenceId

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDgwOTYyNw_size_16_color_FFFFFF_t_70 2][]

当ack信号返回Producer时出现网络异常，通信失败情况如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDgwOTYyNw_size_16_color_FFFFFF_t_70 3][]

当producer发送消息（x2,y2）给broker 时将其追加到消息流中，此时Broker返回Ack信号给Producer时，发生异常导致Producer接收Ack信号失败。对于Producer来说，会触发重试机制，将消息(x2,y2)再次发送。由于引入了幂等性，在每条消息中附带了PID（ProducerID）和SequenceNumber。相同的PID和SequenceNumber发送给Broker，而之前Broker缓存过之前发送的相同的消息，那么消息流中消息就只有一条（x2,y2）不会出现重复发送的情况。

其处理逻辑为：

对于收到的每一条消息，只有当它的序列号的值（SN\_new）比 broker 端中维护的对应的序列号的值（SN\_old）大1（即 SN\_new = SN\_old + 1）时，broker 才会接收它。

*  SN\_new< SN\_old + 1 说明消息被重复写入，broker 可以直接将其丢弃。
 *  SN\_new> SN\_old + 1 说明中间有数据尚未写入，出现了乱序，暗示可能有消息丢失，对应的生产者会抛出 OutOfOrderSequenceException，这个异常是一个严重的异常，后续的诸如 send()、beginTransaction()、commitTransaction() 等方法的调用都会抛出 IllegalStateException 的异常。

## 幂等的缺陷 ##

**单会话有效**

> 只能实现单会话上的幂等性，不能实现跨会话的幂等性。这里的会话，你可以理解为 Producer 进程的一次运行。当你重启了 Producer 进程之后，这种幂等性保证就丧失了。
> 
> 原因：重启之后标识producer的PID就变化了，导致broker无法根据这个<PID,TP,SEQNUM>条件去去判断是否重复。

**单分区有效**

> 只能保证单分区上的幂等性。即一个幂等性 Producer 能够保证某个主题的一个分区上不出现重复消息，它无法实现多个分区的幂等性。
> 
> 原因：在某一个partition 上判断是否重复是通过一个递增的sequence number，也就是说这个递增是针对当前特定分区的，如果你要是发送到其他分区上去了，那么递增关系就不存在了。

# 事务 #

**其他网址**

> [Kafka-事务-解决跨分区跨会话的问题 - KevinT·Mitnick - 博客园][Kafka-_-_ - KevinT_Mitnick -]  
> [kafka事务 | muggle][kafka_ _ muggle]

**简介**

> 前边说过，Kafka 的 Exactly Once 幂等性只能保证单次会话内的精准一次性，不能解决跨会话和跨分区的问题。
> 
> 事务保证Kafka在 Exactly Once 语义的基础上，Producer 和 Consumer 可以跨分区和会话，要么全部成功，要么全部失败。

## **Producer事务** ##

> 为实现跨分区跨会话的事务，需要引入一个全局唯一的 Transaction ID，并将 Producer 获得的 PID 和 Transaction ID 绑定。这样当Producer重启之后，就可以通过正在运行的 Transaction ID 获得原来的 PID。
> 
> 　　为了管理 Transaction，Kafka引入了一个新的组件 Transaction Coordinator。Producer就是通过 Transaction Coordinator 获得 Transaction ID 对应的任务状态。Transaction Coordinator还负责将事务状态写入Kafka的一个内部Topic，这样即使整个服务重启，由于事务状态得到保存，进行中的事务状态可以得到恢复。

## Consumer事务 ##

> 消费者事务的一致性比较弱，只能够保证消费者消费消息是精准一次的（有且只有一次）。消费者有一个参数 islation.level，这个参数指定的是事务的隔离级别。  
> 它的默认值是 read\_uncommitted（未提交读），意思是消费者可以消费生产者未commit的消息。当参数设置为 read\_committed，则消费者不能消费到生产者未commit的消息。

## 事务的使用场景 ##

kafka事务主要是为了保证数据的一致性，现列举如下几个场景供读者参考：

*  producer发的多条消息组成一个事务，这些消息需要对consumer同时可见或者同时不可见；
 *  producer可能会给多个topic发送消息，需要保证消息要么全部发送成功要么全部发送失败（操作的原子性）；
 *  消费者 消费一个topic，然后做处理再发到另一个topic，这个消费和转发的动作应该在同一事务中；
 *  如果下游消费者只有等上游消息事务提交以后才能读到，当吞吐量大的时候就会有问题，因此有了 read committed和read uncommitted两种事务隔离级别

[Kafka--_--_IT_-CSDN_kafka]: https://knife.blog.csdn.net/article/details/118543342
[kafka_lianchaozhao_-CSDN]: https://blog.csdn.net/weixin_40809627/article/details/106918385
[kafka_11_CSDN_-CSDN]: https://blog.csdn.net/king14bhhb/article/details/114742402
[kafka--_atarik_163.com-CSDN]: https://blog.csdn.net/asdfsadfasdfsa/article/details/104806981
[https_kafka.apache.org_documentation_upgrade_11_exactly_once_semantics]: https://kafka.apache.org/documentation/#upgrade_11_exactly_once_semantics
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaXlpbmcwY2FuZ2xhbmc_size_16_color_FFFFFF_t_70]: /images/20221005/1917211304814e8d8f0ec37132e9a058.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDgwOTYyNw_size_16_color_FFFFFF_t_70]: /images/20221005/e1c66540395c470cae07b27220494a57.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDgwOTYyNw_size_16_color_FFFFFF_t_70 1]: /images/20221005/51565600f30244c8b106e2acfdb34e15.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDgwOTYyNw_size_16_color_FFFFFF_t_70 2]: /images/20221005/3ef6fccbac8240188b38d64ebae91a26.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDgwOTYyNw_size_16_color_FFFFFF_t_70 3]: /images/20221005/6ae52f1fdba04357acc1a3b72fb27c10.png
[Kafka-_-_ - KevinT_Mitnick -]: https://www.cnblogs.com/KevinT-Mitnick/articles/13917867.html
[kafka_ _ muggle]: http://muggle.javaboy.org/2021/01/31/kafka%E4%BA%8B%E5%8A%A1/