Kafka 消息丢失全解析

一、Kafka 消息流转全景

消息丢失发生在生产者、Broker、消费者三个环节，每个环节都有不同的触发条件

二、Producer 端消息丢失（最常见）

场景 1：异步发送 + 未处理回调，消息静默丢失

原因

使用 producer.send(msg) 异步发送，不等待 Broker 确认
消息在客户端缓冲区积压，应用崩溃时缓冲区数据丢失
网络闪断导致发送失败，但应用未捕获异常，以为发送成功

代码示例（危险写法）

// ❌ 危险：异步发送，不处理结果
producer.send(new ProducerRecord<>("topic", key, value));
// 网络异常/leader切换时，消息可能静默丢失

解决方案

// ✅ 同步发送，等待结果
Future<RecordMetadata> future = producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        // 必须处理异常：重试 or 记录到本地补偿表
        log.error("发送失败", exception);
    }
});
// 或同步等待
producer.send(record).get(); // 阻塞等待结果

场景 2：acks=0 或 acks=1，Leader 未真正持久化

原因

acks=0：Producer 不等待 Broker 确认，网络丢包时消息直接丢失，应用无感知
acks=1：仅等待 Leader 写入页面缓存（Page Cache）即返回成功，Leader 在 fsync 前宕机，数据丢失
这是 Kafka 高性能的代价——以可靠性换取低延迟

acks 值	含义	丢失风险	适用场景
`acks=0`	不等待确认	极高	日志收集等可容忍丢失场景
`acks=1`	Leader 写入即确认	中	一般业务，允许极低概率丢失
`acks=all`	所有 ISR 副本确认	低	金融、订单等不允许丢失场景

解决方案

Properties props = new Properties();
props.put("acks", "all");           // 所有 ISR 副本确认
props.put("retries", 3);            // 发送失败重试次数
props.put("retry.backoff.ms", 100); // 重试间隔

场景 3：消息过大，超过 max.request.size 被静默丢弃

原因

单条消息超过 max.request.size（默认 1MB），Broker 拒绝但 Producer 可能未正确处理错误
消息被 Kafka 的 recordTooLarge 错误拒绝，若未处理回调异常则静默丢失

场景 4：Producer 缓冲区满，阻塞或丢弃

原因

buffer.memory 默认 32MB，发送速率超过 Broker 处理能力时缓冲区满
缓冲区满后，max.block.ms 超时后抛出 TimeoutException，若未捕获则消息丢失
或配置 block.on.buffer.full=false（旧版本）直接丢弃新消息

解决方案

props.put("buffer.memory", 67108864);   // 调大缓冲区 64MB
props.put("max.block.ms", 60000);       // 阻塞等待最长 60s
// 同时在 send 回调中处理异常，写入本地磁盘或降级队列

三、Broker 端消息丢失

场景 1：Leader 宕机，未同步到 Follower（数据丢失核心场景）

根本原因

Leader 写入 Page Cache 后、fsync 到磁盘前宕机（机器断电/进程被杀）
若 acks=1，Producer 已收到成功确认，但数据未落盘
Follower 尚未拉取到最新数据，选新 Leader 后这部分数据永久丢失

解决方案

# Broker 端关键配置
min.insync.replicas=2          # 写入时必须至少有 2 个 ISR 副本确认
unclean.leader.election.enable=false  # 禁止非 ISR 副本成为 Leader

# Producer 端配合
acks=all

场景 2：unclean.leader.election.enable=true，落后副本成为 Leader

原因

当 ISR 中所有副本都宕机，若允许非 ISR 副本（落后较多）成为新 Leader
原 Leader 恢复后，会以新 Leader 的数据为准，原 Leader 上未复制的数据永久丢失
Kafka 0.11 之前默认值为 true，之后默认为 false

解决方案

# 禁止非 ISR 副本竞选 Leader（Kafka 0.11+ 默认值已是 false）
unclean.leader.election.enable=false

场景 3：副本数不足（replication.factor < 2）

原因

Topic 只有 1 个副本（replication.factor=1），Broker 磁盘损坏或机器下线，数据无法恢复
云环境磁盘快照不能保证 Kafka 数据一致性（Page Cache 中有未刷盘数据）

解决方案

# 创建 Topic 时指定副本数
bin/kafka-topics.sh --create \
  --topic important-topic \
  --partitions 6 \
  --replication-factor 3 \
  --config min.insync.replicas=2

场景 4：日志清理策略导致消息被提前删除

原因

log.retention.hours 默认 168 小时（7天），超过后消息被删除
log.retention.bytes 达到阈值后，最旧段被删除
若 Consumer 消费速度慢，未被消费的消息可能被清理（取决于是否配置了基于 offset 的保留）
Kafka 2.6+ 支持 log.retention.ms=-1 永久保留，但磁盘有风险

四、Consumer 端消息丢失（最隐蔽）

场景 1：自动提交 offset + 消费逻辑异常（经典陷阱）

