摘要网络层把 Produce 请求解析完、RequestHandler 按 apiKey 路由之后消息真正落盘的第一站是 ReplicaManager.appendRecords。这篇进入存储层讲清楚写数据的调用链、appendRecords 内部五步、acks 三种分支以及 acksall 时请求为什么不会把业务线程堵死。关键词Kafka ReplicaManager、appendRecords、Produce写入、acks、DelayedProduce、Partition、Leader副本从一个线上疑问说起前面网络层系列讲完有个问题一直悬着RequestHandler 拿到 ProduceRequest 之后消息到底是怎么写进磁盘的很多人用了几年 Kafka能说出消息写到 Leader、Follower 同步但代码里这一步的入口在哪、acksall 时线程在干嘛并不清楚。我早年排查过一次问题生产环境 acksall 的 Producer 偶发超时但 Broker 的磁盘 IO 很闲。最后定位到是 ISR 副本同步慢导致请求在 Broker 端被挂起等待。当时如果理解了 ReplicaManager 的写入入口和 DelayedProduce 机制这个问题一眼就能看明白。这篇就把这条链路讲透。01 写数据的调用链一次写消息从业务线程到磁盘要穿过五层方法KafkaRequestHandler.run() └─ KafkaApis.handleProduceRequest() // apiKey0 的处理入口 └─ ReplicaManager.appendRecords() // ★ 副本管理器写数据总入口 └─ Partition.appendRecordsToLeader() └─ Log.append() // 真正追加到日志 └─ LogSegment // 落到活跃段页缓存KafkaApis 这一层负责协议相关的事鉴权、解析 acks 和 timeout、把请求里的分区按本机是不是 Leader归类、准备好响应回调。真正和副本、日志打交道的是 ReplicaManager。ReplicaManager 是每个 Broker 一个的副本大管家它最核心的两个方法就是appendRecords写和fetchMessages读Follower 同步也走它。这篇聚焦写入口。appendRecords 的方法签名信息量很大defappendRecords(timeout:Long,// 最多等多久acksall 时用requiredAcks:Short,// 0 / 1 / -1internalTopicsAllowed:Boolean,// 能不能写内部topic如__consumer_offsetsorigin:AppendOrigin,// 客户端写入 / 副本同步entriesPerPartition:Map[TopicPartition,MemoryRecords],// 分区 - 消息responseCallback:Map[TopicPartition,PartitionResponse]Unit,// 完成回调delayedProduceLock:Option[Lock]None,recordValidationStatsCallback:...Map.empty):Unit注意最后两个参数一个是回调一个是延迟锁。这两个东西直接揭示了 Kafka 写入的异步设计——写完本地不一定就回调acksall 时回调会被延迟。02 appendRecords 内部五步进入 appendRecords 后处理分五步第一步前置校验。检查 Broker 当前是否还在 ISR 集合里、是否正在关闭。一个典型场景Broker 刚被 Controller 踢出 ISR此时如果还允许它以 Leader 身份写就会产生不在 ISR 里的 Leader 写入导致数据不一致。校验不过直接返回错误码不做任何写入。第二步分区归类。请求里可能带了几十个分区但当前 Broker 不一定都是它们的 Leader。appendRecords 把分区拆成本地本机是 Leader要真写和远程本机不是 Leader返回 NOT_LEADER_OR_FOLLOWER 让客户端刷新元数据重试两部分。第三步写本地日志。这是核心localProduceResultsappendToLocalLog(internalTopicsAllowed,origin,entriesPerPartition,requiredAcks)// 内部对每个分区// partition.appendRecordsToLeader(records, origin, requiredAcks)// 返回分配的起始offset、logAppendTime、错误码、日志起始offset第四步更新高位水位信息。记录本次写入后各 Leader 分区新的 LEOLog End Offset日志末端位移。这个信息后面 Follower 来同步、HWHigh Watermark高水位推进都要用。第五步按 requiredAcks 决定什么时候回调。这是最关键的分叉点也是下面要讲的。03 acks 三种分支什么时候算写成功requiredAcks决定 Producer 要等多久appendRecords 的处理完全不同acks0不等。消息写进本地日志缓冲后KafkaApis 甚至不会构造正式响应Producer 也不期待响应。吞吐最高但 Leader 宕机就可能丢连写没写成功都不知道。acks1等 Leader。Leader 本地写成功立刻执行 responseCallback 返回。不等任何 Follower。如果 Leader 写成功后、Follower 同步前宕机新选出来的 Leader 没有这条消息还是会丢——这是 acks1 的固有窗口。acks-1 / all等 ISR。这是最复杂的分支。Leader 本地写完后不立即回调而是创建一个 DelayedProduce 对象if(delayedProduceRequestRequired(requiredAcks,entriesPerPartition,localProduceResults)){val delayedProducenewDelayedProduce(timeout,produceMetadata,this,responseCallback,delayedProduceLock)delayedProducePurgatory.tryCompleteElseWatch(delayedProduce,delayedProduceKeys)}DelayedProduce 被丢进一个叫Purgatory炼狱的延迟操作管道里挂起。它在等什么等所有 ISR 中的 Follower 通过 Fetch 请求把这条消息同步过去、各自的 LEO 推进到位、HW 抬升。全部追上或者等到 timeout 超时才触发回调返回响应。04 存储层组件关系与延迟机制理解写入绕不开几个嵌套的组件ReplicaManager每个 Broker 一个持有本 Broker 上所有 PartitionPartition一个 topic-partition 在本 Broker 的对象内部区分 Leader 副本和 Follower 副本Replica一个副本Leader 副本可读可写Follower 副本只负责被动同步Log一个副本对应一个日志由多个 LogSegment 组成.log 数据文件 .index 偏移索引 .timeindex 时间索引只有本机是 Leader 的分区才会走 appendToLocalLog本机是 Follower 的分区写请求会被拒绝数据是靠 Follower 主动 Fetch 拉过来的。DelayedProduce 的完整生命周期是① Leader 本地写完 → 创建 DelayedProduce绑定各分区新LEO和回调 ② 挂入 Purgatory开始计时request.timeout.ms默认30秒 ③ ISR 中的 Follower 陆续发 Fetch 请求同步数据推进各自 LEO ④ 每次 Fetch 处理完都会检查相关 DelayedProduce 是否满足条件 ⑤ ISR 全部追上HW推进→ 触发回调返回成功 或超时仍未追齐 → 返回带错误的响应NotEnoughReplicas 等为什么要设计得这么绕回到开头那个线上问题。如果 acksall 时让 RequestHandler 线程同步等待所有 Follower而 Follower 同步是毫秒到秒级、甚至可能卡住那么默认 8 个业务线程很快会被全部占满Broker 对所有请求失去响应。Purgatory 延迟机制让线程挂起请求但不占用线程——请求以对象形式躺在内存里等条件满足业务线程立刻可以去处理下一个请求。这是用少量线程支撑高并发、同时保证强一致的关键设计。写在最后ReplicaManager.appendRecords 这个入口浓缩了 Kafka 存储层最重要的两个设计思想一是本地写和副本等待解耦——能立即完成的本地追加快速做完不确定耗时的等 ISR交给延迟管道绝不阻塞业务线程。二是用 acks 把可靠性和延迟的选择权交给用户——0、1、all 三档对应三种明确的语义代码里用立即回调 vs DelayedProduce干净地实现了这个分叉。下一篇进入 Partition.appendRecordsToLeader 和 Log.append看消息是怎么分配 offset、怎么追加到具体 LogSegment 的。
