播面

在 Kafka 中，幂等性生产者（Idempotent Producer） 是实现“精确一次（Exactly-Once）”语义的核心机制之一。

简单来说，幂等性是指：无论生产者向 Broker 发送了多少次同一条消息，Broker 最终只会将该消息持久化一次，不会产生重复数据。

一、 为什么需要幂等性生产者？

在没有幂等性的情况下，Kafka 默认的投递语义是“至少一次（At-Least-Once）”。
假设发生以下场景：

1. 生产者发送一条消息到 Broker。
2. Broker 成功将消息写入磁盘。
3. Broker 返回 ACK 给生产者时，由于网络抖动，ACK 丢失了。
4. 生产者迟迟没有收到 ACK，触发重试机制（Retries），再次发送同一条消息。
5. Broker 再次接收并写入磁盘。
   结果：同一条消息在 Kafka 中被存储了两次，导致下游消费者消费到重复数据。

为了解决这个问题，Kafka 在 0.11 版本引入了幂等性生产者。

二、 底层实现原理（PID 和 Sequence Number）

为了实现幂等性，Kafka 引入了两个核心概念：PID (Producer ID) 和 Sequence Number (序列号)。

Kafka Broker 通过维护一个映射字典 <PID, Topic, Partition> -> Sequence Number 来判断消息是否重复。

1. PID (Producer ID)

• 什么是 PID：每个生产者实例在初始化启动时，Broker 都会为其分配一个全局唯一的 ID，称为 PID。
• 特点：这个过程对用户是完全透明的。如果生产者重启，它会被分配一个新的 PID。因此，幂等性只能保证单个生产者会话（Session）内的去重。

2. Sequence Number (序列号)

• 什么是 Sequence Number：对于每个 PID，当它向某个 Topic 的某个 Partition 发送消息时，生产者会为这些消息附加一个从 0 开始单调递增的序列号。
• 粒度：序列号的作用范围是 <PID, Topic, Partition>。也就是说，同一个生产者发送到不同分区的消息，其序列号是独立计算的。

三、 幂等性的核心校验逻辑

Broker 端会在内存中为每个 <PID, Topic, Partition> 维护一个期望的下一个序列号（Expected Sequence Number），假设当前 Broker 记录的该生产者在这个分区的最新序列号为 SN_old。

当 Broker 收到一条带有新序列号 SN_new 的消息时，会进行如下判断：

1. 正常接收 (SN_new == SN_old + 1)

   • 说明这是一条正常的、按顺序到达的新消息。
   • Broker 将消息写入日志，并将记录的序列号更新为 SN_new。

2. 拒绝重复 (SN_new <= SN_old)

   • 说明这是一条已经处理过的老消息（通常是因为网络延迟导致的生产者重试）。
   • Broker 直接丢弃该消息，不进行持久化，但会向生产者返回一个 ACK 成功响应（因为生产者重试就是为了要这个 ACK）。

3. 抛出异常 (SN_new > SN_old + 1)

   • 说明中间有消息丢失了（例如序列号为 SN_old + 1 的消息没收到，直接收到了 SN_old + 2）。
   • Broker 会拒绝这条消息，并抛出 OutOfOrderSequenceException 异常。生产者收到该异常后，通常需要进行干预或触发底层重新同步。

四、 幂等性生产者的局限性

虽然幂等性解决了重试导致的重复问题，但它有严格的作用范围限制：

1. 只能保证单分区（Single Partition）：幂等性状态是针对 <PID, Topic, Partition> 维护的。如果一条消息被路由到了不同的分区，它无法保证去重。
2. 只能保证单会话（Single Session）：一旦生产者应用重启（比如挂掉后重启），它将获得一个全新的 PID。此时 Broker 将无法识别它与之前 PID 的关联，也就无法对重启前后的重试消息进行去重。

补充：为了解决跨分区、跨会话的“精确一次”问题，Kafka 引入了 事务生产者（Transactional Producer）。事务是在幂等性的基础之上构建的。

五、 如何开启幂等性？

非常简单，只需要在生产者的配置中设置：

enable.idempotence=true

注意点：

1. 开启幂等性要求 acks=all（或 -1），且 retries 必须大于 0（通常设为 MAX_VALUE），max.in.flight.requests.per.connection 必须小于等于 5。如果用户配置冲突，Kafka 启动时会报错。
2. 在 Kafka 3.0 及以上版本中，生产者的 enable.idempotence 默认值已经变更为 true（即默认开启幂等性）。