Netty 入站读源码:字节如何从 Socket 走到业务 Handler(源码系列第5篇)

Netty 的读路径不是一次 SocketChannel.read()，而是一轮受分配器和读策略控制的批量读取：EventLoop 触发 Unsafe，Allocator 决定 Buffer，Channel 执行系统调用，Pipeline 传播多次 channelRead，最后以 channelReadComplete 收尾。理解“批量、所有权、完成边界”才能正确处理半包、内存释放和读背压。

业务 Handler 看到的入口通常很简单：

public void channelRead(ChannelHandlerContext ctx, Object msg) {    Request request = (Request) msg;    handle(request);}

但在它之前，Netty 已经完成 Selector 事件分派、ByteBuf 分配、Socket 读取、读循环控制、异常判断和 Pipeline 查找。如果 Pipeline 里还有 Decoder，业务 Handler 甚至完全看不到原始 ByteBuf。

这条链中最重要的不是记住所有方法名，而是区分三个层次：

I/O 就绪：Selector 告诉 EventLoop 某 Channel 可读字节读取：Channel 把 Socket 数据写入 ByteBuf事件传播：Pipeline 把 ByteBuf 或解码后的消息交给 Handler

源码基线：

• Netty netty-4.1.135.Final
• Commit：f05f765d81460799c53123a207f665bf3b465171
• 主文件：AbstractNioByteChannel.java
• 关键方法：NioByteUnsafe.read()、doReadBytes()、fireChannelRead()
• 链接：https://github.com/netty/netty/blob/netty-4.1.135.Final/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java

这张流程图要按“读就绪不等于读完整请求”来理解。OP_READ 只是触发一轮受 RecvByteBufAllocator 控制的批量读取，可能多次分配 ByteBuf、多次调用 SocketChannel.read，也可能在 Decoder 处停下来等待半包。排查读不到业务消息时，应先确认字节是否进入 ByteBuf，再看 Pipeline 是否传播，最后看 Decoder 是否产出了完整消息。

1. OP_READ 只是“现在读可能有进展”

Worker Selector 返回 OP_READ 后，NioEventLoop.processSelectedKey() 调用：

unsafe.read();

此处 Unsafe 是 AbstractNioByteChannel.NioByteUnsafe。它不是业务扩展接口，而是 Channel 内部执行传输操作的入口。

可读事件并不保证一次 read 能拿到完整业务消息，也不保证只能读一次。TCP 是字节流，内核只告诉应用“接收缓冲区目前有数据或连接状态变化”。一次 read 可能得到半个请求、多个请求，甚至返回 -1 表示对端关闭。

因此 Netty 不能把一次 OP_READ 等同于一次 channelRead，更不能把它等同于一个业务请求。读循环和协议解码必须分层。

2. `NioByteUnsafe.read()` 组织整轮读取

核心源码可简化为：

public final void read() {    final ChannelConfig config = config();    if (shouldBreakReadReady(config)) {        clearReadPending();        return;    }    final ChannelPipeline pipeline = pipeline();    final ByteBufAllocator allocator = config.getAllocator();    final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();    allocHandle.reset(config);    ByteBuf byteBuf = null;    boolean close = false;    try {        do {            byteBuf = allocHandle.allocate(allocator);            allocHandle.lastBytesRead(doReadBytes(byteBuf));            if (allocHandle.lastBytesRead() <= 0) {                byteBuf.release();                byteBuf = null;                close = allocHandle.lastBytesRead() < 0;                if (close) {                    readPending = false;                }                break;            }            allocHandle.incMessagesRead(1);            readPending = false;            pipeline.fireChannelRead(byteBuf);            byteBuf = null;        } while (allocHandle.continueReading());        allocHandle.readComplete();        pipeline.fireChannelReadComplete();        if (close) {            closeOnRead(pipeline);        }    } catch (Throwable t) {        handleReadException(pipeline, byteBuf, t, close, allocHandle);    } finally {        if (!readPending && !config.isAutoRead()) {            removeReadOp();        }    }}

这段代码围绕四个变量运转：

• allocator：真正分配 ByteBuf。
• allocHandle：预测大小并决定是否继续读。
• byteBuf：本次系统调用的数据载体。
• pipeline：接管成功读取后的消息传播。

其中 byteBuf = null 是重要的所有权标志。成功 fireChannelRead(byteBuf) 后，当前方法不再负责释放，所有权进入 Pipeline；如果 read 没拿到数据或发生异常且 Buffer 未传播，则当前方法负责 release。

3. RecvByteBufAllocator 不只是决定 Buffer 大小

每次都分配固定 1KB 会浪费系统调用，大连接又可能频繁扩容；每次都分配 64KB，又会让大量小消息连接浪费直接内存。Netty 用 RecvByteBufAllocator.Handle 自适应或按策略选择容量。

Handle 记录本轮和历史读取信息：

