Kafka消息不重复消费的底层原理与幂等实践

做流数据开发的人，基本都会被问过这么一个问题：Kafka怎么保证消费只消费一次？我先说结论，别急着去找一个“万能开关”——Kafka确实提供了exactly-once语义的能力，但在绝大多数真实业务里，真正决定“只消费一次”的是你的消费者程序怎么写，而不是改几个集群参数就完事。

这个问题的完整答案，是一条从生产端到消费端的链路设计：幂等生产者、事务机制、手动提交offset、下游幂等，缺一环都可能出问题。这篇文章我会把每一层都拆开讲，从原理到代码，再到排查经验，尽量用我踩过的坑来帮你减少试错成本。

1. 先把问题说清楚：Kafka到底能不能“恰好一次”

1.1 三种投递语义，别把“不丢”当成“不重”

在讨论Kafka之前，先明确分布式消息系统里三个基础概念：at-most-once、at-least-once、exactly-once。这三者的差别非常关键，很多人争论半天，其实是连基础术语都没有对齐。

Kafka的broker存储消息时，本身是不会有“重复”的。Offset记录了消费者读到了哪个位置，可以理解成书签。消费者每消费一批消息，就应该推进一次书签；但如果书签还没来得及推进，程序就挂了，重启之后消费者会从旧的书签位置重新读一遍，消息自然就重复了。

所以Kafka默认情况下表现出的行为通常是at-least-once，不是exactly-once。这是由offset提交机制直接决定的，跟broker性能、网络快慢没有关系。明白了这一点，你才不会天真地指望“Kafka消息不发重”。

1.2 重复消费到底从哪来：四个高频触发点

我见过太多人把重复消费归咎于“Kafka不稳定”，但绝大多数重复其实发生在消费者自身。我梳理了一下，最高频的触发点有四个：

第一个触发点：消息处理完了，但offset没有提交。比如poll了一批消息，业务代码全部执行成功，就在你准备提交offset的那一瞬间，进程被kill了，或者消费者所在的机器断电了。重启后Kafka发现消费组的上一次提交还在老位置，于是把这批消息又重新发给消费者。这个时候，业务侧已经“处理成功”过一次，第二次再来就是重复处理。

第二个触发点：处理时间太长，触发了消费者被判定下线。max.poll.interval.ms默认是5分钟。如果你的业务处理逻辑比较慢，一次poll的消息超过5分钟还没处理完，消费组协调器会认为这个消费者已经“死掉”，于是触发rebalance，把分区分配给组内的其他消费者。新消费者会从最近一次提交的offset重新消费，重复就产生了。这个场景在生产环境极其常见，尤其是下游数据库有锁等待、接口调用超时的时候。

第三个触发点：offset提交本身失败。消费端在业务成功之后提交offset，但是提交的时候网络抖动，或者是协调器正在rebalance，提交请求没有真正生效。下一次poll还是从旧位置拉消息，于是又重复。

第四个触发点：下游系统的重试造成重复。严格说这个不属于Kafka的重复，但业务侧看起来就是重复。消费端调用了第三方接口，超时之后自动重试，可对方其实已经处理成功了，第二次调用会把同样的数据再写一遍。

这四个触发点的共同特征是：消息在broker里只有一份，重复完全发生在“消费-处理-提交”这一条链路上。所以只盯着Kafka本身去优化，永远找不到根因。

2. 服务端侧的保障：幂等生产者与事务

2.1 幂等生产者：同一会话内的分区级去重

Kafka从0.11版本开始支持幂等生产者，开启方式很简单：enable.idempotence=true。它的底层原理是让生产者给每一条消息分配一个自增的sequence序号，broker端为每个“生产者ID+分区”维护这个序号。如果broker发现收到的消息序号比已经写入的最大序号小，说明这条消息已经写过了，直接判定为重复并拒绝。

这个机制主要防的是“生产者重试”带来的重复。比如生产者发送一批消息，网络超时，客户端自动重试，如果没有幂等保护，broker可能会把同一条消息写两次。开启幂等之后，重试的那条消息带了和之前一样的sequence，broker一比对就知道重复，直接丢弃。

