当前位置：首页 > news >正文

Kafka 为什么会丢消息？根源剖析与避坑指南

news 2026/5/11 23:35:48

一、生产端：消息根本没有发出去

消息在发送到 Kafka 集群之前，或者在发送途中丢失，这是最常见的“丢消息”源头。

1. 盲目配置 acks=0

机制：生产者中acks参数决定了多少个副本确认收到后，才认为消息写入成功。

acks=0：生产者发出去就算成功，不管有没有落到磁盘。如果网络抖动或 Broker 挂掉，消息直接丢弃。
acks=1：默认值，只要 Leader 副本收到就算成功。如果 Leader 宕机，Follower 还没同步，消息丢失。

2. 未处理发送异常

生产者调用send()实为异步操作，若未调用get()或设置回调，发送失败时不会有任何提示，消息在客户端缓冲区或网络传输中丢失。

3. 缓冲区溢出

生产者有缓冲机制（buffer.memory）。若发送速度 > Broker 处理速度，缓冲区满后，如果max.block.ms超时或阻塞策略配置不当，新消息会抛出异常或被丢弃。

生产端配置	可靠性等级	丢消息风险
`acks=0`	最高性能	最高风险
`acks=1`	性能与可靠平衡	中等风险（Leader 切换时）
`acks=all`或`-1`	最高可靠	最低风险（配合`min.insync.replicas`）

二、 Broker 服务端：副本机制失效

即使生产者成功发送，Broker 在存储环节也可能弄丢消息。

1. 副本 ISR 收缩与覆盖写

Leader 负责读写，Follower 负责同步。当 Leader 写成功但未同步给 Follower 就宕机，且新 Leader 没有这条消息，就会丢消息。
脑裂场景：网络抖动导致 Leader 被孤立，但它仍在写数据。集群选出新 Leader 后，旧 Leader 恢复时，其多写的数据会被截断丢弃。

2. 刷盘机制

Kafka 依赖操作系统 PageCache（异步刷盘）。若 Broker 断电或 Crash，PageCache 中未来得及写入磁盘的数据就会丢失。

3. 磁盘故障

强制多副本可以抵御单盘故障。如果replication.factor=1，磁盘损坏直接丢数据。

4. 消息过期清理

如果消费速度严重落后于生产速度，且消息保留时间（log.retention.hours）太短，消息在被消费前就会被物理删除。

三、消费端：根本没处理完就算成功

1. 自动提交的陷阱

Kafka 消费端默认自动提交偏移量。
致命场景：

拉取消息 x~y，offset 已自动提交。
处理到一半（如第 n 条）程序崩溃。
重启后从已提交的 offset 开始消费，中间那部分消息永远无法被处理。

2. 手动提交时机错误

错误示范：先提交 offset，再处理消息。如果在处理前崩溃，消息直接丢失。

3. 消费超时被踢出组

如果单条消息处理耗时超过max.poll.interval.ms（默认 5 分钟），Broker 认为消费者死亡，触发 Rebalance 将分区分配给其他消费者。若原消费者还在处理但未提交 offset，处理结果丢失，新消费者重复消费。

4. 偏移量过期

消费者组空闲超过offsets.retention.minutes（默认 7 天），offset 被删除。重新启动时，若auto.offset.reset设为latest，会跳过积压数据造成丢失；若设为earliest，则大量重复。

四、罪魁祸首：Rebalance 与网络分区

很多丢失不是代码 Bug，而是分布式协调的副作用。

1. Rebalance 期间的“断档”

触发 Rebalance 期间，整个消费组会停止消费，导致消息积压。如果积压时间过长超过保留时间，消息被清除。

2. offset 提交的竞态条件

丢数据：先 commit 后处理，Rebalance 打断处理。
重复数据：先处理后 commit，Rebalance 打断 commit，重启后重复消费。

3. 网络分区与“脏选举”

当 Broker 发生网络分区，若开启不完全 Leader 选举，允许非 ISR 中的副本成为 Leader。这个副本可能缺少大量消息，一旦当选，这些消息就丢失了。

五、实战避坑配置指南

1. 生产端保命配置

properties

# 保证消息不丢的最低配置 acks=all enable.idempotence=true # 幂等性，配合 acks=all 保证不重复不丢顺序 retries=Integer.MAX_VALUE max.in.flight.requests.per.connection=5 # 幂等下可大于1且保证顺序

2. Broker 端加固配置

properties

# Topic 级别 replication.factor=3 min.insync.replicas=2 # 配合 acks=all，至少写入2个副本才算成功 unclean.leader.election.enable=false # 禁止非 ISR 副本选举为 Leader # 刷盘（非必须，依赖副本即可） log.flush.interval.messages=10000

3. 消费端安全配置

java

// Java 示例 props.put("enable.auto.commit", "false"); // 必须手动提交 props.put("max.poll.interval.ms", "300000"); // 根据处理耗时调整 props.put("max.poll.records", "500"); // 控制单次拉取数量，防止超时 // 代码逻辑 while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { try { process(record); // 先处理 } catch (Exception e) { // 建议记录异常并跳过，或引入死信队列，防止阻塞 } } consumer.commitSync(); // 处理完再提交 }