1. 先分清哪些丢消息是Kafka的锅1.1 丢了也分三种交付语义就是根本做消息中间件这些年被问得最多的一个问题就是Kafka会不会丢消息我的回答一直是会但得先看你说的是哪种丢。默认配置下的Kafka确实可能丢消息但经过合理配置它能在绝大多数场景下做到不掉数据。关键不在于Kafka能不能保证而在于你有没有把它配置成不丢的模式。要理解这件事先要搞懂分布式消息系统里常说的三种交付语义。第一种叫最多一次意思是消息最多被送到一次发送失败就直接放弃这种情况下丢消息是设计内的行为。第二种叫至少一次每条消息至少被送达一次系统会尽力重试代价是可能重复投递。第三种叫精确一次每条消息不多不少正好被处理一次这是最理想但也最难实现的目标。Kafka在默认参数下并不是一个保证不丢的系统它默认提供的是at least once也就是至少一次。可为什么现实里还是有人说Kafka把消息弄丢了呢大多数情况下是使用方式让这个至少一次退化成了最多一次比如生产者发送失败没有重试消费者在消息处理完成前就把偏移量提交了或者Broker端副本数不够、副本落后太多导致分区数据不一致。这几种情况叠加起来给人的体感就是Kafka把消息丢了。1.2 一句话定位消息到底丢在哪个环节排查丢消息问题要做的第一件事不是查代码而是先定位丢在哪个环节。一条消息从业务系统走到下游处理程序要经过三个接力区生产者发送到Broker、Broker内部存储与同步、消费者从Broker拉取处理。三个环节丢消息的原因完全不同排查手段也完全不同。环节正常情况是否丢什么情况下会丢核心参数生产端不丢有重试机制发送超时、缓冲区满、异常回调没处理acks、retries、buffer.memory、max.block.msBroker端多副本下一般不丢副本数量不足、ISR列表为空、落后副本被选为Leaderreplication.factor、min.insync.replicas、unclean.leader.election.enable消费端不丢但会重复自动提交offset、消息处理完成前发生Rebalanceenable.auto.commit、手动提交时机很多线上事故看起来是消费端没收到消息查到最后却是生产端已经把消息丢弃了还有一部分是消息其实没丢只是重复消费了给人造成了不对的感觉。我的建议是遇到疑似丢消息的问题先按这个表格把嫌疑范围缩小再逐个环节去看日志和监控指标不然很容易在错误的方向上反复折腾。后面的篇幅就按这三个环节逐一拆开讲。2. 生产端消息还没进Kafka就可能丢的那一半2.1 生产者三个关键参数acks、重试与幂等先明确一个事实Kafka的生产者并不会默默帮你做所有事。客户端把消息发出去之后是等Broker确认还是发完就当作成功完全取决于参数配置。最核心的就是acks参数它有三个值分别对应完全不同的安全级别。设成acks0表示发出去就不管了客户端不等任何确认Broker是不是真的收到全凭缘分。这个模式吞吐最高但Broker写盘失败、网络闪断都可能静默丢消息只适合日志采集、运行指标这类能容忍丢失的场景。设成acks1表示只要Leader分区把消息写进本地日志就算成功不用等Follower副本同步。这在大部分情况下不会丢但存在一个时间窗如果Leader刚好写完就宕机而其他Follower还没复制这条消息选举出新的Leader之后消息就没了。设成acksall也写作acks-1表示要等待所有处于ISR列表中的副本都写入成功客户端才收到成功响应这是最稳妥的选择。执行生产类业务时我一直把acksall作为底线。但只有acks还不够生产者的重试参数一样关键。默认的retries配置在较老版本里是0意味着发送失败不重试消息直接抛异常或静默丢弃。现代版本里retries改成了更大的值但仍然建议显式设置同时把delivery.timeout.ms调大一些给重试留够时间。还有一个容易忽略的点是幂等生产者设置enable.idempotencetrue。它解决的是重试导致重复的问题生产者给每条消息带上序列号Broker根据生产者ID和序列号做去重即使客户端因为网络超时重发了同一批消息Broker也只会保留一份。这个功能从Kafka 0.11版本开始可用新版本甚至默认就打开。实际经验是只要你的Producer客户端版本不老旧就直接开着几乎无副作用。2.2 踩坑实录缓冲区和超时导致的静默丢弃我见过太多明明发送成功却丢消息的案例最后发现问题根本不在Kafka而在生产者的发送方式和异常处理上。Kafka的生产者并不是来一条发一条而是先把消息写进内存缓冲区再批量发送这样设计是为了提高吞吐。这个过程有几个陷阱。第一个陷阱是缓冲区满了之后的行为。生产者有一个buffer.memory参数默认32MB所有待发送消息都在这个池子里排队。如果Broker响应慢发送速度赶不上写入速度缓冲区会被塞满。这时再调用send方法客户端不会无限等下去而是等到max.block.ms超时后抛异常。如果业务代码只做了log.error没有重发也没有落盘消息就悄悄没了。第二个陷阱是异步发送的回调没人处理。用Future方式调用send或者传了Callback之后只记录了日志但没人检查异常类型。有个真实的场景我印象很深某个团队把一条订单消息发到了不存在的Topic上Producer抛出的异常写进了日志但业务逻辑没有感知等反馈订单消息丢了来排查时日志里十几条报错清清楚楚。消息确实发了但也确实发失败了业务侧却以为发成功了。所以生产端的正确做法是把发送逻辑包装成带重试的公共组件捕获发送异常写本地重试表或者发到死信Topic而不是让开发者在每个业务里自己处理异常。序列化异常、分区器异常这类发生在客户端内部的问题消息同样到不了Broker只有统一在封装层拦截才能真正避免静默丢弃。3. Broker端副本、ISR与确认了以后再丢3.1 副本机制与ISR消息安全的最低保证如果说生产端丢消息多半是配置和使用问题那Broker端的丢消息就更多是架构设计问题。Kafka用多副本机制来保证高可用每个Topic的分区都会在多个Broker节点上保存副本其中一个是Leader负责读写请求其他是Follower被动同步数据。只要Leader挂了就从活着的副本里再选一个当新Leader。问题在于怎么定义活着的副本。Kafka引入了一个ISR概念全称是In-Sync Replicas也就是跟Leader保持同步的副本集合。Follower副本会持续向Leader拉取数据并写进自己的日志只要落后时间或落后条数在阈值内它就留在ISR里。如果某个Follower同步很慢甚至卡死它会被踢出ISR不再参与写入确认和Leader选举。为什么要引入ISR而不是让所有副本都算数因为真实的分布式环境里节点故障、网络延迟都是常态如果写一条消息必须等所有副本都返回只要有一个节点慢整个集群吞吐就会被拖垮。ISR是一种折中允许部分副本暂时落后但要求确认只发给能跟上的那一批。理解了这个机制你就明白生产者设acksall时等的其实是ISR内副本的确认而不是天底下所有副本的确认。3.2 高可用配置与丢消息的边界问题来了如果ISR里只剩下Leader自己acksall还有意义吗答案是没有它退化成与acks1差不多的情况。更危险的是如果连Leader也宕机了而ISR里一个其他副本都没有集群还允许选出一个落后很多的副本当新Leader那之前已确认成功的消息查找时就找不到了这部分消息就真的丢了。要堵住这个漏洞Broker端需要三个配置协同工作。第一个是replication.factor也就是副本总数生产环境至少设成3。第二个是min.insync.replicas表示一条消息要写入成功ISR里至少要有多少个副本推荐设2。第三个是unclean.leader.election.enable控制Leader宕机后是否允许从ISR之外的落后副本中选新Leader必须设成false。这三者配合起来的效果是Topic有3个副本写入时必须至少2个副本同步完成才返回成功当集群里只剩1个可用副本时生产端会收到NotEnoughReplicas异常写入失败而不是假成功。这个失败是有意为之宁可让业务暂时写入失败也不能让消息在成功确认之后蒸发。业务侧一旦收到这种异常可以把消息暂存到本地或者另一个队列等集群恢复之后再补发这比让用户发现数据对不上要强得多。3.3 容易被忽略的隐藏细节还有一个日常容易被忽略的点Kafka的消息虽然写进了日志但日志是先落在操作系统的Page Cache里再由系统异步刷到磁盘。如果只是进程崩溃Page Cache中的数据还在基本不会丢但如果整台物理机断电或者磁盘损坏没刷盘的数据就会跟着没。Kafka提供了log.flush.interval.messages等刷盘参数不过生产环境一般不靠调刷盘参数来保数据而是靠多副本来保证安全即使某一块磁盘坏了其他节点还留有副本。同时要区分一种业务上的丢消息那是消息过期被清理了。Kafka的Topic有retention机制默认保留7天或者按大小删除超过保留期的消息会被删除。这不是故障是数据生命周期管理。如果业务场景要求消息长期留存需要单独调大retention或者把数据同步到其他存储系统。4. 消费端offset提交时机和重复消费的老问题4.1 enable.auto.commit是新手最容易踩的坑生产端和Broker端都稳了消费端还有一个隐藏的丢消息入口就是偏移量提交时机。Kafka消费者消费完一条消息后需要提交offset告诉Broker这条我已经处理完了下次从下一条开始给我。这个提交如果做得太早就可能造成消息丢失。很多初学者用的都是默认配置enable.auto.committrueKafka每隔5秒自动提交一次offset。这个默认值对于演示环境没什么问题但在生产环境会埋雷。举个实际例子消费者poll了一批消息开始逐条处理假设处理每条消息要调用一次外部接口比较耗时。处理到一半5秒间隔到了自动提交把当前已经拉取到的最大offset提交了。可这时候还有几条消息正在处理中并没有真正处理完。如果进程恰好在这个时间点崩溃消费者重启后从已提交的offset继续拉取那几条没处理完但已经被提交offset的消息就永远没机会被读取了。这个问题的本质是自动提交并不关心每条消息是否被成功处理它只关心拉取到了哪里。它想当然地认为拉到的消息都会处理成功。一旦处理逻辑比提交间隔慢或者出现异常就会出现数据空洞。老版本的Kafka还有个隐藏属性auto.commit.interval.ms默认是5000毫秒而poll内部只要调用了就会反馈把这个问题放大。所以在生产环境我通常建议直接把enable.auto.commit设为false改成手动提交。4.2 所有处理都成功之后再提交offset改成手动提交之后下一个问题是提交时机。正确的模式很明确先处理完消息再提交offset中间不留任何先提交后处理的缝隙。伪代码大概是这样一个流程循环里先poll拉一批消息然后逐条执行业务逻辑全部执行成功之后再调用commitSync提交这批offset。如果业务逻辑中途抛出异常就不提交offset这样下次拉取会从头开始重新消费这批消息。代价是可能出现重复消费比如消息处理成功了业务结果已经写入数据库但提交offset的那一刻进程崩溃了重启后这条消息会被再处理一遍。重复总比丢失好配合消费端幂等这个模式就能把丢失问题彻底排除。这里要对幂等多说一句。消费端幂等的常见做法有三种利用数据库唯一约束去重、维护一张处理记录表记录已处理的消息ID、或者用Redis去重。每次处理前先查一下是否处理过处理时把消息ID和业务结果一起写入。这样做之后即使同样的消息被投递多次业务数据也不会出现重复。还要注意Rebalance场景。消费者组里的成员发生变更时会触发Rebalance分区会被重新分配。如果在Rebalance过程中还有消息没有处理完而offset又没有被正确提交这些消息会被重新分配给其他消费者实例但对应的offset可能没跟着交极端情况下也会出问题。解决方法是注册ConsumerRebalanceListener在onPartitionsRevoked回调里提交当前已处理完的偏移量再把还没处理的批次拿回来继续处理。5. 那精确一次到底靠什么实现5.1 幂等生产者先解决重试造成重复的问题聊到这一节很多人会问既然至少一次会重复最多一次会丢那Kafka能不能做到精确一次答案是可以但需要组合拳。第一拳是幂等生产者。前文提到过enable.idempotencetrue它的核心机制是生产者为每条消息分配自增序号Broker端会按生产者ID和序号做去重校验。重试发送时即使Socket层重传了同一批次Broker也能识别出这是重复数据不再重复写入。这个功能相当实用因为它解决了一个很隐蔽的重复来源网络超时导致的重试。没有幂等性时一条消息可能因为超时被重发而Broker其实已经写成功了重发的结果就是写入两条相同的数据。但是要理解幂等生产者的去重范围是同一个生产者会话、单个分区内部。如果Producer进程重启它的生产者ID会变之前的序列状态也不连续跨会话的重复就无法被去重。另外它只能保证写入不重复不能保证跨分区的原子性比如一次业务操作往两个Topic各写一条消息可能一个成功一个失败。要解决这类问题需要用到事务机制。5.2 事务API与read_committedKafka从0.11版本开始提供事务API它可以让一个生产者把多条消息发送到多个分区时要么整个事务成功要么全部回滚不会出现部分写入的状态。为了实现这一点生产者需要设置transactional.id并调用initTransactions、beginTransaction、commitTransaction这些方法。消费者这边也要配合设置isolation.levelread_committed这样它只会读到已经提交事务的消息不会读到事务中间状态的脏数据。这种模式特别适合同一个业务动作需要同时更新多个Topic的场景。比如下单服务既要写入订单Topic又要写入库存扣减Topic如果用事务包裹这两个写入就不会出现订单写成功、库存没扣的情况。Kafka Streams做流处理时也会大量用到这套机制。它把消费的offset和中间状态保存在同一份状态存储中用事务保证原子性从而实现端到端的精确一次处理。不过对大多数只做普通消息收发业务的团队来说直接操作事务API会比较重它带来的性能开销和配置复杂度都需要衡量。5.3 精确一次不等于不要钱为什么我建议大部分团队不要一上来就追求精确一次因为精确一次是有代价的。开启事务后生产者的吞吐会下降延迟也会增加事务协调带来了额外的网络交互。你不一定需要为每一条消息都付出这个成本。以我的经验判断先看业务本身库存扣减、转账、订单状态变更这类直接涉及资金或数量对账的场景才需要精确一次或者强幂等日志采集、点击流上报、监控指标这类允许少量重复和缺失的场景用at least once加偶尔手动补偿就够了。中间地带还可以选择另一条路线at least once 消费端幂等。这条路线实现简单运行稳定而且业务幂等能覆盖的不只是重复问题还能挡住Rebalance带来的重复消费一举多得。所以我的默认建议是开启幂等生产者关闭自动提交消费端加幂等逻辑。只有当你在同一事务里要维护多个分区的写入一致性时再去考虑引入事务API。这样做既能保住数据不丢又不会让系统复杂度失控。6. 实战排查清单与配置模板6.1 三个先看第一反应别猜排查Kafka丢消息最忌讳一上来就翻业务代码。我的习惯是先看三个地方基本能定位八九成。第一个看消费组的Lag指标。Lag表示消费者已经落后Leader最新消息多少条。如果Lag忽高忽低但始终有值说明消费速度跟不上还没到丢的程度如果Lag突然归零而业务数据缺失就要怀疑是不是offset被提前提交了。第二个看Broker的离线分区和副本同步情况。Kafka监控面板里有两个指标值得天天盯OfflinePartitions和UnderReplicatedPartitions。前者为0是最低要求表示没有分区失去Leader后者一旦大于0说明有副本没跟上此时如果发生Leader切换容易踩到数据不一致的坑。第三个看前端生产者和客户端日志里的异常比如NotLeaderForPartition、TimeoutException、NotEnoughReplicas、UnknownTopicOrPartition。每个异常指向的问题不同分区Leader变更、网络超时、副本数不足、Topic不存在。把这些报错对应到阶段排查范围就缩小了一大截。6.2 一份可以直接抄的配置模板多年代码实践下来我习惯在项目里维护一份标准配置模板新服务直接用省得每个团队各调各的。这份模板适用于绝大多数消息不能丢的业务场景。生产端配置示例Java语法风格acksall retries10 enable.idempotencetrue delivery.timeout.ms120000 request.timeout.ms30000 max.in.flight.requests.per.connection5 buffer.memory33554432 max.block.ms60000需要注意request.timeout.ms控制一次请求能等的时长delivery.timeout.ms是发送消息的总预算时间两者之间要预留足够余量。如果retries很大但delivery.timeout.ms没调大重试次数其实用不满因为总时间到点就放弃。Broker端核心配置replication.factor3 min.insync.replicas2 unclean.leader.election.enablefalseTopic创建时也要显式指定副本数而不是用默认值。很多集群默认副本数为1一旦节点坏掉就真的丢数据了。给Topic加副本数类似这样将replication-factor设为3。消费端配置enable.auto.commitfalse auto.offset.resetlatest isolation.levelread_committed这里有一个细节auto.offset.reset设成earliest还是latest很容易产生争议。如果消费者组是全新的没有提交过offsetearliest表示从最早的消息开始消费latest表示只消费新消息。生产环境如果追求不丢数据业务上能接受重放历史数据就选earliest如果不希望积压数据惊扰下游选latest。两种选择都可能造成看起来丢消息的现象关键是看业务预期。6.3 经验总结踩过的那几个坑每次讲配置总要补充几个真实案例因为参数是死的踩坑的场景才是活的。有一个系统原先用异步发送方式但发送失败只记录日志不重试。某天集群做滚动重启连续几分钟内发送超时日志里刷了一堆异常业务侧却没反应。等发现时那批消息已经在内存里被GC清掉了。后来改造了发送组件捕获异常后写入本地重发表后台线程扫描重发才彻底解决。还有一个系统消费端开着自动提交平时流量不大看不出问题。一次促销活动流量翻倍消费者处理时间明显变长正好赶上Rebalance一批消息被跳过去了。业务找到我时Lag看起来正常但数据已经对不上。改成手动提交后同一批代码就再没出现过这个问题。最典型的是某个集群当时开机时没关unclean选举。某天一个坏掉的节点重新加入集群因为它带的日志比当前Leader旧很多居然被选成了Leader导致整个分区回退了几个小时的数据。那次之后我检查任何集群的第一件事就是确认unclean.leader.election.enable为false。你可以接受短暂的服务不可用但不能接受数据在成功确认之后凭空消失这是整个方案里不能妥协的底线。最后再分享一个我个人的工作习惯每次上线前都会抽查生产者的回调日志、消费端的手动提交代码并写一个简单的消息巡检任务定期往系统里塞一条带业务唯一标识的探针消息让下游校验“是否收到、是否重复”。长期跑下来这套探针比任何监控面板都早发现问题。做一个消息不丢的系统靠的不是某一个神奇参数而是一整套可验证的机制。