• attemptedBytesRead：计划读取多少。
• lastBytesRead：实际读取多少。
• totalBytesRead：本轮累计字节。
• messagesRead：本轮产生多少 Buffer/消息。
• 是否达到每轮最大消息数。
• 是否可能还有更多数据。

continueReading() 不只看 Socket 是否还有数据，也要考虑公平性。如果一个高流量 Channel 每次就绪都无限读到内核缓冲为空，它可能长时间霸占 EventLoop，其他连接和任务得不到执行。Allocator Handle 因而兼顾吞吐与调度公平。

生产调优不能只盯 SO_RCVBUF。它是内核接收缓冲；RecvByteBufAllocator 控制用户态每次和每轮读取。二者处于不同层级。

4. `doReadBytes()` 才真正调用 JDK SocketChannel

AbstractNioByteChannel.doReadBytes() 大致是：

protected int doReadBytes(ByteBuf byteBuf) throws Exception {    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();    allocHandle.attemptedBytesRead(byteBuf.writableBytes());    return byteBuf.writeBytes(javaChannel(), allocHandle.attemptedBytesRead());}

ByteBuf 根据自身是堆内存还是直接内存，提供适配 NIO Channel 的写入方式。JDK SocketChannel 把数据写入 ByteBuf 的可写区域，成功后推进 writerIndex。

如果是 DirectByteBuf，底层能够更直接地用于 Native I/O；如果是 HeapByteBuf，可能需要临时直接缓冲或额外复制。现代 Netty 默认 allocator 的选择与平台能力有关，网络服务通常更关注池化直接内存。

lastBytesRead 的语义：

• 大于 0：读取成功，Buffer 有可读数据。
• 等于 0：当前没有读到数据，结束本轮。
• 小于 0：对端关闭了输入方向，需要进入关闭处理。

5. `fireChannelRead` 转移消息所有权

读取成功后：

pipeline.fireChannelRead(byteBuf);byteBuf = null;

Pipeline 从 HeadContext 后的第一个入站 Handler 开始寻找。每个 Handler 可以：

• 消费消息并停止传播。
• 转换消息后传播新对象。
• 原样调用 ctx.fireChannelRead(msg)。
• 抛出异常，让异常事件继续传播。

对于引用计数对象，传播责任与释放责任必须二选一。常见正确写法：

@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {    try {        ByteBuf buf = (ByteBuf) msg;        consume(buf);    } finally {        ReferenceCountUtil.release(msg);    }}

如果要继续传播：

ctx.fireChannelRead(msg); // 不在当前 Handler 释放

如果生成新消息并不再传播旧 Buffer，应释放旧 Buffer。SimpleChannelInboundHandler 会对匹配类型自动 release，但一旦消息被异步保存或转发，仍要理解 retain 与所有权。

6. 一次读循环可能触发多次 channelRead

假设内核缓冲里有 20KB 数据，而 allocator 每次分配 8KB。本轮可能发生：

read 8KB -> fireChannelRead(buf1)read 8KB -> fireChannelRead(buf2)read 4KB -> fireChannelRead(buf3)fireChannelReadComplete()

反过来，一次 8KB ByteBuf 里也可能含有十个业务请求。是否是完整消息由 Decoder 决定，不由读循环决定。

channelReadComplete 的准确含义是“本轮由一个读就绪事件驱动的读取批次结束”。它常用于统一 flush、统计本轮字节、触发下一次手动 read。它不是“一个请求处理完成”，也不是“连接暂时永远没有数据”。

Netty 的某些 Handler 会采用 read-complete 时统一 flush，以减少系统调用。这种 batching 可以提高吞吐，但也可能增加少量延迟，需要结合场景。

7. Decoder 在入站链中建立消息边界

最常见 Pipeline：

pipeline.addLast(new LengthFieldBasedFrameDecoder(...));pipeline.addLast(new ProtocolDecoder());pipeline.addLast(new BusinessHandler());

读路径先把任意分块的 ByteBuf 交给帧解码器。帧解码器维护 cumulation，把半包留到下一次 channelRead；当足够组成完整帧时，切出帧继续传播。ProtocolDecoder 再把帧转换为 Java 对象。

于是业务 Handler 看到的消息边界与 Socket read 边界完全解耦：

Socket read #1: 半个请求 ASocket read #2: A 的剩余部分 + 完整 B + 半个 CSocket read #3: C 的剩余部分Business Handler: A, B, C

这正是 Netty 编解码层的价值：Transport 只对字节和连接负责，Codec 对协议边界负责，业务对领域消息负责。

8. 对端关闭不等于立刻暴力 close

read 返回 -1 时，closeOnRead() 会根据 Channel 是否允许半关闭决定行为。TCP 支持输入方向关闭而输出方向仍可写。若配置 ALLOW_HALF_CLOSURE，Netty 可以触发半关闭相关事件；否则通常关闭整个 Channel。

常见生命周期顺序包括：