原因

enable.auto.commit=true（默认值），每隔 auto.commit.interval.ms（默认 5s）自动提交 offset
若 poll 到消息后、业务处理完成前，offset 已被自动提交，此时 Consumer 崩溃，这条消息永远不会再被处理
这是生产环境消息丢失的 "头号杀手"

// ❌ 危险：自动提交 + 异步处理
props.put("enable.auto.commit", "true");  // 默认就是 true

while (true) {
    ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(100));
    // 自动提交已发生，但下面业务逻辑可能失败
    processRecords(records);  // 若此处抛异常，消息已确认，永久丢失
}

解决方案：手动提交 offset

// ✅ 关闭自动提交，业务处理完成后手动提交
props.put("enable.auto.commit", "false");

while (true) {
    ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(100));
    try {
        processRecords(records);  // 业务处理
        // 同步提交，确保提交成功
        consumer.commitSync();
    } catch (Exception e) {
        // 处理失败，offset 不提交，消息会被重新消费
        log.error("处理失败，等待重试", e);
    }
}

场景 2：多线程消费，offset 提交时机错误

原因

将 poll 到的消息丢入线程池异步处理，主线程立即提交 offset
异步线程尚未处理完，Consumer 已提交 offset；若此时 Consumer 重启，这部分消息丢失

解决方案

使用 commitSync(Map<TopicPartition, OffsetAndMetadata>) 按需提交，处理完一批后再提交对应 offset
或使用 ConsumerRebalanceListener 在 rebalance 时提交已处理任务的 offset

场景 3：Consumer 被踢出 Group（心跳超时）

原因

session.timeout.ms 默认 10s，若消费逻辑处理时间过长超过此值，Broker 认为 Consumer 死掉，触发 Rebalance
Rebalance 后分区分配给其他 Consumer，若原 Consumer 已处理完但未提交 offset，新 Consumer 会重新消费（导致重复，不是丢失）
但若原 Consumer 已自动提交 offset，新 Consumer 从已提交位置开始，原 Consumer 处理中的消息丢失

解决方案

props.put("session.timeout.ms", 30000);    // 适当加大会话超时
props.put("max.poll.interval.ms", 300000);  // 处理最大间隔，默认 5 分钟
props.put("enable.auto.commit", "false");   // 关闭自动提交

五、消息丢失场景对比总结

丢失环节	具体场景	丢失概率	检测难度	核心配置
Producer	acks=0，不等待确认	极高	难（静默丢失）	`acks=all`
Producer	异步发送未处理异常	高	中	处理 Callback 异常
Broker	Leader 宕机，数据未复制	中	难	`min.insync.replicas≥2` `unclean.leader.election=false`
Broker	副本数=1，磁盘损坏	极高	易	`replication.factor≥3`
Consumer	自动提交 + 消费失败	极高	中（有日志可查）	`enable.auto.commit=false`
Consumer	多线程消费，提前提交 offset	中	难	手动提交 + 幂等消费

六、如何保证消息不丢失（生产级配置）

Producer 端配置

acks=all
retries=Integer.MAX_VALUE
max.in.flight.requests.per.connection=1（保证有序）
处理 send 回调异常，写入本地补偿表

Broker 端配置

replication.factor=3
min.insync.replicas=2
unclean.leader.election.enable=false
禁用磁盘缓存刷写（依赖副本机制）

Consumer 端配置

enable.auto.commit=false
业务处理成功后手动提交
实现幂等消费（支持重复消费）
isolation.level=read_committed（事务）

监控与告警

监控 UnderReplicatedPartitions
监控 OfflineLogDirectoryCount
监控 Consumer Lag 异常增长
消息轨迹追踪（注入 traceId）

端到端不丢失方案（Exactly-Once 语义）

Producer 幂等性：启用 enable.idempotence=true，避免因重试导致重复消息
事务消息：使用 Kafka 事务（initTransactions/beginTransaction/commitTransaction），实现跨分区的原子写入
Consumer 幂等：消费逻辑实现幂等（基于消息 key 去重、Redis/DB 唯一约束），支持重复消费
兜底补偿：Producer 发送失败时，将消息写入本地 WAL 日志或补偿表，定时重试

七、如何检测消息是否丢失

1. 消息轨迹追踪

每条消息注入全局唯一 traceId，Producer 记录发送日志，Consumer 记录消费日志
通过 Trace 系统比对发送量与消费量

2. 监控 Consumer Lag

kafka-consumer-groups.sh --bootstrap-server ... --describe 查看 Lag
Lag 突然下降（Consumer 跳过了消息）或持续增大（消费停滞）都是异常信号

3. checksum 比对

Producer 发送时计算消息体 checksum，Consumer 消费后比对，不一致说明消息被篡改或丢失

4. 端到端对账

上游业务数据库记录写入条数，与下游 Consumer 消费条数定时比对
这是最可靠但成本最高的检测方式