【Kafka】Exactly Once语义与事务

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

Kafka在0.11.0.0之前的版本中只支持 At Least Once 和 At Most Once 语义，尚不支持 Exactly Once 语义。

但是在很多要求严格的场景下，如使用Kafka处理交易数据， Exactly Once 语义是必须的。我们可以通过让下游系统具有幂等性来配合Kafka的 At Least Once 语义来间接实现 Exactly Once 。但是：

该方案要求下游系统支持幂等操作，限制了Kafka的适用场景
实现门槛相对较高，需要用户对Kafka的工作机制非常了解
对于Kafka Stream而言，Kafka本身即是自己的下游系统，但Kafka在0.11.0.0版本之前不具有幂等发送能力

因此，Kafka本身对 Exactly Once 语义的支持就非常必要。

操作原子性

操作的原子性是指，多个操作要么全部成功要么全部失败，不存在部分成功部分失败的可能。

实现原子性操作的意义在于：

操作结果更可控，有助于提升数据一致性
便于故障恢复。因为操作是原子的，从故障中恢复时只需要重试该操作（如果原操作失败）或者直接跳过该操作（如果原操作成功），而不需要记录中间状态，更不需要针对中间状态作特殊处理

实现事务机制的几个阶段

幂等性发送

上文提到，实现 Exactly Once 的一种方法是让下游系统具有幂等处理特性，而在Kafka Stream中，Kafka Producer本身就是“下游”系统，因此如果能让Producer具有幂等处理特性，那就可以让Kafka Stream在一定程度上支持 Exactly once 语义。

为了实现Producer的幂等语义，Kafka引入了 Producer ID （即 PID ）和 Sequence Number 。每个新的Producer在初始化的时候会被分配一个唯一的PID，该PID对用户完全透明而不会暴露给用户。

对于每个PID，该Producer发送数据的每个 <Topic, Partition> 都对应一个从0开始单调递增的 Sequence Number 。

类似地，Broker端也会为每个 <PID, Topic, Partition> 维护一个序号，并且每次Commit一条消息时将其对应序号递增。对于接收的每条消息，如果其序号比Broker维护的序号（即最后一次Commit的消息的序号）大一，则Broker会接受它，否则将其丢弃：

如果消息序号比Broker维护的序号大一以上，说明中间有数据尚未写入，也即乱序，此时Broker拒绝该消息，Producer抛出 InvalidSequenceNumber
如果消息序号小于等于Broker维护的序号，说明该消息已被保存，即为重复消息，Broker直接丢弃该消息，Producer抛出 DuplicateSequenceNumber

上述设计解决了0.11.0.0之前版本中的两个问题：

Broker保存消息后，发送ACK前宕机，Producer认为消息未发送成功并重试，造成数据重复
前一条消息发送失败，后一条消息发送成功，前一条消息重试后成功，造成数据乱序

事务性保证

上述幂等设计只能保证单个Producer对于同一个 <Topic, Partition> 的 Exactly Once 语义。

另外，它并不能保证写操作的原子性——即多个写操作，要么全部被Commit要么全部不被Commit。

更不能保证多个读写操作的的原子性。尤其对于Kafka Stream应用而言，典型的操作即是从某个Topic消费数据，经过一系列转换后写回另一个Topic，保证从源Topic的读取与向目标Topic的写入的原子性有助于从故障中恢复。

事务保证可使得应用程序将生产数据和消费数据当作一个原子单元来处理，要么全部成功，要么全部失败，即使该生产或消费跨多个 <Topic, Partition> 。

另外，有状态的应用也可以保证重启后从断点处继续处理，也即事务恢复。

为了实现这种效果，应用程序必须提供一个稳定的（重启后不变）唯一的ID，也即 Transaction ID 。 Transactin ID 与 PID 可能一一对应。区别在于 Transaction ID 由用户提供，而 PID 是内部的实现对用户透明。

另外，为了保证新的Producer启动后，旧的具有相同 Transaction ID 的Producer即失效，每次Producer通过 Transaction ID 拿到PID的同时，还会获取一个单调递增的epoch。由于旧的Producer的epoch比新Producer的epoch小，Kafka可以很容易识别出该Producer是老的Producer并拒绝其请求。

有了 Transaction ID 后，Kafka可保证：

跨Session的数据幂等发送。当具有相同 Transaction ID 的新的Producer实例被创建且工作时，旧的且拥有相同 Transaction ID 的Producer将不再工作。
跨Session的事务恢复。如果某个应用实例宕机，新的实例可以保证任何未完成的旧的事务要么Commit要么Abort，使得新实例从一个正常状态开始工作。