channelReadComplete  -> userEventTriggered(input shutdown) 或 channelInactive  -> channelUnregistered

协议层要明确半关闭语义。例如客户端不再发送请求，但服务端还需要发送剩余响应时，直接 close 可能截断输出；普通短请求协议则通常无需保留半关闭。

异常也不一定总是立即关闭。handleReadException 会先传播已读到的数据，再触发 readComplete 和 exceptionCaught，最后根据异常类型和连接状态决定关闭。这样可以避免本轮已经成功读取的消息被无故丢弃。

9. AUTO_READ 决定谁控制下一轮读取

默认 AUTO_READ=true，Channel Active 后持续关注 OP_READ，每次就绪自动执行上述读循环。这适合大多数服务，但在消费能力有限时可能不断把网络数据搬进用户态，形成内存积压。

关闭 AUTO_READ 后，业务需要显式：

ctx.read();

常见手动读取模式：

@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) {    if (businessCapacityAvailable()) {        ctx.read();    }}

这能把背压从业务处理能力传回 Socket 读取。内核接收缓冲填满后，TCP 窗口会自然收缩，对端发送速度下降。

但手动 read 很容易忘记恢复，表现为连接正常、无异常、就是不再收消息。排查时要检查 SelectionKey 是否仍关注 OP_READ、readPending 状态和业务分支是否遗漏 ctx.read()。

这张时序图把所有权变化单独画出来，很适合定位 ByteBuf 泄漏和提前释放。ByteBuf 从 Allocator 出来后，成功 fireChannelRead 就交给 Pipeline；Decoder 如果保留半包，要负责 cumulation 生命周期；业务 Handler 如果终止传播，就要消费并释放。凡是跨线程保存或异步处理，都必须先 retain 或使用 retained slice。

10. 入站读路径的典型故障

第一，EventLoop 读到了数据，但某个 Handler 没调用 fireChannelRead，后续业务永远收不到。可在 Pipeline 各节点增加短期日志定位传播断点。

第二，Decoder 每次都不推进 readerIndex，也不产出消息。ByteToMessageDecoder 会检测这种情况并避免死循环，但业务会表现为累积 Buffer 增长。检查长度字段、字节序和协议版本。

第三，Handler 吞掉 ByteBuf 却不 release，导致直接内存泄漏。开启适当级别的 ResourceLeakDetector，并检查最后消费点。

第四，业务线程异步使用 ByteBuf，却没有 retain。EventLoop 返回后上游自动 release，异步线程访问时抛 IllegalReferenceCountException。

第五，单连接持续大流量，读循环配置过激，其他 Channel 延迟升高。检查 maxMessagesPerRead、allocator 策略和 EventLoop 分布。

11. 读性能优化要先判断瓶颈层

读取慢可能在不同层：

层次	现象	优先检查
网络/内核	RTT 高、丢包、窗口小	TCP 指标、网卡、接收缓冲
EventLoop	一批连接一起抖	线程栈、Handler 耗时、任务队列
内存分配	Allocation/GC 压力	allocator、直接内存、池命中
解码	CPU 高、cumulation 大	协议复杂度、复制、帧长度
业务	channelRead 耗时长	阻塞调用、锁、线程池

简单扩大接收 Buffer 不一定提高吞吐；如果 Decoder CPU 已满，只会让积压更快进入用户态。简单增加 EventLoop 也不一定有效；如果瓶颈是共享业务池或单连接顺序处理，线程数增加帮助有限。

12. 总结：读路径的三个关键词是批量、边界、所有权

完整主链可以压缩为：

OP_READ  -> NioEventLoop  -> NioByteUnsafe.read  -> RecvByteBufAllocator.allocate  -> SocketChannel.read  -> pipeline.fireChannelRead  -> Decoder / Handler  -> fireChannelReadComplete

批量意味着一次就绪可能多次 read；边界意味着 Socket 分块不等于协议消息；所有权意味着 ByteBuf 传播与释放必须明确。

下一篇转向出站：writeAndFlush() 为什么不是一个动作，消息如何进入 ChannelOutboundBuffer，跨线程写如何回到 EventLoop，以及 Socket 一次写不完时为什么要注册 OP_WRITE。

参考资料

1. AbstractNioByteChannel.java：https://github.com/netty/netty/blob/netty-4.1.135.Final/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java
2. RecvByteBufAllocator.java：https://github.com/netty/netty/blob/netty-4.1.135.Final/transport/src/main/java/io/netty/channel/RecvByteBufAllocator.java
3. DefaultMaxMessagesRecvByteBufAllocator.java：https://github.com/netty/netty/blob/netty-4.1.135.Final/transport/src/main/java/io/netty/channel/DefaultMaxMessagesRecvByteBufAllocator.java
4. ByteBuf.java：https://github.com/netty/netty/blob/netty-4.1.135.Final/buffer/src/main/java/io/netty/buffer/ByteBuf.java