做后端和中间件这几年最头疼的问题之一就是“消息不重不漏”。尤其当你处理的账务数据、商品库存、用户积分直接和真金白银挂钩时一条消息重复消费和丢失消费后果截然不同轻则数据对不上重则出现资损事故。Kafka 的 Exactly-Once 语义可以说是这套机制的核心答案但它的实现方式并不像字面上那么“一键开启”。很多同学对 Kafka Exactly-Once 的印象可能停留在“把 enable.idempotence 打开再配上事务就完事大吉了”。真上了生产环境你会发现事情远没那么简单。这篇就结合我自己的实操经验把幂等生产者、事务、消费隔离级别、端到端不重不漏的落地方法一点点拆开讲清楚顺便把那些容易踩的坑给你捋一遍。适合正在维护消息中间件、做实时数据链路或者想搞明白 Kafka 底层原理的朋友参考。1. 先从“消息为什么会重、会丢”说起1.1 重复和丢失是消息系统的原罪要理解 Kafka 怎么保证“不重不漏”先得搞清楚消息为什么会重复、会丢失。本质上这是分布式系统网络不确定性带来的副作用。最常见的重复场景是网络超时重试。生产者把消息发到 brokerbroker 其实已经写入成功了但 ack 响应在网络上超时了。生产者一看没收到确认按策略发起重试于是同一条消息又被写了一次。另一个高频场景在消费端消费者拉取了一批消息本地业务处理完了还没来得及提交 offset触发了一次消费者组 rebalance分区被重新分配这批消息又会被拉起来处理一遍。丢失场景也毫不罕见。Kafka 写入消息时先落 page cache异步刷盘如果 broker 在数据刷到磁盘之前宕机这部分消息就可能永久丢失。再比如生产者没有开启 acksall 并且副本数不够leader 挂了之后未同步的数据也会丢。这三种语义的关系可以这样理解语义含义代价At-Most-Once最多一次消息可能丢但绝不会重复可能丢数据At-Least-Once至少一次消息不会丢但可能重复可能重复数据Exactly-Once精确一次消息不丢也不重需要复杂机制配合Kafka 早期版本默认提供的是 At-Least-Once也就是保证不丢但重复是允许的。直到 0.11 版本之后才真正提供了幂等生产者和事务机制把 Exactly-Once 从理想拉进了现实。1.2 Exactly-Once 的本质不丢加上不重有一个关键认知我建议你先刻在脑子里Exactly-Once 并不是一种独立的、凭空产生的新协议它的本质是“At-Least-Once 保证不丢”和“幂等机制保证不重”的组合。这个道理放到生活里特别好理解。你网购了一个包裹快递员的投递策略分三种。第一种把包裹放到门口就走丢了就丢了这叫最多一次。第二种快递员一定要看着你签收才走如果不小心多敲了一次门你可能会收到两个一样的包裹这叫至少一次。第三种快递员送一次然后系统会自动识别“这个订单已经签收过了”即使网络重复通知了快递员你也绝不会收到第二个包裹这才是精确一次。Kafka 走的是同样的路生产端通过序列号和事务让同一条记录不会被 broker 重复写入消费端通过幂等设计和事务性提交保证同一个业务操作不会被重复执行。丢的问题靠副本机制和 acks 配置兜底然后再在这个基础上把“重”的问题解决掉。1.3 Kafka 官方“Exactly-Once”的准确范围这部分必须泼一盆冷水。Kafka 官方所说的 Exactly-Once准确含义是在 Kafka 内部、通过 Kafka 事务实现的处理一次。具体包含三层能力生产者到 broker 的消息写入不重复依靠幂等生产者跨分区、跨 topic 的多条消息可以原子性提交要么全部消费者可见要么全部不可见配合消费端的 read_committed 隔离级别不会读取到未提交或已中止事务的消息。但要注意它并不承诺“你的业务数据库只被写入一次”。如果你的流程是“从 Kafka 消费消息然后调用外部接口写自己的业务库”Kafka 的手再长也伸不到外部系统里。这部分要做到不重不漏必须由应用层自己设计。后文会专门拆解。2. 第一道防线幂等生产者解决“写重复”2.1 没开幂等时你到底会遇到什么我最早排查过一个问题某个订单系统往 Kafka 里发消息下游消费后发现同一笔订单的流水出现了两条两条的 offset 还不一样。生产端日志显示那段时间网络有抖动producer 重试了好几次而且每次都报“success”。原因就是第一次请求已经到达 broker 并写入成功但 ack 回来时超时生产者误以为失败又发了一次。这类重复有一个共同特征生产者的重试次数越多、网络抖动越频繁重复率越高。你从下游数落地的消息条数和上游实际的业务量对比能明显看出多余的记录。2.2 幂等机制的核心原理PID 序列号Kafka 的幂等生产者解决的就是这个问题。它引入三个关键概念Producer IDPID每次 producer 初始化时会向 broker 申请一个全局唯一的 PID这个值对客户端代码不可见。正常情况下它就是生产者身份的代表。Sequence Number序列号在同一个 PID 内发往每个分区TopicPartition的消息都维护一个从 0 开始、单调递增的序号。Broker 端去重校验broker 会为每个 (PID, TopicPartition) 缓存最新的序列号。收到消息时检查序号如果当前序号等于“上一序号 1”正常写入如果当前序号小于等于已经收到的最大序号说明这是一条重试的旧消息直接丢弃并返回成功让生产者以为这次写入成功了如果当前序号远大于预期说明中间有消息丢了返回 OutOfOrderSequenceException。这套机制你可以简单理解成 TCP 的滑动窗口协议只不过 Kafka 把它简化成了严格的单序号递增判断。你不需要在业务里写任何序列号相关代码一切都发生在协议层面。2.3 配置要点和版本注意事项用起来其实很简单核心就一个参数Properties props new Properties(); props.put(bootstrap.servers, broker1:9092,broker2:9092); props.put(key.serializer, org.apache.kafka.common.serialization.StringSerializer); props.put(value.serializer, org.apache.kafka.common.serialization.StringSerializer); props.put(enable.idempotence, true);但开启这个配置之后producer 内部会有一系列隐式修改新手特别容易踩坑acks 会被自动调整为 all也就是说生产者必须等所有 ISR 副本都写入成功才算成功这样才不会在副本层面丢数据retries 会变得非常大默认允许长时间重试max.in.flight.requests.per.connection 会被限制到 5 以内这个限制是为了避免过多未确认的请求导致序列号跨度过大。所以如果你在配置里手动写了 acks0 或者 acks1再开 enable.idempotencetrue那是自相矛盾的启动时就会报错或者被忽略。版本方面也有要求broker 和客户端都必须处于 0.11 以上。如果你的集群混合了老版本的 broker幂等在这个分区上可能无法真正生效。实践上我建议客户端版本尽量跟集群版本保持一致至少大版本不要低于集群。2.4 幂等生产者的边界别指望它解决所有重复很多人的误区是开了幂等就一劳永逸。事实上幂等只保证“同一个 PID 生命周期内、发往同一个分区的消息不重复”。有两类情况它管不到第一生产者进程重启。每次 new KafkaProducer 都会生成新的 PID新旧 PID 之间没有任何关联broker 不认识你之前写过的序列号。假设有一条消息在旧进程里已经发送成功但业务以为失败重启后重新发送这次拿着新的 PID 再写一遍broker 会觉得这是一个全新生产者照写不误。第二跨分区原子性。幂等生产者对单个分区有效但它做不到“两个分区一起写成功或者一起不写”。如果你的业务需要把一条订单结果同时写进两个 topic写完第一个之后进程崩溃第二个没写下游看到的两个 topic 状态是不一致的。这两个场景的逻辑延伸就是 Kafka 事务。幂等是事务的一个底层基础但不是全部。3. 进阶Kafka 事务解决“跨分区原子写”3.1 为什么幂等还不够我举个具体例子。某电商系统的支付完成回调触发一条消息消费者处理逻辑是先向积分 topic 写入一条“增加积分”的事件再向通知 topic 写入一条“发送推送”的事件。如果积分写成功、通知写失败用户会发现自己积分多了但没收到推送对账和排查都会非常痛苦。更麻烦的是消费 offset 的处理。消费原消息的 offset 需要提交产出合并结果需要原子性这两个动作如果不在同一个事务里就会出现“数据处理了但 offset 没提交”或“offset 提交了数据却写了一半”的尴尬。Kafka 事务就是为了把这多个动作绑成一个原子单元。3.2 事务核心组件从协调器到控制标记Kafka 事务涉及几个重要角色理解它们对排查问题很有帮助。TransactionalId事务 ID客户端配置的一个字符串相当于逻辑事务生产者的身份。它最主要的作用是让生产者重启之后可以找回之前事务的状态也用于“fencing”掉僵尸生产者。Transaction Coordinator事务协调器。每个 TransactionalId 经过哈希会分配到一个协调器它负责推进事务状态、写入事务日志、以及向参与事务的分区写控制标记。__transaction_state 主题内部主题持久化事务的元数据和状态。可以类比消费组位移用的 __consumer_offsets 主题。Control Batch 控制消息协调器在事务提交或中止时会在所有参与事务的分区上写入一条特殊的标记消息COMMIT 或 ABORT下游消费者就是靠读到这个标记来决定事务内消息是否可见。状态机事务状态大致经历 Empty - Ongoing - PrepareCommit - CompleteCommit 或者 PrepareAbort - CompleteAbort。你不需要记全只需要知道对外只有“提交成功”和“中止”两个状态。执行一个事务时生产者必须先向协调器发起 initTransactions协调器会给它分配 PID并且做好恢复逻辑。之后 producer 发送的每一条消息协调器都会跟踪它们涉及哪些分区并在 prepareCommit 阶段向所有分区写 marker。3.3 生产者端事务代码示例直接看代码比背概念直观得多ProducerString, String producer new KafkaProducer(props); // 初始化事务内部会注册 transactionalId 并拿到 PID producer.initTransactions(); try { // 开启一个事务 producer.beginTransaction(); producer.send(new ProducerRecord(points-topic, userId, points-event)); producer.send(new ProducerRecord(notify-topic, userId, push-event)); // 如果希望把消费到的 offset 一起提交到事务里配合消费者使用 producer.sendOffsetsToTransaction(offsetMap, consumerGroupId); producer.commitTransaction(); } catch (Exception e) { // 回滚整个事务 producer.abortTransaction(); throw e; }这里有几个实际开发中的硬性要求同一个 TransactionalId 不能被多个生产者实例同时使用。一旦两个实例都发起 initTransactions后一个会把前一个的 producer epoch 拉高前一个再发送时就会收到 ProducerFencedException。在实际部署中这个 id 必须是全局唯一的你可以按“业务名 实例编号”的方式生成。事务期间不要做耗时的 RPC 调用。比如在 beginTransaction 之后却去调用一个 30 秒才会返回的外部接口会让事务悬挂很久最终触发 transaction.timeout.ms 超时回滚。sendOffsetsToTransaction 是“消费-处理-提交”模型的关键它让你在当前事务里把消费到的分区位移也一并提交从而让“处理结果”和“消费位移”保持原子。3.4 消费端隔离级别read_committed 与 read_uncommitted事务能不能对你生效还要看消费者的隔离级别设置。默认情况下Kafka 消费者是 read_uncommitted意思是所有写入的消息它都能看到不管事务是提交了还是中途中止了。这会带来一个问题如果下游在事务还没 commit 时就读取到了这组消息碰巧事务又因为异常回滚了这条消息实际上不算数你却已经处理了。想让事务真正有隔离语义消费者要做一点调整props.put(isolation.level, read_committed);设置成 read_committed 之后消费者遇到一个事务的第一条未提交消息时会暂停返回该分区的数据直到读到 commit marker才开始向业务代码返回这批消息如果读到的是 abort marker中间那部分数据会被直接丢弃。这就保证了“要么读不到要么读到完整已提交的内容”。代价是“读延迟”。如果某个事务长时间不提交read_committed 消费者会一直等着你观察到的现象就是该分区消费的 lag 在涨但业务消费进度一动不动。很多线上事故就是这个原因导致的下游超时。3.5 事务到底保证了什么不保证什么总结一下Kafka 事务做到的跨分区原子写入让 read_committed 消费者不会读到未提交和已中止的消息事务提交和 offset 提交成为一个原子动作。它没做到的事务提交之后生产者进程如果再尝试对同一个事务做写入会被拦截但如果是另一个完全无关的业务操作它管不到。更重要的它管不到你的外部系统比如事务结果还要调第三方支付、写 MySQL、发短信这些不在 Kafka 的参与范围内。要理解事务的边界就是 Kafka 集群边界。凡是跨出 Kafka 的动作都需要应用层再想办法。4. 端到端实战真正把“不重不漏”落到业务里4.1 经典消费-处理-写库循环的风险点大家最常见的模型是这样的while (true) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(100)); for (ConsumerRecordString, String record : records) { // 1. 解析业务消息 // 2. 调用业务服务 / 写数据库 // 3. 更新缓存 } consumer.commitSync(); // 最后提交 offset }这个流程的风险一眼就能看出来如果第二步执行成功第三步 offset 提交失败下一次 rebalance 或重启后这批消息又会被重新消费。如果你的第二步不是幂等操作那么一次重复消费就会变成一次重复扣款、重复加积分。这时候即使 Kafka 生产者那边再怎么开幂等也救不了你。所以真正要做到端到端不重不漏Kafka 只是其中一半应用层必须自己设计另一半。4.2 方案一目标存储层做幂等兜底这是最实用、也是我推荐最先做的一道防线。它的逻辑很朴素把业务消息里的唯一标识订单号、流水号、事件 ID作为目标数据库的唯一索引或主键。重复消费时第二次插入会被数据库拒绝或者覆盖成相同的结果。比如做账单流水同步表设计可以这样CREATE TABLE t_bill_flow ( id VARCHAR(64) PRIMARY KEY, account_id VARCHAR(64) NOT NULL, amount DECIMAL(10, 2) NOT NULL, create_time DATETIME, UNIQUE KEY uk_event_id (id) );消费时执行 insert ignore如果影响行数为 0说明这条事件已经处理过直接跳过。用这个办法就算消费端重复拉取了消息最终表里也只有一条记录。如果业务是更新操作而不是插入比如用户余额变更可以用版本号做乐观锁UPDATE t_account SET balance balance - #{amount}, version version 1 WHERE account_id #{accountId} AND version #{oldVersion};如果更新影响行数为 0说明版本已经变了这条消息大概率是重复或者乱序需要结合具体状态判断。这类“目标存储幂等”的思路比在任何地方加分布式锁都靠谱因为数据库的唯一约束是天然的全局去重器。4.3 方案二本地事务表与 Kafka offset 形成一个假边界还有一种经典做法是借助数据库事务来伪造一个跨系统的“事务边界”。思路是这样的消费消息后不要立即直接处理业务先在一个数据库事务里写入业务记录同时在同一个事务里往一张消费进度表写入“(consumer_group, topic, partition, offset)”的对应关系两个写操作绑定在同一个本地事务里一起 commit数据库事务提交成功之后再向 Kafka 提交 offset。如果 Kafka offset 提交失败没关系下一次重复消费时应用先去查询消费进度表发现这条 offset 已经处理过了直接跳过。这个方案的本质是让“消息是否处理过”这个事实持久化到数据库用数据库事务保证业务和进度记录同生共死。它不是真正的分布式事务但配合业务幂等键能在绝大多数场景下达到“最终不重不漏”。4.4 方案三链路全在 Kafka 内时用 Kafka Streams EOS如果你的数据处理逻辑完全在 Kafka 内部完成比如“消费源订单 topic做聚合或过滤写回结果 topic”那么可以用 Kafka Streams 的 Exactly-Once guarantee。简单来说Streams 库把“读取消息、更新内部状态存储、写结果消息、提交 offset”这几步封装成了同一个事务。你只需要在配置里写props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, exactly_once);Streams 会自动管理好事务边界这对做实时 ETL、聚合计算、数据清洗的人来说非常省心。但要注意一旦处理过程中掺进外部 RPC比如查数据库、调接口这段外部逻辑依然不受保护需要回到前两种方案。4.5 不重不漏的四个自检问题我在上线前会拿下面这份问题清单逐条问自己的链路生产端有没有开启幂等如果生产者会经常重启旧的 PID 会不会造成重启后的重复写消费端能不能接受重复如果不能目标表有没有全局唯一键更新操作有没有版本校验是否需要跨分区原子写如果需要事务和 read_committed 配好了吗有没有多个实例并发消费同一分区并发场景下去重必须去存储层做不要只放在进程内存里。5. 常见问题与疑难排查实录5.1 开启了幂等线上还是偶发重复这个问题我问过很多人最后定位到的原因高度一致生产端应用发生过重启或者它内部因为某种原因重新创建了 KafkaProducer 实例。前面讲过new KafkaProducer 会拿到新的 PID旧 PID 写入的数据和新 PID 之间没有序列号继承关系broker 无法去重。还有一种常见情况是消息从两个不同的 topic 写入但下游只按某个业务表字段去重。幂等只保证单个 topic 的写入不重复不保证跨 topic 的业务一致性这个时候需要借助事务或落库唯一键去兜底。排查思路看 producer 日志里 PID 初始化记录和重试日志再在消息里找一个业务唯一 ID 去统计存在多条重复记录的概率。如果是消息本身出现了不同 offset 的副本大概率是生产侧重启或重试导致需要在消费侧加去重字段。5.2 ProducerFencedException 频繁出现这个异常在开启事务后很常见原因通常是同一个 TransactionalId 被多个生产者实例同时使用。比如某服务原来单实例部署后来扩了多个副本但配置里 TransactionalId 还写死了同一个字符串于是一个实例 initTransactions 时另一个实例就被“隔离”了所有写入都报 ProducerFencedException。解决方案就是确保 TransactionalId 全局唯一。部署多实例时最好保证每个实例处理不同的分区TransactionalId 按“业务主题 实例编号”构造。同时检查一下是不是有旧进程没有正常退出和新的实例抢 ID。5.3 read_committed 消费端出现明显延迟如果你的消费者设置成了 read_committed某一天发现某个分区的 lag 飙升业务一直在空转很可能是遇到了一个“悬挂事务”。当某个生产者发起事务但迟迟不 commit或者在没有 commit 的情况下崩溃了事务协调器要等到 transaction.timeout.ms默认 60 秒左右具体看版本超时后才会把它回滚read_committed 消费者在这段时间里会一直等待 marker不能消费这个分区。排查时可以看一下生产者侧的事务相关指标以及集群里活跃事务的状态。更直接的治理手段是事务内绝不放大耗时操作并对正常的业务事务设置合理的 transaction.timeout.ms避免一个异常事务长期占用消费进度。5.4 offset 提交成功了还是反复消费这种场景多见于多线程消费模型。主线程拉取了 500 条消息分发给多个子线程并行处理offset 提交却是在主线程里做的。子线程处理较慢rebalance 发生还没处理完的消息就丢了但 offset 已经提交了于是这批消息既没有被完整处理也不会再被拉取。这类问题本质上是回调顺序和消费模型不匹配建议要么手动确认全部完成再提交 offset要么把串行依赖的部分收敛成单线程避免“处理一半就提交成功”的错觉。5.5 线上如何确认消息没有重复最后分享一个验证手段。在消息结构里预留一个全局唯一的 requestId 或 eventId 字段消费端在处理前把这批 id 写入一个幂等表或者布隆过滤器。上线前做一轮重放测试人为重启消费者、触发 rebalance、模拟生产端重试再统计幂等表中是否出现重复事件。只要底层在这个层面能保证单事件原子处理线上即使有一些不可控的重试也不会造成真实重复。我个人的体感是Kafka 的 Exactly-Once 不是“一键开启”的功能而是一套组合拳。生产侧用幂等和事务解决写入重复消费侧用隔离级别和幂等设计解决处理重复最后再靠目标库的唯一索引守住底线。很多团队以为事务一开就万事大吉查到最后往往发现最扎实的还是那条数据库唯一约束它有最后一道闸门的作用。链路里每多一层设计重复的风险就少一分但前提是你理解每层的边界——Kafka 管不到外部世界能兜底的是你自己的幂等设计。