1. 消息丢失这件事为什么大家争论的焦点都错了聊Kafka的消息不丢失网上百分之九十的文章都在讲生产端什么ack机制、retries参数、幂等生产者一套组合拳打完好像消息只要进了Broker就万事大吉。但真正在生产环境把数据搞丢的恰恰是消费端。我见过太多团队踩同一个坑生产端配置拉满Zookeeper和Broker的副本参数调得板板正正结果凌晨大促一压测消费者Rebalance一触发一批已提交offset但还没来得及处理完的消息直接人间蒸发。老板问起来Kafka监控面板看着一切正常因为Broker视角里这些消息确实没丢是消费者自己把已处理的位置标记了消息还在但业务数据已经对不上了。先说一个最反直觉的结论Kafka消费端丢消息绝大多数情况不是Kafka的错而是消费端代码逻辑的问题。很多人把消息不丢失等同于把enable.auto.commit设为false以为关掉自动提交就稳了结果手动提交的时机不对丢得更隐蔽。这篇文章不聊那些烂大街的生产端参数专门把消费端这条链路拆开揉碎从配置到代码再到Rebalance的极限场景把真正能落地的处理方式讲清楚。这套内容适合谁如果你是刚接触Kafka、正在写第一个消费者程序的初级开发你会明白为什么网上那套照着抄就完了的教程会在生产环境翻车如果你已经被线上丢消息折腾过几轮你会看到几个和你之前排查思路完全不同的排查方向。2. 消费端丢消息的底层链路到底是在哪个环节丢的要搞清楚怎么防丢先得搞清楚消息从Broker到业务数据库这条路上哪些位置存在丢的可能性。我习惯把消费端的消息链路拆成四个环节每一个环节的丢失原因和防护手段完全不同。2.1 环节一Broker推给消费者的时候Kafka的消费模型是消费者主动去Broker拉数据不是Broker推给消费者。这个拉的动作里有个非常经典的坑fetch.max.bytes和max.partition.fetch.bytes这两个参数控制单次拉取的消息字节数上限。你设置了1MBroker那边一条消息2M消费者压根拉不下来消息会一直停留在Broker里但消费端日志看起来就像没消息。这种情况严格说不算丢但对业务来说数据延迟等同于丢失。更隐蔽的是网络层面的问题。拉取过程中如果消费者和Broker之间的连接断开消息还在Broker里下次重新拉取能继续读。所以这个环节其实丢不了消息真正的问题出在拉取之后的处理。2.2 环节二消费者处理中宕机消息从Broker拉过来网络包已经进到消费者进程的内存里了。这个时候消费者宕机、被kill -9、或者整个Pod被调度到另一台机器上内存里的这批消息就没了。Broker里虽然还有这份数据但消费者恢复后如果直接继续拉新消息这批没处理完的就永远不会被重新消费。这个环节是消费端丢消息的大户。原因在于大多数人的消费代码长这样while (true) { ConsumerRecordsString, String records consumer.poll(1000); for (ConsumerRecordString, String record : records) { process(record); // 业务处理 } consumer.commitSync(); // 处理完后提交 }这段代码的问题在于万一process(record)处理到一半比如消息已经写进了数据库但还没执行到下一行代码或者消息在业务系统里产生了副作用但进程就崩了commitSync()根本不会执行。等消费者重启它会从上次提交的offset继续消费但那条已经产生业务副作用的记录会再次被拉取。2.3 环节三offset提交的时机黑洞offset提交在Kafka消费端里是双重性最强的机制——它既是防丢的武器又是丢消息的根源。先梳理一下offset是什么。每个分区里有一批消息每一条消息有一个递增的序号这个序号就是offset。消费者消费到哪一条了就把这个序号提交到Kafka的内部topic__consumer_offsets里。下次消费者重启它问Kafka我上次看到哪了Kafka告诉它一个offset它从这个位置继续拉。问题来了enable.auto.committrue时消费者会在后台每隔auto.commit.interval.ms默认5秒自动提交当前拉取到的最大offset。假设你poll了一批消息正在一条条处理处理到第3条时后台提交线程一看poll返回的最大offset是100直接提交了100。结果第3条处理失败或者第4条还没来得及处理就宕机重启后消费者从100继续消费第4条到第100条之间的消息全部跳过。2.4 环节四Rebalance时的重复消费与丢失这是整个消费端最容易被忽略的场景。消费者组里某个实例挂了、或者新增了一个消费者Kafka会触发Rebalance把分区重新分配。Rebalance发生的瞬间所有消费者停止消费各自提交当前offset然后重新分配分区。正常情况下每个消费者会在ConsumerRebalanceListener的onPartitionsRevoked回调里做一次同步提交确保分区被移交出去前已经把处理完的消息offset提交了。但如果你的代码用的是自动提交或者手动提交的间隔太大Rebalance发生时那些已经poll出来但还没处理完的消息offset还没提交分区就被分给另一个消费者了。新消费者从旧offset开始消费部分消息被重复处理如果旧消费者其实已经处理完了只是没来得及提交那新消费者只能重新处理一遍造成重复如果旧消费者还没处理完新消费者同样从头拉数据不会丢但会造成大面积的看起来像丢消息的重复或者遗漏。3. 解析真正能防丢的消费端配置组合网上很多教程讲消费端配置翻来覆去就是把enable.auto.commit设为false然后就没下文了。只关自动提交是不够的得理解每个参数在一条完整消费链路里到底起什么作用然后组合起来用。3.1 关闭自动提交是起点不是终点enable.auto.commitfalse确实是必须的一步但它解决的核心问题是把offset提交的主动权从定时器手里拿回来交到业务代码手里。自动提交的本质是我poll到了就算我处理完了这在绝大多数业务场景下都是错的。但关掉自动提交之后你马上会撞上下一个问题什么时候手动提交提交早了跟自动提交没区别提交晚了消息一多消费者提交offset的间隔太长一旦发生Rebalance大量消息被重复消费。我实际用的配置是这样的后续会解释每个值的考量enable.auto.commitfalse auto.offset.resetlatest max.poll.records500 max.poll.interval.ms600000 session.timeout.ms30000 heartbeat.interval.ms8000 isolation.levelread_committedmax.poll.records默认500很多人不知道这个参数对消息丢失的影响。假设你一条消息处理要100ms500条就是50秒如果你的max.poll.interval.ms保持默认的300秒也就是说消费者必须在300秒内再次调用poll否则Kafka认为这个消费者卡死了触发Rebalance。如果你的处理逻辑耗时超过这个窗口就会反复Rebalance每次Rebalance等于一次潜在的offset提交混乱消息丢失的风险指数级上升。所以我的建议是先估算单条消息的平均处理时间然后反推max.poll.records的合理值。单条处理100ms想让一次poll的消息在60秒内处理完max.poll.records就设为500到600之间同时把max.poll.interval.ms调到至少两倍于这个窗口。3.2 手动提交的正确姿势先处理再提交最后poll这里说的是一种在工程上被反复验证过的模式先poll一批数据逐条处理全部成功后再提交整个批次然后进入下一轮poll。落成代码是这样Properties props new Properties(); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500); while (true) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(1000)); if (records.isEmpty()) { continue; } // 先全部处理完 for (ConsumerRecordString, String record : records) { process(record); } // 全部处理成功后再提交 consumer.commitSync(); }这段逻辑的核心是批次内全成功才提交。只要commitSync()执行了就说明这一批500条全部处理完成。如果中途宕机最多重复消费这一批里的一部分但不会丢失任何一条。这个方案的代价是吞吐量。处理500条需要60秒那这60秒里消费者不会去拉新数据因为它在阻塞地处理当前批次。对于绝大多数业务场景这个吞吐量是够用的因为真正缺的不是吞吐是数据准确性。3.3 提交失败时怎么办重试不要吞异常commitSync()不是永远成功的。Broker在提交offset时如果遇到网络抖动、分区Leader切换可能抛异常。很多代码把提交包在try-catch里异常了打印一条日志就继续这是相当危险的。我的处理方式是提交失败必须重试重试依然失败就记录详细上下文并进入告警状态必要时退出进程让运维介入。因为offset提交失败意味着消费者实际已经处理完但Kafka不认账下次启动它会重复消费这些消息。重复消费一般还能兜底但如果你在process里做了非幂等操作重复消费就是事故。3.4 配合消息状态表做最终一致性兜底配置和代码做得再到位也只能保证Kafka不主动丢不能保证业务处理不失败。有没有一种方案让消息丢了也能捞回来有而且是在生产环境被验证过很多年的方案消费端本地状态表 定时补偿任务。思路是这样的在消费端维护一张消息处理状态表每条消息的处理状态包括待处理处理中已完成。消费流程变成拉取到一批消息先检查消息ID在本地状态表里的状态如果已完成直接跳过并提交offset如果待处理或处理中执行业务逻辑业务逻辑执行成功把状态标记为已完成再提交offset提交offset成功后删掉或归档状态记录。这套方案的核心价值在于offset提交的语义从我拉到过这条消息变成我确认处理完这条消息。当消费端重启或Rebalance发生时即使offset回到了很久以前消费者从旧位置重新拉消息时状态表里已完成的消息会被直接略过不会造成重复处理而待处理的消息则会被重新执行不会造成丢失。这个方案的代价是每消费一条消息就多一次状态表的读写综合性能损耗大约在10%到15%。但对订单、支付、库存这类数据准确性强相关的业务来说这个成本完全值得。4. 消息幂等丢失问题里最容易被忽略的一半前面讲了大量防丢手段但防丢和防重复是同一个问题的两面。消费者处理一批消息时宕机重启后从旧offset开始重新拉取那批消息里有一部分其实已经处理完了这时候它们会被再次处理。如果你的业务逻辑不是幂等的重复处理可能造成比丢消息还严重的后果。举个例子一个扣减库存的消息第一次执行扣了5件第二次执行又扣了5件消费者认为自己在补偿重复消费业务库存却已经在错误的路上越走越远。4.1 什么是消息幂等它和防丢是什么关系消息幂等指的是同一条消息无论被消费多少次产生的业务效果都和消费一次相同。实现方式通常有三种天然幂等更新操作本身就是幂等的比如update table set stockstock-5 where idxxx执行两次结果是扣10件注意这不幂等但如果改成update table set stock10 where idxxx执行两次结果一样这才是幂等。业务单据判重每次处理前查一下这笔订单是否已经处理过处理过就直接返回。乐观锁版本号消息里带一个版本号处理时对比数据库里的版本版本一致才执行更新否则忽略。4.2 用状态表同时解决防丢和幂等上一节提到的消息状态表天然就是一个幂等方案。因为状态表的存在同一条消息第二次进来时查状态表发现已完成直接跳过业务逻辑不会重复执行。这是防丢方案和幂等方案合二为一的状态不需要单独再搞一套去重逻辑。实操中的状态表DDL可以参考这样一个结构按业务量大小决定用MySQL还是RedisCREATE TABLE msg_process_status ( msg_id VARCHAR(64) PRIMARY KEY, status TINYINT NOT NULL COMMENT 0待处理 1处理中 2已完成, biz_key VARCHAR(128), create_time DATETIME DEFAULT CURRENT_TIMESTAMP, update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP );Redis实现的话用SET msg:proc:xxx 2配合过期时间效果类似。关键是写入和查询必须是原子的否则并发场景下会互相覆盖。工程上常见的做法是用SETNXEXPIRE或者用Lua脚本包住查询更新两步操作。5. 处理Rebalance这个隐藏杀手两个回调的实战写法Rebalance是消费端丢消息的高发场景也是网上资料写得不清楚的重灾区。Kafka提供的ConsumerRebalanceListener有两个回调onPartitionsRevoked和onPartitionsAssigned。字面意思分别是分区被收回和分区被分配大部分人知道要在这两个回调里做点事但具体做什么很多教程在瞎写。5.1 onPartitionsRevoked释放前的最后一道保险onPartitionsRevoked触发时机是消费者即将失去某些分区原因可能是消费者宕机、心跳超时、主动退出或者有新的消费者加入导致分区重分配。正确的做法是在这个回调里做一次同步提交把当前消费者已经处理完的消息offset提交掉让下一个接手分区的消费者能尽量从已处理的位置继续consumer.subscribe(Collections.singletonList(topic), new ConsumerRebalanceListener() { Override public void onPartitionsRevoked(CollectionTopicPartition partitions) { // 尽量把当前处理完的offset提交掉 consumer.commitSync(); // 如果内存里还有未处理完的消息可以考虑把状态表里它们的状态回退为待处理 pendingRecords.forEach(record - markAsPending(record)); } Override public void onPartitionsAssigned(CollectionTopicPartition partitions) { // 拿到新分区后根据状态表判断哪些消息已经处理过了从合适的位置开始消费 // 如果配合状态表方案这里可以执行跳过已完成消息的策略 } });注意commitSync()在onPartitionsRevoked里如果抛异常不要直接吞掉。虽然这个回调里的异常不会影响Rebalance本身但可能造成offset没提交完你可以先重试一次不行就记录下来交给状态表方案兜底。5.2 onPartitionsAssigned新主人开局时的反丢失手段onPartitionsAssigned触发时机是这个消费者拿到了新分区的管理权。很多极端情况下上一个消费者挂在处理到一半的状态offset还是老的新消费者会从老的位置重新拉一大堆消息。如果消费端没有幂等机制这些消息会被重复处理一轮。配合状态表方案后新消费者可以在onPartitionsAssigned之后做一次扫描把已经处理中或者已完成的位点确认一下。一个比较实用的做法是先查询状态表里这个分区的最大已完成消息ID然后从那个位置附近开始消费而不是从旧的offset开始。但这要求你的状态表和offset之间有清晰的对应关系实操起来比较复杂。更稳妥的做法是直接消费靠状态表的判重逻辑去过滤。5.3 静态消费组成员从源头减少Rebalance如果业务允许使用Kafka 2.3之后引入的静态消费组成员static membership能大幅降低Rebalance频率。配置group.instance.id后消费者实例在Session超时时间内报废重建不会触发Rebalance分区控制在当前实例手里offset也保持在当前实例里数据不经过交接环节丢失概率自然降低。代价是运维复杂度和分区分布的灵活性下降适合消费者数量比较稳定、不需要频繁弹性伸缩的场景。我维护的一个核心订单消费组就用了这个特性配上两个原始终端节点轮换近一年没有发生过因Rebalance导致的消费位点异常。6. 消费端性能瓶颈和防丢之间的平衡取舍防丢手段做全套性能一定受影响这是绕不开的话题。但受影响不等于不能用关键在于找到你的业务允许的平衡点。6.1 手动提交模式天生比自动提交慢但慢得值自动提交模式下poll一返回就触发后台定时提交消息处理几乎不等待任何额外操作吞吐量最高。改成手动提交每个批次必须等所有消息处理完才能commitSynccommitSync本身还有一次网络RTT吞吐量的损失来自这两处。我实测过的一个订单消费组自动提交模式峰值吞吐约2.3万条/秒换成处理完再同步提交后掉到1.8万条/秒损失大约20%。但对订单系统来说1.8万条/秒远高于业务峰值这20%换来的是一条消息都不丢的可信度这笔账很划算。6.2 异步提交真的不适合你除非你懂它的代价commitAsync()是多少人为了挽回那20%性能而踩的坑。异步提交的好处是不阻塞主线程丢消息的风险在于commitAsync提交的是当前拉取位点而不是当前处理完成位点。如果异步提交还在路上主线程又处理完了下一批消息并再次异步提交后一次提交把前一次覆盖掉前一次提交的offset比实际处理进度还超前一旦宕机中间那段消息直接丢失。如果实在要用异步提交唯一的正确姿势是在onPartitionsRevoked里补一次同步提交用同步提交兜底最终一致性consumer.commitAsync(new OffsetCommitCallback() { Override public void onComplete(MapTopicPartition, OffsetAndMetadata offsets, Exception exception) { if (exception ! null) { log.error(异步提交失败等待同步提交兜底, exception); } } }); // 在Rebalance时机或程序退出前强制同步提交 consumer.commitSync();6.3 增大max.poll.records谨慎对待很多人为了提升吞吐把max.poll.records调到2000、5000这个操作在消息体积大、处理耗时的场景下等于给自己埋雷。max.poll.interval.ms默认300秒如果一次poll的2000条消息处理时间超过300秒消费者会被判定失联触发Rebalance。Rebalance一旦频繁发生前面讲的所有防丢措施都可能被打乱。我的建议是不要盲目调大max.poll.records先测单条消息的平均处理耗时然后把max.poll.records * 单条耗时控制在max.poll.interval.ms的三分之一以内留出缓冲。7. 一个真实的生产事故复盘消息是怎么在我的消费组里人间蒸发的理论讲了这么多分享一个我实际经历过的线上事故。这个case现在回看挺低级但当时排查了整整一个下午因为它不是你打开监控就能一眼看出来的问题。7.1 事故现象某支付回调消费组某天下午延迟监控突然报警。消费者组日志没有任何异常没有Rebalance风暴没有连接错误消费者进程还在跑CPU也不高。但业务端反馈有一批订单一直等不到支付结果。7.2 排查过程第一步看消费组lag。Kafka自带的kafka-consumer-groups.sh显示lag为0也就是说消费者已经把topic里所有消息都拉完了。但业务数据库里那批订单确实没有状态更新。第二步怀疑是用户代码里某个分支逻辑出问题。翻代码发现支付回调处理逻辑是这样的拉取消息解析出订单号查数据库订单更新状态。问题代码长这样if (order null) { // 订单不存在直接提交offset以为不用处理 consumer.commitSync(); continue; }当时大家的判断是订单不存在就不需要处理但实际上那批订单是因为订单数据延迟落库导致消费端查不到订单直接跳过了。等订单落库完成offset已经被提交这条回调消息永远不会再被拉取。第三步确认根因后解决方案是给消费端增加订单不存在时进入重试队列的逻辑而不是直接提交offset跳过。同时把状态表方案里的待处理状态用起来让查不到订单的消息先置为待处理靠一个定时任务延迟三秒后重新投递到本地重试队列。7.3 这个case给到的经验消息不丢失不只是Kafka配置的事消费端业务逻辑里提前判死也会造成数据丢失。任何查不到就直接跳过的分支都要问自己一句话这条消息现在处理不了它以后会不会能处理如果会就不要急着提交offset放在重试机制里比放在Kafka里更可控。8. 从防丢策略到可观测性怎么知道丢没丢防丢手段再完备如果没有观测手段你永远不知道自己有没有丢。这就像保险箱做得再坚固没有报警器丢没丢东西你只能靠猜。8.1 消费位点监控是基础kafka-consumer-groups.sh --describe --group your-group能看每个分区的current-offset和log-end-offset两者差值就是lag。监控lag能发现消费慢和消费卡住但发现不了消息被错误跳过——被跳过导致的问题lag看是0。8.2 端到端数据比对才是防丢的终极手段给每条消息一个全局唯一ID在消息入口侧生成业务处理完成后再把消息ID和业务状态记下来比如订单ID 最后处理的消息ID存在业务表里。对账任务定期扫描源头生产记录和目标业务记录凡是业务表里的状态落后于生产端记录就说明消费端丢了或者偏了。这套对账方案的成本比状态表方案还高一点但它的价值在于消息不丢失不是靠配置来保证的是靠对账来证明的。生产环境里配置再完美的系统也会因为人为失误、依赖故障、数据异常出现意外端到端的对账是最底层的兜底。8.3 消费端的监控埋点要做到什么程度至少三个指标要有消费速率每秒消费多少条异常下降说明消费者在处理上卡壳。处理成功率每条消息是否有业务层面的成功/失败统计失败率异常升高说明业务逻辑出现系统性错误。重试次数分布进入重试机制的消息平均重试几次、最大重试几次重试次数集中在某个阈值说明消息内容可能有问题。说白了Kafka官方文档教的是配置调参生产经验教的是怎么让丢消息这件事从发生了才知道变成刚有苗头就看见。9. 结合热词补充那些围绕Kafka的高频疑问怎么理解搜索热词里有几个和消费端不丢消息强相关的问题顺带一起说了。9.1 Kafka有没有UI界面怎么可视化监控消费进度有常见的开源工具有Kafka Monitor、Kafka Eagle、Kafka UI后者是CNCF的项目对消费组的lag展示做得比较直观。在kafka-ui里能看到每个消费组的分区、当前offset、lag以及消费者的活跃状态。用它观察Rebalance事件比敲命令行舒服得多。安装很简单起一个Docker容器配置好Kafka的broker地址和Zookeeper地址就能用。但它只是看的层面真的要做端到端对账还是得靠自己的对账任务。9.2 消费端处理1M大消息该注意什么Kafka默认单条消息message.max.bytes是1M但这个值可以被调大。消费端处理大消息时max.poll.records不能按条数来估要按字节数来估。一条1M的消息拉500条就是500M这部分数据的反序列化和处理会造成明显的GC压力。我的做法是大消息场景下调小max.poll.records到50到100之间同时把消费者的堆内存预留好处理完的消息对象尽早释放。另外考虑把大消息的原始字节流先落到本地磁盘业务只处理一个文件路径引用能避免JVM OOM。9.3 Kafka消息延迟高是不是消费端配置的问题消息延迟高先看lag。lag大说明生产快、消费慢这时候优先查消费端的瓶颈是处理逻辑慢还是线程数不够。如果消费组只有一个消费者而topic有多个分区可以考虑增加消费端实例数让每个消费者各管一部分分区吞吐量线性提升。还要警惕一个隐蔽点某个消费者处理某一条消息特别慢比如外部接口超时重试十几次会拖慢整个partition的消费速度其他分区的消费者早消费完了它一个拖后腿的导致整个group的lag高居不下。解决方案是给每条消息的处理设置独立的超时阈值超时消息进重试队列不阻塞主循环。10. 我踩过几次坑之后留给我自己的检查清单做消费端防丢这么久我最后会把所有要点的检查清单贴在发布脚本里每次上线消费端代码前逐项过一遍enable.auto.commit是否确认设置为false手动提交是不是在批次全部处理成功之后执行的commitSync失败时是重试还是记录告警ConsumerRebalanceListener的onPartitionsRevoked里有没有同步提交兜底业务逻辑里有没有查不到就直接跳过的分支状态表/幂等机制是否覆盖了所有消费者实例max.poll.records乘以单条处理耗时是否远小于max.poll.interval.ms有没有端到端对账任务在定期核对消息是否完整消费。这份清单里每一项背后都是真实事故换来的教训。Kafka官方文档教的是机制生产环境教的是敬畏。消息不丢失从来不是某个参数调到最优就一劳永逸的事它是配置、代码、运维、监控四层防线全部到位之后的综合结果。