Kafka写数据入口：消息进了Broker之后交给谁-开发者社区

摘要：网络层把 Produce 请求解析完、RequestHandler 按 apiKey 路由之后，消息真正落盘的第一站是 ReplicaManager.appendRecords。这篇进入存储层，讲清楚写数据的调用链、appendRecords 内部五步、acks 三种分支，以及 acks=all 时请求为什么不会把业务线程堵死。
关键词：Kafka ReplicaManager、appendRecords、Produce写入、acks、DelayedProduce、Partition、Leader副本

从一个线上疑问说起

前面网络层系列讲完，有个问题一直悬着：RequestHandler 拿到 ProduceRequest 之后，消息到底是怎么写进磁盘的？很多人用了几年 Kafka，能说出"消息写到 Leader、Follower 同步"，但代码里这一步的入口在哪、acks=all 时线程在干嘛，并不清楚。

我早年排查过一次问题：生产环境 acks=all 的 Producer 偶发超时，但 Broker 的磁盘 IO 很闲。最后定位到是 ISR 副本同步慢导致请求在 Broker 端被"挂起"等待。当时如果理解了 ReplicaManager 的写入入口和 DelayedProduce 机制，这个问题一眼就能看明白。这篇就把这条链路讲透。

01 写数据的调用链

一次写消息从业务线程到磁盘，要穿过五层方法：

KafkaRequestHandler.run() └─ KafkaApis.handleProduceRequest() // apiKey=0 的处理入口 └─ ReplicaManager.appendRecords() // ★ 副本管理器写数据总入口 └─ Partition.appendRecordsToLeader() └─ Log.append() // 真正追加到日志 └─ LogSegment // 落到活跃段（页缓存）

KafkaApis 这一层负责"协议相关"的事：鉴权、解析 acks 和 timeout、把请求里的分区按"本机是不是 Leader"归类、准备好响应回调。真正和副本、日志打交道的，是 ReplicaManager。

ReplicaManager 是每个 Broker 一个的副本大管家，它最核心的两个方法就是appendRecords（写）和fetchMessages（读，Follower 同步也走它）。这篇聚焦写入口。

appendRecords 的方法签名信息量很大：

defappendRecords(timeout:Long,// 最多等多久（acks=all 时用）requiredAcks:Short,// 0 / 1 / -1internalTopicsAllowed:Boolean,// 能不能写内部topic（如__consumer_offsets）origin:AppendOrigin,// 客户端写入 / 副本同步entriesPerPartition:Map[TopicPartition,MemoryRecords],// 分区 -> 消息responseCallback:Map[TopicPartition,PartitionResponse]=>Unit,// 完成回调delayedProduceLock:Option[Lock]=None,recordValidationStatsCallback:...=Map.empty):Unit

注意最后两个参数：一个是回调，一个是延迟锁。这两个东西直接揭示了 Kafka 写入的异步设计——写完本地不一定就回调，acks=all 时回调会被延迟。

02 appendRecords 内部五步

进入 appendRecords 后，处理分五步：

第一步，前置校验。检查 Broker 当前是否还在 ISR 集合里、是否正在关闭。一个典型场景：Broker 刚被 Controller 踢出 ISR，此时如果还允许它以 Leader 身份写，就会产生不在 ISR 里的 Leader 写入，导致数据不一致。校验不过直接返回错误码，不做任何写入。

第二步，分区归类。请求里可能带了几十个分区，但当前 Broker 不一定都是它们的 Leader。appendRecords 把分区拆成本地（本机是 Leader，要真写）和远程（本机不是 Leader，返回 NOT_LEADER_OR_FOLLOWER 让客户端刷新元数据重试）两部分。

第三步，写本地日志。这是核心：

localProduceResults=appendToLocalLog(internalTopicsAllowed,origin,entriesPerPartition,requiredAcks)// 内部对每个分区：// partition.appendRecordsToLeader(records, origin, requiredAcks)// 返回：分配的起始offset、logAppendTime、错误码、日志起始offset

第四步，更新高位水位信息。记录本次写入后各 Leader 分区新的 LEO（Log End Offset，日志末端位移）。这个信息后面 Follower 来同步、HW（High Watermark，高水位）推进都要用。

第五步，按 requiredAcks 决定什么时候回调。这是最关键的分叉点，也是下面要讲的。

03 acks 三种分支：什么时候算"写成功"

requiredAcks决定 Producer 要等多久，appendRecords 的处理完全不同：

acks=0（不等）。消息写进本地日志缓冲后，KafkaApis 甚至不会构造正式响应（Producer 也不期待响应）。吞吐最高，但 Leader 宕机就可能丢，连写没写成功都不知道。

acks=1（等 Leader）。Leader 本地写成功，立刻执行 responseCallback 返回。不等任何 Follower。如果 Leader 写成功后、Follower 同步前宕机，新选出来的 Leader 没有这条消息，还是会丢——这是 acks=1 的固有窗口。

acks=-1 / all（等 ISR）。这是最复杂的分支。Leader 本地写完后不立即回调，而是创建一个 DelayedProduce 对象：

if(delayedProduceRequestRequired(requiredAcks,entriesPerPartition,localProduceResults)){val delayedProduce=newDelayedProduce(timeout,produceMetadata,this,responseCallback,delayedProduceLock)delayedProducePurgatory.tryCompleteElseWatch(delayedProduce,delayedProduceKeys)}

DelayedProduce 被丢进一个叫Purgatory（炼狱）的延迟操作管道里"挂起"。它在等什么？等所有 ISR 中的 Follower 通过 Fetch 请求把这条消息同步过去、各自的 LEO 推进到位、HW 抬升。全部追上，或者等到 timeout 超时，才触发回调返回响应。

04 存储层组件关系与延迟机制

理解写入绕不开几个嵌套的组件：

ReplicaManager：每个 Broker 一个，持有本 Broker 上所有 Partition
Partition：一个 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 等）

为什么要设计得这么绕？回到开头那个线上问题。如果 acks=all 时让 RequestHandler 线程同步等待所有 Follower，而 Follower 同步是毫秒到秒级、甚至可能卡住，那么默认 8 个业务线程很快会被全部占满，Broker 对所有请求失去响应。Purgatory 延迟机制让线程"挂起请求但不占用线程"——请求以对象形式躺在内存里等条件满足，业务线程立刻可以去处理下一个请求。这是用少量线程支撑高并发、同时保证强一致的关键设计。