企业官网建设流程全解析

前情回顾与本篇视角

前两篇讲了网络层的粘包和拆包——Selector 怎么从 TCP 字节流中还原出完整的响应帧。但还原出响应只是第一步，响应本身不携带请求体，只有一个 correlationId。Kafka 怎么知道这个响应对应哪个请求？怎么触发业务方的 callback？这就是 InFlightRequests 的工作。

InFlightRequests 是整个异步发送模型的核心基础设施：请求发出后暂存在这里，响应到达后按 correlationId 匹配，触发 callback，处理成功或失败。

01 InFlightRequests：暂存请求的结构

请求发出时，NetworkClient.doSend() 做三件事：分配自增的 correlationId、构造 ClientRequest、加入 InFlightRequests，然后写入通道。

InFlightRequests 结构：Map<Node, Deque<ClientRequest>>，按目标节点分组的双端队列。每个节点一个队列，FIFO 顺序。max.in.flight.requests.per.connection（默认5）限制每节点在途请求数。

ClientRequest 暂存的关键信息：

• correlationId：int32 自增 ID，请求唯一标识，响应中带回用于匹配
• callback：RequestCompletionHandler，响应到达时触发 onComplete(metadata, exception)
• createdTimeMs：请求发出时间戳，用于超时检测
• destination / request：目标节点和原始请求体，重试时重新发送

响应匹配流程：Selector.poll() 读到完整响应 → 从响应头取出 correlationId 和 source → 从该节点 InFlight 队首取出 ClientRequest（FIFO）→ 校验 correlationId 一致 → 解析响应体 → 触发 callback.onComplete()。

02 响应处理的三种结果

解析 ProduceResponse 后，检查每个 partition 的 errorCode，分三种情况处理。

成功（errorCode=0）：

• batch.done(baseOffset, timestamp)：设置偏移量和时间戳
• FutureRecordMetadata.complete()：业务线程的 future.get() 返回
• callback.onComplete(metadata, null)：触发用户回调，exception=null
• InFlight 计数减 1，可以发送新请求

可重试失败（RetriableException）：

• 如 NOT_LEADER_FOR_PARTITION、CORRUPT_MESSAGE 等临时错误
• 请求放回 InFlight 队首，下次 poll 重新发送
• 重试次数 +1，同时触发元数据刷新（leader 可能变了）
• 重试上限双条件：attempts < retries 且 now - createdTimeMs < request.timeout.ms

不可重试失败（NonRetriableException）：

• 如 INVALID_TOPIC_EXCEPTION、MESSAGE_TOO_LARGE 等配置或数据错误
• batch.done(exception)：Future 以异常完成
• callback.onComplete(null, e)：业务回调收到异常
• InFlight 计数减 1，不重试

关键参数：retries（默认 int 最大值）、retry.backoff.ms（默认100ms）、request.timeout.ms（默认30000ms）、max.in.flight（默认5）、enable.idempotence（开启后 max.in.flight≤5）、acks（0/1/all）。

03 暂存请求的生命周期与超时检测

请求从发出到完成，经历 T0→T1→T2→T3 四个阶段。但响应可能永远不到达——超时或连接断开时，Kafka 怎么处理暂存请求？

正常路径：T0 请求发出（createdTimeMs=now，加入 InFlight）→ T1 写入通道（selector.send()）→ T2 响应到达（NetworkReceive.done()，匹配 InFlight 队首）→ T3 触发 callback（batch.done()，Future 完成，从 InFlight 移除）。

超时检测：每轮 poll() 调用 maybeThrowTimeoutException()，遍历所有节点的 InFlight 队列。只检查队首（最老的请求）：now - createdTimeMs > request.timeout.ms → 超时。队首超时则整个队列的请求都失败（FIFO 顺序保证），触发 callback.onComplete(null, TimeoutException)。

连接断开：read 返回 -1 或 IOException → 连接断开。mute 该节点，所有在途请求标记失败（可重试的放回队首，不可重试的抛异常）。ConnectionStates 标记 DISCONNECTED，下次 poll 尝试重连。

为什么需要暂存状态？

• 请求和响应解耦：发送时不需要同步等待，Sender 可以继续发其他请求
• correlationId 匹配：响应不带请求体，靠 ID 找回原始请求和 callback
• 重试支持：失败时保留原始请求体，重新发送不需要业务层重构
• 超时检测：记录 createdTimeMs，每轮 poll 检查，避免请求永久挂起
• 流量控制：max.in.flight 限制在途请求数，防止 Broker 过载
• 顺序保证：同一节点队列 FIFO，响应按顺序匹配，保证消息顺序
• 幂等性：配合 PID + sequence number，重试不会重复写入
• 内存可控：在途请求数有上限，不会无限累积导致 OOM

写在最后

InFlightRequests 看起来只是个「存请求等响应」的队列，但它是整个异步发送模型的核心。correlationId 匹配、三种结果处理、双条件重试上限、超时检测、连接断开恢复——每一个边界情况都有对应的处理机制。

这也是读源码的价值：不是看用了什么复杂数据结构，而是看怎么用简单的队列 + 自增 ID 把异步通信的边界情况处理干净。下一篇看 Producer 端的完整异常处理体系——同步异常和异步异常怎么分类、怎么传递给业务方。