上一篇【第61篇】Kafka可靠性保证全解析——acks、ISR、min.insync.replicas那点事下一篇【第63篇】Kafka副本机制深度解析——Leader选举是如何保证数据不丢的摘要“消息队列丢消息是每个后端工程师的噩梦。Kafka的消息不丢失不是某一个配置能搞定的它是一个从生产者到Broker到消费者的全链路工程。任何一环松懈都可能千里之堤溃于蚁穴”。本文是Kafka可靠性系列的第二篇聚焦实战。我们将从生产者端重试、幂等、Broker端副本、配置、消费者端offset提交三个维度逐层拆解防丢策略。结尾附上一个真实的线上故障复盘——某电商平台在双十一期间丢消息的完整排查过程。一、全链路防丢全景图在开始逐层剖析之前先搞清楚消息从生产到消费的完整旅程中哪些环节可能丢消息【消息全链路丢失风险点】 生产者 (Producer) Broker 消费者 (Consumer) ┌──────────────┐ ┌──────────────────┐ ┌──────────────────┐ │ │ │ │ │ │ │ ① 网络问题 │ │ ④ Leader 宕机 │ │ ⑥ 自动提交 offset│ │ 发送失败 │ │ Follower 未同步│ │ 消息未处理完 │ │ │ │ │ │ │ │ ② 序列化异常 │ │ ⑤ 磁盘满/崩溃 │ │ ⑦ 先提交后处理 │ │ 消息未发出 │ │ 数据未持久化 │ │ 但处理失败 │ │ │ │ │ │ │ │ ③ 重试耗尽 │ │ │ │ ⑧ Rebalance 期间 │ │ 丢入黑洞 │ │ │ │ 消费中断 │ │ │ │ │ │ │ └──────────────┘ └──────────────────┘ └──────────────────┘ 结论三个端点、八个风险点需要三端协防二、生产者端——发送出去只是第一步2.1 发送失败不可怕可怕的是你不知道它失败了// ❌ 错误姿势发送即丢弃producer.send(newProducerRecord(topic,key,value));// 没有回调发送失败了也不知道// ✅ 正确姿势带回调感知结果producer.send(newProducerRecord(topic,key,value),newCallback(){OverridepublicvoidonCompletion(RecordMetadatametadata,Exceptionexception){if(exception!null){// 发送失败写入本地文件兜底writeToFallbackFile(record);// 发出告警alert(Kafka发送失败: exception.getMessage());}}});2.2 重试机制的合理配置Kafka生产者的重试是自动的但需要正确配置参数组合【重试机制工作原理】 Producer 发送 msg100 │ ▼ ┌─────────┐ 失败 ┌──────────┐ │ 第1次发送│─────────► │ 等待重试 │ └─────────┘ └────┬─────┘ │ │ │ 成功 ┌──────▼──────┐ ▼ │ 第2次发送 │ 返回OK └──────┬──────┘ │ │ 失败 │ ▼ ┌──────────┐ │ 第N次... │ └─────┬────┘ │ ┌─────────▼──────────┐ │ 超过 delivery.timeout│ │ → 最终失败回调通知 │ └────────────────────┘// 重试配置PropertiespropsnewProperties();props.put(acks,all);props.put(retries,Integer.MAX_VALUE);// 无限重试props.put(delivery.timeout.ms,120000);// 2分钟内仍可重试props.put(request.timeout.ms,30000);// 单次请求30秒超时props.put(retry.backoff.ms,100);// 重试间隔100ms指数退避// delivery.timeout.ms 是总超时超过这个时间仍未成功 → 最终失败// retries * retry.backoff.ms 不能超过 delivery.timeout.ms2.3 幂等性——防止消息重复重试最大的副作用是重复消息。开启幂等性后Kafka自动为每条消息赋予标识Broker端自动去重// 开启幂等性Kafka 0.11props.put(enable.idempotence,true);// 这意味着// 1. Kafka 自动设置 acksall// 2. 自动设置 retriesInteger.MAX_VALUE// 3. 自动设置 max.in.flight.requests.per.connection ≤ 5// 4. 每条消息带上 Producer ID (PID) 和 sequence number【幂等性工作原理】 无幂等性 Producer → send(msg1, seq0) → Broker写入 msg1 Producer → retry(msg1, seq0) → Broker又写入 msg1 ← 重复 有幂等性 Producer → send(msg1, seq0) → Broker写入 msg1, 记录 PID_idlast_seq0 Producer → retry(msg1, seq0) → Broker检查: PID_id的last_seq0 → seq0 不大于 last_seq0 → 判定为重复忽略 ← 防住了2.4 生产者防丢checklist配置项推荐值防护的丢失场景acksallLeader宕机未同步enable.idempotencetrue网络重试导致重复retriesInteger.MAX_VALUE暂时性故障delivery.timeout.ms120000给重试足够时间max.block.ms默认60000buffer满时不要无限阻塞回调处理必须实现感知发送失败兜底机制写本地文件/DB最终失败时的保险三、Broker端——数据安全的中心堡垒3.1 核心配置# server.properties 防丢配置 # 1. 默认副本数全局默认 default.replication.factor3 # 2. 最小同步副本数全局默认 min.insync.replicas2 # 3. 禁止不干净的Leader选举绝对不要改 unclean.leader.election.enablefalse # 4. 日志刷盘间隔建议不设置信任OS的页缓存 # log.flush.interval.messages10000 ← 不推荐太耗性能 # log.flush.interval.ms1000 ← 不推荐 # 5. 自动创建Topic生产环境建议关闭 auto.create.topics.enablefalse3.2 副本数Replication Factor的规划【副本数选择指南】 RF1: PROD环境不要用 ┌──────────────────────────────────────┐ │ Broker1 → 数据全丢 │ └──────────────────────────────────────┘ RF2: 最低生产标准 ┌──────────────────────────────────────┐ │ Broker1 → Broker2 顶上 │ │ 但不能同时坏两个 │ │ min.isr1: 单点故障仍可写但有丢风险 │ │ min.isr2: 任意1个不可用就拒绝写入 │ └──────────────────────────────────────┘ RF3: 推荐配置 ★ ┌──────────────────────────────────────┐ │ Broker1 → Broker2/3 顶上 │ │ 建议 min.isr2容错且可写 │ │ 可以扛住2个Broker故障但写入会阻塞 │ └──────────────────────────────────────┘3.3 Broker端的物理可靠性【Broker挂掉场景矩阵】 场景A: 进程挂了机器正常 ┌──────────────────────────────────────────┐ │ 进程被kill/oom/GC卡死 │ │ → 页缓存里的数据还没刷盘会丢吗 │ │ │ │ 不会丢因为副本机制 │ │ Follower 1 和 Follower 2 也存着同样的数据 │ │ Leader 重启后从 Follower 拉取恢复 │ └──────────────────────────────────────────┘ 场景B: 机器宕机磁盘完好 ┌──────────────────────────────────────────┐ │ 机器断电 → 重启 → 磁盘数据还在 │ │ → 页缓存丢失但 ~10分钟以内的数据 │ │ 可以从 Follower 恢复副本机制 │ │ → ~10分钟以前的从磁盘恢复OS已刷盘 │ └──────────────────────────────────────────┘ 场景C: 磁盘坏了 ┌──────────────────────────────────────────┐ │ 磁盘物理损坏 → 数据无法恢复 │ │ → 还好有副本其他Broker有完整数据 │ │ → 坏盘Broker重启后从Leader全量同步 │ │ → RF≥3至少还有一个副本活着 │ └──────────────────────────────────────────┘3.4 不要手动设置刷盘频率有句老话“让操作系统管内存你管好业务逻辑就好。”【为什么推荐不设置 log.flush.interval.messages】 情况A不设刷盘推荐 ┌──────────────────────────────────────────┐ │ 消息写入 → 页缓存内存→ OS异步刷盘 │ │ 性能极好 │ │ 数据安全通过副本机制保证 │ │ Leader宕机 → Follower有全量数据 │ └──────────────────────────────────────────┘ 情况B每次写入都刷盘 ┌──────────────────────────────────────────┐ │ 消息写入 → 页缓存 → fsync() → 磁盘 │ │ 性能很差跟随机写差不多了 │ │ 数据安全单机可靠性↑但吞吐暴跌 │ │ 而且本来就有副本兜底没必要... │ └──────────────────────────────────────────┘ 结论副本机制已经提供了数据安全 再手动刷盘属于双重保险过度四、消费者端——消费完再提交别急着说好了4.1 自动提交的陷阱// ❌ 自动提交的经典坑props.put(enable.auto.commit,true);props.put(auto.commit.interval.ms,5000);// 每5秒自动提交// 场景推演// T0: poll() 拿到 msg1~msg100耗时1ms// T1: 开始业务处理 msg1~msg100处理到 msg50// T2: 5秒到了自动提交 offset100 ← 但 msg51~100 还没处理完// T3: 消费者挂了// T4: 重启后从 offset100 开始消费// T5: msg51~100 被跳过了 ← 消息丢失4.2 手动提交的正确姿势// ✅ 方案一先处理再同步提交最安全但性能最低while(true){ConsumerRecordsString,Stringrecordsconsumer.poll(Duration.ofMillis(100));for(ConsumerRecordString,Stringrecord:records){processRecord(record);// 先处理业务逻辑}// 所有消息处理完后再提交offsetconsumer.commitSync();// 同步提交确保提交成功}// ✅ 方案二异步提交 同步兜底推荐try{while(true){ConsumerRecordsString,Stringrecordsconsumer.poll(Duration.ofMillis(100));for(ConsumerRecordString,Stringrecord:records){processRecord(record);}// 异步提交不阻塞处理consumer.commitAsync(newOffsetCommitCallback(){OverridepublicvoidonComplete(MapTopicPartition,OffsetAndMetadataoffsets,Exceptione){if(e!null){log.error(提交失败: ,e);}}});}}finally{// 关闭前最后一次用同步提交确保不丢consumer.commitSync();}// ✅ 方案三逐条提交精确控制最精确但也很慢while(true){ConsumerRecordsString,Stringrecordsconsumer.poll(Duration.ofMillis(100));for(ConsumerRecordString,Stringrecord:records){try{processRecord(record);// 处理成功一条就提交这一条的 offset1MapTopicPartition,OffsetAndMetadataoffsetnewHashMap();offset.put(newTopicPartition(record.topic(),record.partition()),newOffsetAndMetadata(record.offset()1));consumer.commitSync(offset);}catch(Exceptione){// 处理失败不提交 → 下次重新消费这条消息log.error(处理失败: {},record,e);}}}4.3 消息去重——冪等消费即使手动提交offset也可能因为提交前崩溃导致重复消费。消费者端需要实现幂等性// 幂等消费的常见方案基于唯一键去重publicvoidconsumeRecord(ConsumerRecordString,Stringrecord){StringbusinessKeyextractBusinessKey(record);// 订单ID、事件ID等// 方案1Redis去重// 利用 Redis 的 SETNX设置24小时过期StringcacheKeykafka:dedup:businessKey;BooleanfirstTimeredis.setnx(cacheKey,1,Duration.ofHours(24));if(!firstTime){log.info(重复消息跳过: {},businessKey);return;// 已处理过跳过}// 方案2数据库唯一约束try{// 假设 event_log 表有 UNIQUE(business_key)jdbcTemplate.update(INSERT INTO event_log (business_key, payload, create_time) VALUES (?,?,?),businessKey,record.value(),newDate());doBusinessLogic(record);// 处理业务}catch(DuplicateKeyExceptione){log.info(重复消息跳过: {},businessKey);}}4.4 消费者防丢checklist配置/实践推荐方式防护的丢失场景提交方式手动提交commitSync自动提交时机早于处理完成offset提交时机消息全部处理完后提交后处理失败幂等消费Redis/DB 去重表提交前崩溃导致重复enable.auto.commitfalse自动提交抢占控制auto.offset.resetearliest(一般)/latest(如果可接受丢)新消费者组的行为关闭前提交commitSync()infinallyJVM关闭时未提交Rebalance时提交ConsumerRebalanceListener.onPartitionsRevokedRebalance时未提交五、全链路exactly-once方案把三端的防丢策略串起来就是exactly-once的完整方案【全链路 Exactly-Once 实现】 生产者端 Broker端 消费者端 ┌──────────────┐ ┌──────────────────┐ ┌──────────────────┐ │ acksall │ │ RF3 │ │ 手动提交offset │ │ 幂等Producer │ │ min.isr2 │ │ 先处理后提交 │ │ retriesMAX │ │ unclean✗ │ │ 幂等消费(去重) │ │ callback兜底 │ │ 滚动重启策略 │ │ RebalanceListener│ │ 事务Producer │ │ 多机房部署 │ │ finally提交offset│ └──────┬───────┘ └────────┬─────────┘ └────────┬─────────┘ │ │ │ ▼ ▼ ▼ 消息不会发丢 消息不会存丢 消息不会消丢 多 Topic 事务跨分区原子写入 ┌──────────────────────────────────────────────────────────┐ │ producer.beginTransaction(); │ │ producer.send(topicA, key, value); │ │ producer.send(topicB, key, value); │ │ // 同时写入消费者端的去重表 │ │ jdbcTemplate.update(INSERT INTO dedup_table ...); │ │ producer.commitTransaction(); // 原子提交 │ │ // 如果去重表写入失败整个事务回滚 │ └──────────────────────────────────────────────────────────┘六、线上故障复盘双十一幽灵订单事件6.1 故障现象时间2024年双十一凌晨2点现象客服收到用户投诉——“钱扣了订单没生成”技术表现支付回调消息在Kafka中丢失订单服务没收到付款通知6.2 排查过程【故障排查时间线】 00:30 流量开始上升每秒 8000 笔订单 01:00 Broker2 磁盘使用率 92%开始频繁触发 GC 01:15 Broker2 被剔除 ISRFollower跟不上 01:20 Controller 发现 Broker2 异常尝试 Leader 迁移 01:22 迁移期间Broker1 成为新 Leader 01:23 原 Broker2 上的部分消息未同步到 Broker1 → 丢失 排查发现 1. acks1 → Leader写入就返回Follower没确认 2. min.insync.replicas1默认值→ 没有防丢门槛 3. 原Leader宕机时有约200条消息没同步给Follower 4. 这些消息对应的订单钱扣了但订单没生成6.3 改进措施# 改进1提升acks级别 acksall # 从 ack1 改为 all # 改进2增加 min.insync.replicas min.insync.replicas2 # 从 1 改为 2 # 改进3禁止不干净的选举 unclean.leader.election.enablefalse # 已经设置没问题 # 改进4生产者回调兜底 # 发送失败时写入数据库队列定时重试// 改进5生产者端最坏情况兜底producer.send(record,(metadata,exception)-{if(exception!null){// 写入本地待重试文件failedMessageWriter.append(record);// 每分钟由定时任务扫描并重试// 超过1小时仍未成功 → 人工介入}});// 改进6消费者端幂等性// 支付回调消息中的 transaction_id 做去重// 即使重复消费也不会重复创建订单6.4 效果对比指标改进前改进后acks1allmin.isr12消息丢失双十一丢200条历经多次大促0丢失吞吐量影响-延迟5ms吞吐基本不变生产者兜底无文件兜底定时重试告警响应手动排查Prometheus指标自动告警本篇小结消息不丢失是三端联合工程缺一不可生产者端acksall 幂等 重试 callback兜底确保消息发出去并被确认Broker端RF3 min.isr2 uncleanfalse确保消息存住了且有多份备份消费者端手动提交 先处理后提交 幂等消费确保消息消费完且不重复记住口诀发的要确认、存的要多份、吃的要消化了再买单。下一篇我们将深入Kafka最核心的机制之一——副本同步与Leader选举彻底搞清楚那个Follower是怎么追上Leader的经典问题。上一篇【第61篇】Kafka可靠性保证全解析——acks、ISR、min.insync.replicas那点事下一篇【第63篇】Kafka副本机制深度解析——Leader选举是如何保证数据不丢的
【Kafka源码解读和使用指南】第62篇：Kafka数据不丢失实战指南——生产者、Broker、消费者三端防护
上一篇【第61篇】Kafka可靠性保证全解析——acks、ISR、min.insync.replicas那点事下一篇【第63篇】Kafka副本机制深度解析——Leader选举是如何保证数据不丢的摘要“消息队列丢消息是每个后端工程师的噩梦。Kafka的消息不丢失不是某一个配置能搞定的它是一个从生产者到Broker到消费者的全链路工程。任何一环松懈都可能千里之堤溃于蚁穴”。本文是Kafka可靠性系列的第二篇聚焦实战。我们将从生产者端重试、幂等、Broker端副本、配置、消费者端offset提交三个维度逐层拆解防丢策略。结尾附上一个真实的线上故障复盘——某电商平台在双十一期间丢消息的完整排查过程。一、全链路防丢全景图在开始逐层剖析之前先搞清楚消息从生产到消费的完整旅程中哪些环节可能丢消息【消息全链路丢失风险点】 生产者 (Producer) Broker 消费者 (Consumer) ┌──────────────┐ ┌──────────────────┐ ┌──────────────────┐ │ │ │ │ │ │ │ ① 网络问题 │ │ ④ Leader 宕机 │ │ ⑥ 自动提交 offset│ │ 发送失败 │ │ Follower 未同步│ │ 消息未处理完 │ │ │ │ │ │ │ │ ② 序列化异常 │ │ ⑤ 磁盘满/崩溃 │ │ ⑦ 先提交后处理 │ │ 消息未发出 │ │ 数据未持久化 │ │ 但处理失败 │ │ │ │ │ │ │ │ ③ 重试耗尽 │ │ │ │ ⑧ Rebalance 期间 │ │ 丢入黑洞 │ │ │ │ 消费中断 │ │ │ │ │ │ │ └──────────────┘ └──────────────────┘ └──────────────────┘ 结论三个端点、八个风险点需要三端协防二、生产者端——发送出去只是第一步2.1 发送失败不可怕可怕的是你不知道它失败了// ❌ 错误姿势发送即丢弃producer.send(newProducerRecord(topic,key,value));// 没有回调发送失败了也不知道// ✅ 正确姿势带回调感知结果producer.send(newProducerRecord(topic,key,value),newCallback(){OverridepublicvoidonCompletion(RecordMetadatametadata,Exceptionexception){if(exception!null){// 发送失败写入本地文件兜底writeToFallbackFile(record);// 发出告警alert(Kafka发送失败: exception.getMessage());}}});2.2 重试机制的合理配置Kafka生产者的重试是自动的但需要正确配置参数组合【重试机制工作原理】 Producer 发送 msg100 │ ▼ ┌─────────┐ 失败 ┌──────────┐ │ 第1次发送│─────────► │ 等待重试 │ └─────────┘ └────┬─────┘ │ │ │ 成功 ┌──────▼──────┐ ▼ │ 第2次发送 │ 返回OK └──────┬──────┘ │ │ 失败 │ ▼ ┌──────────┐ │ 第N次... │ └─────┬────┘ │ ┌─────────▼──────────┐ │ 超过 delivery.timeout│ │ → 最终失败回调通知 │ └────────────────────┘// 重试配置PropertiespropsnewProperties();props.put(acks,all);props.put(retries,Integer.MAX_VALUE);// 无限重试props.put(delivery.timeout.ms,120000);// 2分钟内仍可重试props.put(request.timeout.ms,30000);// 单次请求30秒超时props.put(retry.backoff.ms,100);// 重试间隔100ms指数退避// delivery.timeout.ms 是总超时超过这个时间仍未成功 → 最终失败// retries * retry.backoff.ms 不能超过 delivery.timeout.ms2.3 幂等性——防止消息重复重试最大的副作用是重复消息。开启幂等性后Kafka自动为每条消息赋予标识Broker端自动去重// 开启幂等性Kafka 0.11props.put(enable.idempotence,true);// 这意味着// 1. Kafka 自动设置 acksall// 2. 自动设置 retriesInteger.MAX_VALUE// 3. 自动设置 max.in.flight.requests.per.connection ≤ 5// 4. 每条消息带上 Producer ID (PID) 和 sequence number【幂等性工作原理】 无幂等性 Producer → send(msg1, seq0) → Broker写入 msg1 Producer → retry(msg1, seq0) → Broker又写入 msg1 ← 重复 有幂等性 Producer → send(msg1, seq0) → Broker写入 msg1, 记录 PID_idlast_seq0 Producer → retry(msg1, seq0) → Broker检查: PID_id的last_seq0 → seq0 不大于 last_seq0 → 判定为重复忽略 ← 防住了2.4 生产者防丢checklist配置项推荐值防护的丢失场景acksallLeader宕机未同步enable.idempotencetrue网络重试导致重复retriesInteger.MAX_VALUE暂时性故障delivery.timeout.ms120000给重试足够时间max.block.ms默认60000buffer满时不要无限阻塞回调处理必须实现感知发送失败兜底机制写本地文件/DB最终失败时的保险三、Broker端——数据安全的中心堡垒3.1 核心配置# server.properties 防丢配置 # 1. 默认副本数全局默认 default.replication.factor3 # 2. 最小同步副本数全局默认 min.insync.replicas2 # 3. 禁止不干净的Leader选举绝对不要改 unclean.leader.election.enablefalse # 4. 日志刷盘间隔建议不设置信任OS的页缓存 # log.flush.interval.messages10000 ← 不推荐太耗性能 # log.flush.interval.ms1000 ← 不推荐 # 5. 自动创建Topic生产环境建议关闭 auto.create.topics.enablefalse3.2 副本数Replication Factor的规划【副本数选择指南】 RF1: PROD环境不要用 ┌──────────────────────────────────────┐ │ Broker1 → 数据全丢 │ └──────────────────────────────────────┘ RF2: 最低生产标准 ┌──────────────────────────────────────┐ │ Broker1 → Broker2 顶上 │ │ 但不能同时坏两个 │ │ min.isr1: 单点故障仍可写但有丢风险 │ │ min.isr2: 任意1个不可用就拒绝写入 │ └──────────────────────────────────────┘ RF3: 推荐配置 ★ ┌──────────────────────────────────────┐ │ Broker1 → Broker2/3 顶上 │ │ 建议 min.isr2容错且可写 │ │ 可以扛住2个Broker故障但写入会阻塞 │ └──────────────────────────────────────┘3.3 Broker端的物理可靠性【Broker挂掉场景矩阵】 场景A: 进程挂了机器正常 ┌──────────────────────────────────────────┐ │ 进程被kill/oom/GC卡死 │ │ → 页缓存里的数据还没刷盘会丢吗 │ │ │ │ 不会丢因为副本机制 │ │ Follower 1 和 Follower 2 也存着同样的数据 │ │ Leader 重启后从 Follower 拉取恢复 │ └──────────────────────────────────────────┘ 场景B: 机器宕机磁盘完好 ┌──────────────────────────────────────────┐ │ 机器断电 → 重启 → 磁盘数据还在 │ │ → 页缓存丢失但 ~10分钟以内的数据 │ │ 可以从 Follower 恢复副本机制 │ │ → ~10分钟以前的从磁盘恢复OS已刷盘 │ └──────────────────────────────────────────┘ 场景C: 磁盘坏了 ┌──────────────────────────────────────────┐ │ 磁盘物理损坏 → 数据无法恢复 │ │ → 还好有副本其他Broker有完整数据 │ │ → 坏盘Broker重启后从Leader全量同步 │ │ → RF≥3至少还有一个副本活着 │ └──────────────────────────────────────────┘3.4 不要手动设置刷盘频率有句老话“让操作系统管内存你管好业务逻辑就好。”【为什么推荐不设置 log.flush.interval.messages】 情况A不设刷盘推荐 ┌──────────────────────────────────────────┐ │ 消息写入 → 页缓存内存→ OS异步刷盘 │ │ 性能极好 │ │ 数据安全通过副本机制保证 │ │ Leader宕机 → Follower有全量数据 │ └──────────────────────────────────────────┘ 情况B每次写入都刷盘 ┌──────────────────────────────────────────┐ │ 消息写入 → 页缓存 → fsync() → 磁盘 │ │ 性能很差跟随机写差不多了 │ │ 数据安全单机可靠性↑但吞吐暴跌 │ │ 而且本来就有副本兜底没必要... │ └──────────────────────────────────────────┘ 结论副本机制已经提供了数据安全 再手动刷盘属于双重保险过度四、消费者端——消费完再提交别急着说好了4.1 自动提交的陷阱// ❌ 自动提交的经典坑props.put(enable.auto.commit,true);props.put(auto.commit.interval.ms,5000);// 每5秒自动提交// 场景推演// T0: poll() 拿到 msg1~msg100耗时1ms// T1: 开始业务处理 msg1~msg100处理到 msg50// T2: 5秒到了自动提交 offset100 ← 但 msg51~100 还没处理完// T3: 消费者挂了// T4: 重启后从 offset100 开始消费// T5: msg51~100 被跳过了 ← 消息丢失4.2 手动提交的正确姿势// ✅ 方案一先处理再同步提交最安全但性能最低while(true){ConsumerRecordsString,Stringrecordsconsumer.poll(Duration.ofMillis(100));for(ConsumerRecordString,Stringrecord:records){processRecord(record);// 先处理业务逻辑}// 所有消息处理完后再提交offsetconsumer.commitSync();// 同步提交确保提交成功}// ✅ 方案二异步提交 同步兜底推荐try{while(true){ConsumerRecordsString,Stringrecordsconsumer.poll(Duration.ofMillis(100));for(ConsumerRecordString,Stringrecord:records){processRecord(record);}// 异步提交不阻塞处理consumer.commitAsync(newOffsetCommitCallback(){OverridepublicvoidonComplete(MapTopicPartition,OffsetAndMetadataoffsets,Exceptione){if(e!null){log.error(提交失败: ,e);}}});}}finally{// 关闭前最后一次用同步提交确保不丢consumer.commitSync();}// ✅ 方案三逐条提交精确控制最精确但也很慢while(true){ConsumerRecordsString,Stringrecordsconsumer.poll(Duration.ofMillis(100));for(ConsumerRecordString,Stringrecord:records){try{processRecord(record);// 处理成功一条就提交这一条的 offset1MapTopicPartition,OffsetAndMetadataoffsetnewHashMap();offset.put(newTopicPartition(record.topic(),record.partition()),newOffsetAndMetadata(record.offset()1));consumer.commitSync(offset);}catch(Exceptione){// 处理失败不提交 → 下次重新消费这条消息log.error(处理失败: {},record,e);}}}4.3 消息去重——冪等消费即使手动提交offset也可能因为提交前崩溃导致重复消费。消费者端需要实现幂等性// 幂等消费的常见方案基于唯一键去重publicvoidconsumeRecord(ConsumerRecordString,Stringrecord){StringbusinessKeyextractBusinessKey(record);// 订单ID、事件ID等// 方案1Redis去重// 利用 Redis 的 SETNX设置24小时过期StringcacheKeykafka:dedup:businessKey;BooleanfirstTimeredis.setnx(cacheKey,1,Duration.ofHours(24));if(!firstTime){log.info(重复消息跳过: {},businessKey);return;// 已处理过跳过}// 方案2数据库唯一约束try{// 假设 event_log 表有 UNIQUE(business_key)jdbcTemplate.update(INSERT INTO event_log (business_key, payload, create_time) VALUES (?,?,?),businessKey,record.value(),newDate());doBusinessLogic(record);// 处理业务}catch(DuplicateKeyExceptione){log.info(重复消息跳过: {},businessKey);}}4.4 消费者防丢checklist配置/实践推荐方式防护的丢失场景提交方式手动提交commitSync自动提交时机早于处理完成offset提交时机消息全部处理完后提交后处理失败幂等消费Redis/DB 去重表提交前崩溃导致重复enable.auto.commitfalse自动提交抢占控制auto.offset.resetearliest(一般)/latest(如果可接受丢)新消费者组的行为关闭前提交commitSync()infinallyJVM关闭时未提交Rebalance时提交ConsumerRebalanceListener.onPartitionsRevokedRebalance时未提交五、全链路exactly-once方案把三端的防丢策略串起来就是exactly-once的完整方案【全链路 Exactly-Once 实现】 生产者端 Broker端 消费者端 ┌──────────────┐ ┌──────────────────┐ ┌──────────────────┐ │ acksall │ │ RF3 │ │ 手动提交offset │ │ 幂等Producer │ │ min.isr2 │ │ 先处理后提交 │ │ retriesMAX │ │ unclean✗ │ │ 幂等消费(去重) │ │ callback兜底 │ │ 滚动重启策略 │ │ RebalanceListener│ │ 事务Producer │ │ 多机房部署 │ │ finally提交offset│ └──────┬───────┘ └────────┬─────────┘ └────────┬─────────┘ │ │ │ ▼ ▼ ▼ 消息不会发丢 消息不会存丢 消息不会消丢 多 Topic 事务跨分区原子写入 ┌──────────────────────────────────────────────────────────┐ │ producer.beginTransaction(); │ │ producer.send(topicA, key, value); │ │ producer.send(topicB, key, value); │ │ // 同时写入消费者端的去重表 │ │ jdbcTemplate.update(INSERT INTO dedup_table ...); │ │ producer.commitTransaction(); // 原子提交 │ │ // 如果去重表写入失败整个事务回滚 │ └──────────────────────────────────────────────────────────┘六、线上故障复盘双十一幽灵订单事件6.1 故障现象时间2024年双十一凌晨2点现象客服收到用户投诉——“钱扣了订单没生成”技术表现支付回调消息在Kafka中丢失订单服务没收到付款通知6.2 排查过程【故障排查时间线】 00:30 流量开始上升每秒 8000 笔订单 01:00 Broker2 磁盘使用率 92%开始频繁触发 GC 01:15 Broker2 被剔除 ISRFollower跟不上 01:20 Controller 发现 Broker2 异常尝试 Leader 迁移 01:22 迁移期间Broker1 成为新 Leader 01:23 原 Broker2 上的部分消息未同步到 Broker1 → 丢失 排查发现 1. acks1 → Leader写入就返回Follower没确认 2. min.insync.replicas1默认值→ 没有防丢门槛 3. 原Leader宕机时有约200条消息没同步给Follower 4. 这些消息对应的订单钱扣了但订单没生成6.3 改进措施# 改进1提升acks级别 acksall # 从 ack1 改为 all # 改进2增加 min.insync.replicas min.insync.replicas2 # 从 1 改为 2 # 改进3禁止不干净的选举 unclean.leader.election.enablefalse # 已经设置没问题 # 改进4生产者回调兜底 # 发送失败时写入数据库队列定时重试// 改进5生产者端最坏情况兜底producer.send(record,(metadata,exception)-{if(exception!null){// 写入本地待重试文件failedMessageWriter.append(record);// 每分钟由定时任务扫描并重试// 超过1小时仍未成功 → 人工介入}});// 改进6消费者端幂等性// 支付回调消息中的 transaction_id 做去重// 即使重复消费也不会重复创建订单6.4 效果对比指标改进前改进后acks1allmin.isr12消息丢失双十一丢200条历经多次大促0丢失吞吐量影响-延迟5ms吞吐基本不变生产者兜底无文件兜底定时重试告警响应手动排查Prometheus指标自动告警本篇小结消息不丢失是三端联合工程缺一不可生产者端acksall 幂等 重试 callback兜底确保消息发出去并被确认Broker端RF3 min.isr2 uncleanfalse确保消息存住了且有多份备份消费者端手动提交 先处理后提交 幂等消费确保消息消费完且不重复记住口诀发的要确认、存的要多份、吃的要消化了再买单。下一篇我们将深入Kafka最核心的机制之一——副本同步与Leader选举彻底搞清楚那个Follower是怎么追上Leader的经典问题。上一篇【第61篇】Kafka可靠性保证全解析——acks、ISR、min.insync.replicas那点事下一篇【第63篇】Kafka副本机制深度解析——Leader选举是如何保证数据不丢的
