Kafka响应准备：服务端发送响应前做了哪些事

摘要：RequestHandler 处理完请求之后，Response 不是直接写回客户端的——它要经过组装、入队、序列化、注册 OP_WRITE 等一系列准备。这篇把服务端发送响应前的完整准备工作讲透。

关键词：Kafka Response、responseQueue、NetworkSend、序列化、OP_WRITE、Selector

前情：处理完不等于发出去了

上一篇讲了 RequestHandler 处理完请求返回 Response。但这个 Response 还只是一个 Java 对象，离客户端收到差得远——它要变成字节、要找到对应的连接、要等 Socket 可写。

而且这里有个关键约束：业务线程不能直接写 Socket。Selector 不是线程安全的，RequestHandler 线程直接操作会和 Processor 线程冲突。

01 响应准备全景

整个准备链路分三层：

业务层（RequestHandler线程）
    ↓ sendResponse()
responseQueue（每个Processor一个）
    ↓ processNewResponses()
IO层（Processor线程）
    ↓ 序列化 + 注册OP_WRITE
Selector 真正写回

为什么每个 Processor 一个 responseQueue？ 因为请求从哪个 Processor 进来，响应就得回哪个 Processor——这个 Processor 持有对应的 Selector 和连接状态。RequestHandler 从 RequestContext 里拿到 processorId，把 Response 放进对应的队列。

// RequestChannel.sendResponse()
public void sendResponse(Response response) {
    int processorId = response.context().processorId();
    responseQueues.get(processorId).add(response);
}

02 Response 对象结构

一个 Response 对象包含四部分：

public class Response {
    private final RequestContext context;  // 原始请求上下文（找连接）
    private final ResponseHeader header;   // correlationId
    private final Struct body;             // 响应体
    private final Thunk onComplete;       // 发送完成回调
}

context 最重要——里面有 processorId 和 connectionId，决定这个响应要写回哪个连接。

header 只有一个关键字段 correlationId，和请求一一对应。客户端靠它匹配"哪个请求收到了响应"。

onComplete 是发送完成后的回调，用来更新统计指标（请求耗时、发送字节数等）。

03 序列化：对象变字节

Processor 从 responseQueue 取出 Response 后，第一件事是序列化成 ByteBuffer：

// 1. 计算总大小（header + body）
int size = response.sizeOf();

// 2. 分配 ByteBuffer，先写4字节length
ByteBuffer buffer = ByteBuffer.allocate(size + 4);
buffer.putInt(size);

// 3. 写 ResponseHeader
response.header().write(buffer);

// 4. 写 body
response.body().write(buffer);

// 5. 包装成 NetworkSend
NetworkSend send = new NetworkSend(connectionId, buffer);

为什么要先算大小再分配？ 因为 TCP 协议要求最前面写 4 字节 length，告诉对端这条消息多长。如果边写边算，length 字段就没法填——你不知道总长度。

这和前面讲的粘包拆包是配套的：发送端先写 length 再写内容，接收端先读 length 就知道要读多少字节。

04 注册 OP_WRITE：等可写才发

序列化好的 NetworkSend 不是立即写出去的，而是交给 Selector：

selector.send(send);

Selector 内部做两件事：

1. 把 NetworkSend 放进对应连接的待发送队列
2. 对这个 SocketChannel 注册 OP_WRITE 事件

为什么不直接写？ 因为 Socket 发送缓冲区可能满了——对端收得慢，缓冲区一满直接写会阻塞。注册 OP_WRITE 后，等缓冲区有空间了，Selector 的 poll() 会返回可写事件，Processor 才真正写出去。

这就是非阻塞 IO 的核心：永远不主动等，等事件通知。

05 响应的三种特殊去向

不是所有响应都能正常写回：

连接已断开：客户端在等响应的时候断了，Processor 找不到对应的 KChannel，直接丢弃响应并清理资源。

Fetch 长轮询延迟：Fetch 请求拉不到数据时不会立即返回，Response 会被暂存起来，等有新数据或者 max.wait.ms 超时了才入队。这是 Kafka 消费实时性的关键——消费者发一次 Fetch 就能等 500ms，不用疯狂轮询。

连接被 mute：前面讲过的背压机制，连接处于 mute 状态时响应也要排队等 unmute。

写在最后

响应准备的核心就一句话：业务线程只负责产出 Response，IO 线程负责真正发送，中间用 responseQueue 解耦。序列化先算大小再写 length、OP_WRITE 等可写事件，这些细节都是为了在高并发下不阻塞、不丢数据。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/zmj_0817/article/details/166257471