Kafka消息不丢实战：acks、ISR、位移提交与幂等消费全解析

1. 消息到底在哪个环节丢的先给可靠性画一张路线图做Kafka的人几乎都听过一句话“Kafka不丢消息”。可真到了生产环境因为消息丢失半夜被叫起来排查的人不在少数。为什么会这样因为“Kafka不丢消息”从来不是一个默认事实而是一个需要在生产端、Broker端、消费端三个环节分别做对配置、写对代码之后才能拿到的结果。任何一个环节用了默认值、图省事丢消息的隐患就已经埋下了。一条消息从业务进程产生到消费者真正把业务逻辑执行完中间要经过三个握手点生产者把消息发给Broker并等待确认Broker在副本之间同步数据消费者拉取消息后提交位移。这三个点分别对应三个核心原则搞懂它们后面所有配置和代码就都有了依据生产端没有收到Broker的确认就不能认为消息已经送达。Broker端副本没有同步完成就不能认为消息已经持久化。消费端业务逻辑没有处理成功就不能提交位移。这三句话听起来简单但实际落地时每一个环节都有大量细节。比如生产端你设置了acksall但没配min.insync.replicasBroker端照样可能在你只有一个副本存活时给你回“成功”消息随后就没了比如消费端你手动提交位移了但提交的时机放在“处理业务之前”那处理逻辑一旦抛异常这条消息就永远丢了。另外还需要分清一个概念Kafka在默认配置下提供的是At Least Once语义也就是“至少一次”不保证不重复。为了保证“不丢”我们接受可能出现的重复消息重复的问题由消费端幂等来解决。那种“既要完全不丢、又要完全不重复、还要性能拉满”的诉求在分布式系统里是不存在的。先把这一点想清楚后面很多设计决策就不会纠结了。2. 生产端拦截acks、重试与幂等生产者怎么配合生产端是消息丢失的第一道防线也是配置项最多、最容易出错的地方。很多人一上来就把acks设成all以为这样就万事大吉了其实这只是第一步。2.1 acksall和min.insync.replicas是组合拳acks参数有三个可选值先看它们各自意味着什么acks取值行为丢消息风险acks0生产者发完消息不等待任何确认立即认为发送成功极高网络抖动、Broker宕机、分区不可用都会无声无息地丢acks1Leader副本写入本地日志后即返回成功不等待Follower同步中等Leader在Follower完成同步前宕机选主后消息就没了acksall等待ISR中所有副本都写入成功后才返回低但还需要配合min.insync.replicas才有实际意义这里最容易被忽略的是acksall并不是“等所有副本写完”而是“等ISR集合里的副本写完”。ISR是保持同步的副本集合如果某个Follower落后太多会被踢出ISR。当ISR里只剩Leader一个副本时acksall实际上退化成acks1消息的安全保障已经没了但生产者完全感知不到。所以生产环境必须同时设置min.insync.replicas它的含义是“接受写入请求的副本最少要有几个在ISR里”。在Broker端的server.properties里加上min.insync.replicas2如果副本数为3这个值设置为2是比较常见的组合允许一个副本宕机或落后但至少还有一个Follower跟着Leader消息写入时不会出现“只有一个副本”的裸奔状态。当存活副本数小于2时Broker会拒绝写入请求生产者的send会返回异常这时候业务方会收到报错而不是收到“假装成功”的确认——宁可写入失败也不能悄悄丢消息。2.2 幂等生产者让重试不产生脏数据生产端的另一个陷阱是重试。Kafka的Producer内置了重试机制网络抖动、Broker短暂不可用时自动重发消息这是保证不丢的重要手段。但重试有一个副作用如果上一次请求实际已经写入了Broker只是响应超时生产者重试就会导致同一条消息被写入多次产生重复数据。Kafka从0.11版本开始支持幂等生产者解决的就是这个问题。开启方式很简单props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);开启后生产者会给每条消息附带一个递增的序列号Broker端会对同一个生产者、同一个分区的消息去重重复写入的请求会被忽略。需要注意开启幂等生产者时acks会被自动升级为all同时max.in.flight.requests.per.connection不能超过5这个值控制着单个连接上最多有多少个未确认的请求在途超过5会与幂等机制的序列号校验冲突。有了幂等生产者之后重试就变得安全了网络超时导致的重发、Leader切换导致的重发都不会再产生重复消息。这一点在压测环境里很容易验证——故意断网几秒重启Producer再对比写入总量开启幂等前后的差距非常明显。2.3 发送回调是唯一的“送达凭证”别只打日志写Kafka Producer的时候最常见的坑是在send方法后不关心返回结果。send是异步的它只是把消息放进了内部的发送缓冲区真正发出去是后台线程的事。如果你的业务代码是producer.send(record);那消息可能在缓冲区内、可能在网络途中、可能已经到达Broker但响应丢了你完全不知道。正确的做法是注册Callback回调在回调里检查异常producer.send(record, (metadata, exception) - { if (exception ! null) { // 写入失败需要走补偿流程记录日志、写入本地失败表、或者降级处理 handleSendFailure(record, exception); } else { // 发送成功可以记录发送轨迹 recordSendSuccess(metadata.topic(), metadata.partition(), metadata.offset()); } });不要只在回调里打印一行日志就完事。生产环境的经验是发送失败的消息必须有一个兜底存储。最简单实用的方案是搞一张本地消息表发送失败时把原始消息落库定时任务扫描这张表重新投递。Kafka自带的重试解决的是“临时性失败”那些Broker持续不可写、消息格式错误、权限问题之类的异常重试也救不回来必须靠业务侧的补偿机制兜底。这里还有一个隐藏参数值得关注delivery.timeout.ms它规定了消息从进入Producer缓冲到发送完成或失败的总时间上限。默认值一般是120秒如果你的重试次数配得特别大但delivery.timeout却很小那重试可能还没执行完就被超时截断了结果还是丢消息。所以调优的时候retries、retry.backoff.ms、delivery.timeout.ms这三个参数要放在一起看不要单独调某一个。3. Broker端存住副本机制、ISR与刷盘的取舍消息到了Broker是不是就安全了还不是。Broker层面要解决的核心问题是Leader挂了消息还在不在。3.1 副本数的意义Leader不是保险箱Kafka的一个分区在物理上会有多个副本其中一个充当Leader负责处理客户端的读写请求其余是Follower只负责从Leader拉取数据保持同步。消息丢失的一个典型场景是消息写入Leader后Follower还没来得及同步Leader突发宕机。此时Controller会从ISR集合里选出新的Leader而那条还没来得及同步的消息就永远丢失了。所以Topic的副本数直接影响可靠性。开发环境经常看到有人用默认的1副本创建Topic这在生产环境等于裸奔。建议Topic创建时统一设置bin/kafka-topics.sh --bootstrap-server kafka-1:9092 \ --create --topic order-events \ --partitions 12 --replication-factor 3创建完用describe命令检查每个分区的副本分配情况确认没有多个副本落在同一台机器上如果Kafka节点分布在多个机架最好还能配置rack感知让副本跨机架分布避免整机架断电导致全部副本同时失效。bin/kafka-topics.sh --describe --bootstrap-server kafka-1:9092 --topic order-events看输出的Leader和Replicas列。当一个副本同步落后时它的状态会从ISR列表里被移除此时该分区就处于“降级”状态。正常的describe输出里每个分区的ISR列表应该和Replicas列表一致或者只差少数副本。3.2 关闭unclean选举宁可短暂不可用也不让丢消息Kafka的副本同步有一个特殊情况Leader宕机后如果ISR里没有可用的副本了比如所有在ISR中的副本都挂了那是不是要选一个落后很多的Follower当新Leaderunclean.leader.election.enable这个参数就是控制这个行为的设为true允许从“不同步”的副本中选Leader服务可用性优先但必然丢消息。设为false不允许选不同步的副本宁可这个分区暂时不可用也不丢消息。对这个参数网上的讨论很多我的建议非常明确生产环境必须为false。unclean.leader.election.enablefalse原因很简单unclean选举是在“可用性”和“一致性”之间做抉择选了可用性就要接受消息丢失的后果。而Kafka本身通过ISR机制已经能处理大部分故障场景——ISR里的副本都还活着的情况下Leader宕机是可以正常选主的不需要unclean选举介入。真正触发unclean选举的场景是多个副本同时宕机这种极端情况这种时候宁可等副本恢复也不要选一个落后了一万条消息的Follower上台。如果业务真的在意可用性超过一致性建议从架构层面考虑而不是在Kafka这里放开口子。3.3 刷盘策略Kafka靠副本兜底不靠强迫落盘Kafka的消息写入Leader后实际上先落在操作系统的Page Cache里由操作系统异步刷到磁盘。很多人第一次知道这件事时会慌消息还在内存里机器突然断电不就丢了吗确实会丢但Kafka的设计哲学是单机断电这种极端情况交给副本机制来兜底而不是靠每台机器刷盘。log.flush.interval.messages和log.flush.interval.ms这两个参数控制的是“多长时间/多少条消息强制刷盘一次”。网上有些教程会教你把它们调成1让每条消息都立刻刷盘。这么做确实能降低单机断电导致的数据丢失风险但代价是写入性能断崖式下跌吞吐量可能掉一个数量级。而即便你做了强制刷盘也没法保证万无一失——存储硬件、文件系统层面的故障依然存在。正确的思路是接受“依赖Page Cache 副本机制”的架构设计用replication.factor3、min.insync.replicas2、acksall这套组合保证数据在多台机器上都有副本这样任何单台机器断电、磁盘损坏数据都还在另外两台机器上。刷盘参数保持默认就好不要乱调。Broker层面还有一个小细节controller节点的稳定性。Kafka的Controller负责分区Leader选举、副本分配等元数据操作Controller所在的Broker如果经常GC停顿或网络抖动会影响整个集群的副本同步状态。所以集群节点最好不要混跑其他重型应用JVM的堆内存也要根据分区数量合理设置避免频繁FullGC。4. 消费端守住最后一道关手动提交与幂等消费很多人认为消息不丢失是生产端和Broker端的事消费端顶多就是重复消费。这个认知在面试里还能糊弄过去在真实生产环境里会出大问题——消费端恰恰是最容易“丢消息”的地方而且丢了还特别难查。关键在于位移提交的时机。4.1 自动提交是最大的“假成功”Kafka消费者默认开启自动提交位移enable.auto.committrue auto.commit.interval.ms5000消费者每5秒会自动把当前拉取到的位移提交给Broker。问题来了如果你的业务处理耗时较长比如一条消息要处理2秒拉取完一批消息后刚处理到一半自动提交线程把位移提交了此时提交的是“已经拉取到的最大的位移”而不是“已经处理完成的位置”此刻消费者进程重启Kafka会认为这批消息已经消费完了从已提交的位移继续消费。那些“拉了但没处理完”的消息就这样被跳过了——对业务而言这就是丢消息。自动提交的本质是“拉取即消费”的假设它只适合那些“处理逻辑极快、丢了也无所谓”的场景比如日志采集、指标上报。但凡消息里带的是订单、支付、库存这类有业务含义的数据必须关掉自动提交enable.auto.commitfalse4.2 手动提交的正确姿势与重平衡的坑关掉自动提交之后手动提交的时机就变得非常重要。最常见的安全做法是“先业务处理再提交位移”props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 200); props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000); props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000); props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000); while (true) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(1000)); for (ConsumerRecordString, String record : records) { // 第一步处理业务例如写入订单库、调用下游系统 processMessage(record); } // 第二步这一批都处理成功了才提交位移 consumer.commitSync(); }这个模式保证的是只有处理成功的消息位移才会往前推进。如果处理中途抛异常位移不会提交下次重启会从上次提交的位置重新消费最多出现重复绝不会出现丢失。但这里要注意两个细节。第一commitSync是同步提交性能上会有一点损耗但可靠性最好建议默认使用。如果追求性能用commitAsync异步提交一定要在回调里处理失败情况并且注意异步提交的完成顺序和调用顺序并不保证一致进程关闭前必须补一次同步提交确保最终位移不丢失。第二max.poll.interval.ms这个参数。如果你的业务处理时间很长超过这个阈值默认5分钟消费者会被判定为“失联”触发重平衡分区会被分配给其他消费者。重平衡本身不会丢消息但如果你的位移提交逻辑写得不清晰重平衡期间的状态就很容易搞混。我见过一个案例消费者用线程池异步处理消息主线程poll到消息后丢给线程池就立刻提交位移了结果线程池还没处理完消费者发生重平衡那些正在处理中的消息对应的分区被分给另一个实例处理结果和位移就完全对不上了。这就是典型的“提交过早”。处理逻辑和提交位移必须保持严格的先后顺序这条线不能破。还有一个重平衡相关的坑处理消息时如果调用了外部接口而这个外部接口偶尔超时导致处理时间超过了max.poll.interval.ms消费者就会被踢出消费组。此时最稳妥的方案是先把处理时间优化控制在阈值以内如果实在做不到可以适当调大max.poll.interval.ms或者在处理长耗时任务时采用“先落库、再异步处理、按状态确认”的模式而不是在poll循环里同步阻塞。4.3 幂等消费设计接受重复消除重复的影响前面说过为了保证不丢我们接受“有可能重复”。那消费端就要有能力处理重复消息。很多业务同学一听到“要写幂等”就头疼其实幂等消费没有想象中那么复杂关键是找到业务的天然幂等键。以订单系统为例订单号就是天然的幂等键。消费端处理消息时先查一下订单表如果这个订单号已经存在且状态为“已处理”就直接返回否则才执行正式逻辑。这样即使同一条消息被重复投递对业务数据也不会产生叠加影响。更通用的做法是建一张消息消费记录表主键设为消息的唯一标识通常是消息的key如果没有key可以用topicpartitionoffset拼一个消费记录表msg_id主键、topic、partition、offset、业务主键、处理状态、处理时间消费流程就变成根据消息的msg_id查消费记录表如果已存在“处理成功”的记录直接提交位移。如果不存在开启本地事务同时写入消费记录和业务数据保证两者要么都成功、要么都回滚。事务提交后再提交位移。这种方式把“消息消费”和“业务数据写入”绑定在同一个数据库事务里天然解决了幂等和一致性问题。虽然对数据库有一点额外压力但从可靠性角度讲这是很多公司实际在用的成熟方案。Redis的setnx也可以做类似去重但因为Redis本身可能丢失数据如果对可靠性要求高还是建议落库。5. 如何证明没丢监控指标、Lag排查与对账机制配置都做了、代码也写了接下来的问题是你怎么知道线上真的没丢这一节不讲概念讲实操用哪些指标、哪些命令、哪些方法去验证消息的可靠性。5.1 Broker端核心指标UnderReplicatedPartitions不能一直大于0Broker端最关键的一个指标是UnderReplicatedPartitions它表示“正在同步中的副本数低于正常值”的分区数量。正常情况下这个值应该为0。如果你在监控面板上看到它持续大于0说明有分区副本同步落后或副本已经不在ISR里这是消息丢失的前兆——因为一旦此时Leader宕机那些没同步完的消息就没了。查看方式有两种。Kafka自带命令行# 查看某个Topic所有分区的ISR状态 bin/kafka-topics.sh --describe --bootstrap-server kafka-1:9092 --topic order-events以及通过JMX监控。生产环境建议用Prometheus Grafana收集Kafka的JMX指标重点盯这几个UnderReplicatedPartitions长期大于0要告警。IsrShrinksPerSec / IsrExpandsPerSecISR频繁收缩扩张说明有副本经常掉队。OfflinePartitionsCount分区Leader离线数大于0说明有分区不可用。RequestHandlerAvgIdlePercent请求处理线程的平均空闲率太低说明Broker负载过高。NetworkProcessorAvgIdlePercent网络线程空闲率类似。如果你不想从零搭监控也可以先用AKHQ这类Kafka可视化工具快速看一下集群状态。AKHQ能直观展示Topic列表、分区ISR状态、消费组Lag信息还能查看Connector任务状态排查问题时比命令行高效很多。不过可视化工具适合快速查看长期监控还是得落到指标系统里。5.2 消费Lag排查不是所有高Lag都是消费慢消费Lag即消费组当前消费到的位置和生产端最新写入位置之间的差值是判断消费端是否正常的最直观指标。bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 \ --group order-service --describe输出里看LAG列代表这个消费组还有多少条消息没消费完。Lag持续增长说明消费速度跟不上生产速度正常情况下应该趋近于一个较小值。这里要区分两种Lag高的原因。一种是消费组整体处理能力不足表现为所有分区Lag都在涨这时候加消费者实例数有可能解决前提是分区数大于消费者数。另一种是某个特定分区的Lag特别高其他分区正常这大概率是“消息卡在某个业务处理上了”——比如消费端处理某条消息时抛了异常或者走到了死信分支但位移一直没提交导致这个分区后续消息都堵住。排查这种问题先看消费组日志里有没有异常堆栈再看是不是有消息处理超时不要一上来就加实例。还有一个容易被忽略的点生产端和消费端的时间戳。如果生产者的消息时间戳和Broker服务器时间不一致通过时间维度去看Lag会产生误导。排查时尽量以“位移差值”为准不要以时间判断消费进度。5.3 对账表把“感觉没丢”变成“数据证明没丢”监控指标能告诉你“损坏风险”但有些消息丢失是“静默”的——配置错、代码逻辑错、位移提交时机错它不会产生任何告警只有业务数据对不上时才会被察觉到。这时候就需要对账机制兜底。我的做法是在生产端写消息时同时往一张消息轨迹表里插一条记录记录这条消息的topic、partition、offset、业务主键、发送状态。消费端消费成功后在轨迹表里更新状态为“已消费”。然后定期比如每小时跑一个对账任务扫描那些“已发送但长时间未消费”的消息人工确认是否真的丢了再决定是否需要重放。这套东西做起来不复杂但对业务数据的一致性非常有用。它解决的不只是“消息丢失”还包括“消息到达了但业务没处理”的情况。很多线上事故的定位最后都是靠这样的对账日志还原出问题的全貌。如果你的业务体量不大可以不用建完整的对账平台但至少要保证生产端和消费端都有落地的日志表为排查留一条路。6. 几个容易让人迷糊的问题事务、Exactly Once与Kafka和RabbitMQ的差异关于Kafka消息不丢失还有一些高频出现的疑问这里集中梳理一下。这些问题在面试里也经常被追问。6.1 Kafka的事务能解决“不丢不重”吗Kafka从0.11起支持事务API通过transactional.id配合事务消息可以实现跨分区原子写入。但很多人的理解是“开启了事务消息就不会丢也不会重复”。这个理解是有偏差的。Kafka事务解决的是“生产端写入多个分区的原子性”问题以及“Consume-Transform-Produce”这种流处理场景下消费位移和写入结果之间的原子性问题。如果你只是普通的“业务系统 - Kafka - 业务系统”这种消息中转开启事务并不会让消息可靠性变得更高反而会引入大量性能开销事务协调器的提交确认、事务日志的刷盘等。真正能让“生产到消费”整体具备Exactly Once语义的场景是Kafka Streams这类流处理框架它把“读取Kafka消息、计算结果、写回Kafka、提交位移”这四步包在一个事务里要么全成功要么全回滚。普通消费者自己写业务逻辑时要么用“事务幂等表”要么接受“至少一次业务幂等”不必非要用Kafka的事务API复杂度不成比例。6.2 Kafka和RabbitMQ在消息可靠性设计上有什么本质区别这个对比经常出现在面试题里但日常工作中理解它也有实际价值。两者在可靠性设计上的核心差异可以概括为Kafka以日志为核心消息有序存储在分区里消费者通过位移自主控制消费进度可靠性建立在“多副本 位移提交时机”上。它的设计假设是消费者可能会离线很久Broker不会为了消费者删除消息所以天然适合“消息延迟消费”“消息回溯”等场景。RabbitMQ以队列为核心消息投递给消费者后如果消费者未确认消息不会被删除unacked状态。它的可靠性更多依赖“手动ack机制”和“持久化交换机/队列/消息”三件套。所以如果你要用RabbitMQ保证不丢必须同时开启持久化而且消费者一定要手动ackKafka这边则不强制消费者立即确认重点反而在“位移提交不要早于业务完成”。同样是“消息不丢失”两边的设计思路和关键参数完全不同这也是为什么网上总有教程强调“不要用Kafka的自动提交就像不要用RabbitMQ的自动ack”是一样的道理。6.3 消费端遇到“毒丸消息”怎么办所谓毒丸消息就是单条消息本身内容有问题导致消费端每次处理都抛异常。如果不处理位移提交不了消费线程会一直卡在这条消息上整个分区都被堵死。这种场景不算“消息丢失”但它的危害和消息丢失一样大——后面的消息全部积压Lag直线上升。常规解法是把这类消息单独隔离捕获异常后判断是否是业务异常如果是将消息内容落库死信表然后手动提交位移继续消费后面的消息。但这有个前提你确认这条消息确实“必定处理不了”否则会掩盖真正的逻辑Bug。经验是第一次遇到异常时先重试几次重试仍失败的再进死信表死信表要有专门的告警不能让消息悄悄进去就完事了。我在实际项目中就是把“保存死信 更新消费记录 提交位移”串在同一个流程里等于是给消费链路加了一个旁路既不阻塞主流程又不放过问题消息。线上跑下来比一直卡住整个分区要省心得多。最后再分享一个经验做了几年Kafka相关的系统经历过半夜被叫起来查消息丢没丢的日子我的体会是消息可靠性这件事最怕的是“以为做好了”。acksall配了min.insync.replicas没配等于白配手动提交写了但放在处理逻辑前面等于没写监控面板搭了但只看CPU和内存关键的分区ISR状态一个都没盯等于白搭。给新团队的建议是先把本文提到的三个环节的配置和代码都过一遍然后在压测环境做一次破坏性演练——杀掉一个Broker节点重启消费端服务批量灌数据看看有没有消息丢。演练通过了再提“不丢消息”这回事也不迟。真正稳的Kafka使用方不是靠某一次配置调优而是靠一套“配置、代码、监控、对账”四位一体的机制在兜底。