但是要特别注意两个边界。第一，幂等性只在同一个生产者会话内有效。生产者重启之后，broker会重新分配一个新的生产者ID，之前的sequence记录就失效了。第二，这个去重是按分区来的，不跨分区。也就是说，如果一条消息被生产到topic的不同分区，幂等机制管不了。

所以幂等生产者解决的是“生产者写入时因为重试导致的重复”，还远远不是“消费者只处理一次”。它更准确的定位，是exactly-once链路里最基础的一块拼图。

2.2 事务性写入：跨分区原子提交

比幂等生产者更高一级的是事务性写入。Kafka的事务机制主要解决“多分区原子写”的问题。举个例子，一个订单事件需要同时写入订单topic和积分topic，正常情况下这两条消息必须要么都成功，要么都不出现。如果借助普通生产者，第一条写成功了、第二条失败了，业务系统就会看到订单存在但积分没加。

Kafka事务引入了几个关键组件：transactional.id、事务协调器、内部主题__transaction_state。生产者通过initTransactions()初始化事务，然后调用beginTransaction()开启事务，在事务里发送若干条消息，最后用commitTransaction()提交或abortTransaction()中止。

broker端会让这些消息处于“未提交”状态，只有事务成功提交后才对外可见。消费者端则需要设置isolation.level=read_committed，这样它只能读到已经提交事务的消息；如果保持默认的read_uncommitted，仍然会读到未提交甚至最终被abort的消息，这等于自欺欺人。

2.3 重要边界：服务端事务解决不了“消费端处理”的问题

到这里，很多人觉得自己已经在生产端开了幂等、开了事务，消费端也设置了read_committed，应该足够exactly-once了吧？不对。这一整套服务端机制，能保证的是“写入Kafka时消息不重不丢”，但对消费者拿到消息之后做了什么，Kafka完全管不到。

举个例子。你的消费者从Kafka里读到一条订单消息，然后往MySQL里update了一行记录，再调用了一个第三方支付接口。如果消费端在update之后、调用接口之前挂了，Kafka会认为这条消息没消费成功，重新发给另一个消费者实例。第二个消费者又把MySQL更新了一遍，可能还会再调用一次支付接口。这个过程里，Kafka的事务机制没有任何办法阻止第二次业务执行。

真正能说得上exactly-once的，是Kafka Streams这类框架：它把“消费消息、执行状态处理、产出新消息”全部纳入同一个事务边界内。但对于普通消费者程序——尤其是要写数据库、调外部接口的场景——服务端事务只是必要条件，不是充分条件。

3. 消费侧到底要怎么做：三类可落地方案

3.1 唯一键去重：最简单也最靠得住的做法

说到消费侧保证只处理一次，我最推荐、也是生产里用得最多的方案，就是唯一键去重。核心思路是：给每一条消息定义一个全局唯一的幂等键，消费端在处理消息时，把这个键写入一个带唯一约束的存储里，如果写入冲突，说明这条消息已经处理过，直接跳过。

这个方案的关键点在于：去重标记和真正的业务操作必须在同一个事务里完成。如果这两个动作分离，就会出现两种恶心情况：要么去重标记写成功了，但业务没执行，消息就丢了；要么业务执行了，但去重标记没写上，下次消息又来了，重复处理。

我举个例子。你消费Kafka消息，更新订单状态，同时插入一条订单流水。正确做法是在同一个数据库事务里，先插入一条去重记录，再执行订单更新：

@Transactional(rollbackFor = Exception.class) public void handleWithDedupe(String msgKey, OrderEvent event) { int rows = consumeRecordMapper.insertIgnore(msgKey, "order-handler", new Date()); if (rows == 0) { // 这条消息之前处理过了 return; } orderMapper.updateStatus(event.getOrderId(), event.getStatus()); }

对应的去重表DDL，要确保msg_key是唯一索引：

CREATE TABLE consume_record ( msg_key VARCHAR(128) PRIMARY KEY, handler VARCHAR(64) NOT NULL, consume_time DATETIME NOT NULL ) ENGINE=InnoDB;

这里我特别提醒一个坑：不要用“业务主键”当去重键，要用“消息编号”。同一个订单可能有“已付款”“已发货”两条不同消息，如果去重键是订单号，第二条消息就会被误判成重复。正确做法是在生产消息时给每条消息生成一个全局唯一ID，比如UUID，作为消息的key。去重键选错了，后面再怎么改都没用。

很多人会用Redis做去重，比如SETNX。这个方案不是不行，但要清楚它的局限：Redis和业务数据库是两套独立的存储系统，你很难做到“业务写库成功但Redis标记没写成功”时的原子回滚。所以我的习惯是：只有业务链路本身可接受短暂误差时，才用Redis去重；核心资金、订单类链路，老老实实用数据库唯一索引。

3.2 本地事务包围“处理+offset提交”：把重复变成无害

如果你的业务操作天然无法构造唯一键，或者消息本身就是一个无法幂等的动作，那可以考虑第二种思路：把offset提交也纳入业务事务，作为一个普通记录来管理。

具体做法是：单独建一张消费位点表，记录当前消费组消费到哪个topic的哪个分区、offset是多少。这张表和你的业务表在同一个数据库里，在同一个事务里写入。事务提交成功，业务和offset同时更新；事务失败，两者一起回滚。重启后，消费者从这张表里读出来上次的offset，通过seek()手动定位，继续消费。

这个方案的优点是，它真正解决了经典的“业务成功了，offset没提交成功”的窗口问题。因为offset不再是Kafka协调器帮你记，而是你的业务事务的一部分。缺点是：你必须自己维护消费位点，处理rebalance时分区分配变化的情况，开发量不小；而且如果业务是跨多个数据库的，单库事务就不够用了。

如果不想维护这个位点表，更轻量的一种方式依然成立：业务成功后再手动提交offset，同时接受“提交失败导致重复消息”的可能性。牺牲一点性能，配合3.1的去重表，把重复消息拦截掉，业务仍然是只处理一次。

3.3 Kafka事务+read_committed：端到端流处理的专用方案

第三种方案主要适用于Kafka到Kafka的数据管道，或者使用Kafka Streams做流处理的场景。Kafka Streams内部天然支持exactly-once，设置processing.guarantee=exactly_once_v2就可以。它把“消费、状态更新、发送结果、提交offset”全部放进同一个Kafka事务里，消费者读结果时用read_committed，就能保证端到端的恰好一次。

如果你不用Kafka Streams，而是用原生Consumer和Producer手写这个链路，也是可以实现的。核心思路是：每处理一批消息，把结果发到目标topic，同时把消费位点也通过sendOffsetsToTransaction提交到同一个事务里，再一起提交事务。只要事务提交，结果消息和offset就同时生效；事务中止，则什么都没发生。

我强调一下：这个方案的唯一适用前提是，你的处理过程不能有Kafka事务之外的外部副作用。如果你在process()里写了MySQL、调了HTTP接口，事务abort时这些外部写入是回不去Kafka事务的。所以它只适合“做纯计算转换、然后产出新消息”的管道型任务。

3.4 三个方案怎么选：先看下游，再定设计

我按自己的经验总结了一张对比表，选型的时候基本就按这个来：

4. 实操：关键参数与代码示例

4.1 消费端必改的四个参数

不管用哪种方案，消费者端的四个参数我建议直接固定下来，按下面的配置来：

如果你的业务处理链路比较重，还要同步考虑fetch.max.bytes和max.partition.fetch.bytes的限制，避免单次拉取数据量超过预期。

4.2 场景A：Kafka到MySQL的幂等消费完整示例

我实际项目中用得最多的是这种结构：从Kafka读取订单事件，更新MySQL订单状态，并插入记录。用Spring Kafka配合去重表，完整代码大概是这样：

@Component public class OrderEventConsumer { @KafkaListener(topics = "order-event", groupId = "order-group") public void onMessage(ConsumerRecord<String, OrderEvent> record, Acknowledgment ack) { try { orderEventService.handleWithDedupe(record.key(), record.value()); ack.acknowledge(); } catch (Exception e) { // 记录异常，根据业务决定是否重试或进入死信队列 log.error("consume order event failed, key={}", record.key(), e); } } }

@Service public class OrderEventService { @Transactional(rollbackFor = Exception.class) public void handleWithDedupe(String msgKey, OrderEvent event) { int inserted = consumeRecordMapper.insertIgnore(new ConsumeRecord(msgKey, "order-handler", new Date())); if (inserted == 0) { return; } orderMapper.updateStatus(event.getOrderId(), event.getStatus()); } }

消费者处理完后手动acknowledge()，如果ack失败也没关系，因为下一次重复消费时会被去重表拦截，业务不会再次执行。这个方法解决的是“业务只处理一次”，至于Kafka重复拉取了几次，其实无所谓。

需要注意的是，insertIgnore在重复时会返回0，这个判断要放在事务最前面。如果业务逻辑比较复杂，建议把这层去重单独抽象成一个公共方法，所有消费者统一入口，避免每个人写出不同的防重逻辑。

4.3 场景B：Kafka到Kafka的事务链路完整示例

再看一个用Kafka事务做端到端恰好一次的写法。这一步比较底层，适合不依赖Kafka Streams、自己管理原生API的场景：

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-etl-producer"); props.put(ProducerConfig.ACKS_CONFIG, "all"); KafkaProducer<String, String> producer = new KafkaProducer<>(props); producer.initTransactions(); Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(); while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1)); if (records.isEmpty()) { continue; } producer.beginTransaction(); try { for (ConsumerRecord<String, String> record : records) { String result = transform(record.value()); // 纯计算，不能写外部系统 producer.send(new ProducerRecord<>("order-etl-result", record.key(), result)); offsets.put( new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1) ); } producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata()); producer.commitTransaction(); } catch (Exception e) { producer.abortTransaction(); // 等待下次poll，从旧offset重新消费 } }

这个写法的关键是sendOffsetsToTransaction。它把offset和结果消息放在同一个事务里提交，避免“结果写成了但offset没提交”的经典重复窗口。消费者读结果topic时，也必须设置isolation.level=read_committed，否则说不定哪天就读到abort掉的脏数据了。

5. 常见问题与排查实录

5.1 高频问题速查表

我整理了日常排查中经常遇到的几个问题和快速解决方向：

5.2 三个典型排查案例

第一个案例，是某日志采集链路。消费者每次poll 500条消息，每条消息要做日志解析和字段规整，结果处理时间超过了5分钟，消费者被踢出组，触发rebalance，日志监控里反复出现重复消费告警。排查时先看日志有没有RebalanceInProgressException，再确认max.poll.interval.ms是否够用。最后我把max.poll.records调到100，max.poll.interval.ms调到10分钟，同时把耗时的日志解析丢到独立线程池处理，问题就消失了。

第二个案例，是某订单系统的重复流水。消费者更新订单状态并插入流水，但去重表插入和业务更新不在同一个事务里，导致并发场景下两个消费者同时处理同一条消息，一个插了去重记录，另一个没插到，业务就重复执行了。修复方式就是3.1说的：把去重插入和业务操作放进同一个@Transactional里，给流水表也加了唯一索引，彻底拦住重复。

第三个案例，是Kafka到数仓的管道。生产端启用了事务，消费端原先用的read_uncommitted，结果读到了一批abort掉的消息，数仓里出现脏数据。改成read_committed之后脏数据消失，但因为消费端的offset提交和业务写入没绑定，仍然有重复。最后在数仓侧对消息ID加了唯一约束，消费端再做一次去重，链路才算完全干净。

这个案例很典型，它说明整个exactly-once链路里只改一个环节是没用的：生产端事务管写入，消费端要管处理与提交，下游还要管幂等。三个环节都配合上，才有稳定效果。

6. 最后再说几句实在话

做了这么多年数据链路，我最大的体会是：exactly-once从来不是一个“参数”，而是一个“系统语义”。你不能指望开一个开关，Kafka就帮你把所有事情都搞定。更多时候，你要做的是想清楚每一次重复发生在哪一层，然后针对那一层去做幂等设计。

我给你一个最实际的选型建议：如果下游是数据库，优先用唯一键加本地事务去重；如果下游是第三方接口，接口侧必须做幂等控制，消费端再配合出重逻辑，否则神仙也救不了；如果下游是另一个Kafka系统，才优先考虑Kafka事务。按这个顺序来设计，基本不会翻车。

最后再分享一个小技巧：排查重复消费问题时，不要只盯着Kafka的日志，把消费端业务日志、offset提交日志、下游系统的写入日志三个时间线放在一起对齐看，重复发生的那个节点会非常清晰。这个习惯帮我省了至少一半的排查时间，也推荐给你试试。