Netty 事件循环与背压：一个 loop、一条 pipeline、一次引用计数
前面的 NIO 页讲清了 Selector 的就绪语义、部分读写与 OP_WRITE 开关；这一页进入 Netty 的具体实现：事件循环怎样跑 IO 与任务、pipeline 怎样组织编解码、ByteBuf 的引用计数在失败路径上怎样泄漏，以及"接收方读不动"怎样反压到发送方。
本页是 U67（#104）的 E04 权威正文。可运行结论来自 labs/netty-core 的 17 个测试：真实 Netty 4.2.19.Final（与 Spring Boot 4.0.9 BOM 的 netty.version 完全一致）、真实 NioEventLoopGroup、真实 loopback socket、真实 LengthFieldBasedFrameDecoder 与 ByteBuf 引用计数。源码摘录固定 netty/netty tag netty-4.2.19.Final（提交 e16031ea…），由 scripts/check_source_excerpts.py 在线复核。
一句话主线：事件循环是单线程的串行执行器：你放进 handler 的每一毫秒阻塞，都会变成同一 loop 上其他连接的排队；你少 release 的每一个 ByteBuf，都会变成一次泄漏。
一条连接在 Netty 里的生命周期
| 阶段 | 发生什么 | 失败会怎样 |
|---|---|---|
| 注册 | channel 注册到某个 EventLoop | 一个 channel 终生绑定同一个 loop |
| 就绪 | selector 返回就绪事件 | 就绪 ≠ 一整帧可读（NIO 页已讲） |
| 解码 | ByteToMessageDecoder 累积并切帧 | 半包保留、粘包拆分、坏长度抛异常 |
| 业务 | handler 在 loop 线程执行 | 阻塞代码拖垮同 loop 的其他连接 |
| 写出 | writeAndFlush 排队到出站方向 | 写失败以 future 完成，不抛异常 |
| 关闭 | close → channelInactive → 资源释放 | 未释放的 ByteBuf 泄漏 |
版本、工程与观察入口
固定 Netty 4.2.19.Final（Boot 4.0.9 BOM 管理的版本，未自行挑版本）、JDK 21（Liberica 21.0.11+11）、Maven 3.9.9、macOS arm64。全部实验在本机 loopback 与进程内执行，不需要外部服务。
公开入口在仓库根目录：
python3 labs/netty-core/run.py入口执行 clean test、核对 3 份 Surefire 报告（合计 17 个测试）、校验 17 条 U67 观察行，把环境、Netty 组件版本、源码 tag 与限制写入运行证据。
事件循环：单线程串行执行
线程归属
handler 里的 Thread.currentThread().getName() 是 nioEventLoopGroup-N-M——不是调用方线程，也不是业务线程池。一个 channel 终生绑定一个 EventLoop（NioEventLoop 的 register 在 loop 线程里执行注册）。真实观察：U67 eventLoopThread=PASS handlerOnLoop=true submitOnLoop=true。
loop 里跑什么：IO 与任务
Netty 4.2 的 SingleThreadIoEventLoop.run() 是循环的骨架（固定源码，SingleThreadIoEventLoop.java，201–232 行）：
protected void run() {
assert inEventLoop();
ioHandler.initialize();
do {
final int ioRatio = this.ioRatio;
final long taskQuantumNs;
// reset before calling runIo() which is responsible for calling reportActiveIoTime(...).
activeIoTimeNanos = -1;
runIo();
if (ioRatio == 100 || activeIoTimeNanos == -1) {
taskQuantumNs = maxTaskProcessingQuantumNs;
} else {
// Give tasks a budget proportional to the time just spent on IO, still bounded by the
// configured maximum so a burst of IO activity cannot starve the task queue indefinitely.
long quantum = Math.min(maxTaskProcessingQuantumNs, activeIoTimeNanos * (100 - ioRatio) / ioRatio);
if (quantum == 0) {
// Ensure we always use a "deadline"
taskQuantumNs = maxTaskProcessingQuantumNs;
} else {
taskQuantumNs = quantum;
}
}
if (isShuttingDown()) {
ioHandler.prepareToDestroy();
}
// Now run tasks for the computed amount of time before trying to run IO again.
runAllTasks(taskQuantumNs);
// We should continue with our loop until we either confirmed a shutdown or we can suspend it.
} while (!confirmShutdown() && !canSuspend());
}三个工程含义：
- IO 与任务在同一线程交替执行：
runIo()处理就绪事件，runAllTasks()执行提交到 loop 的任务。任务是 IO 之后的"第二轮"，不是并行。 - 任务有预算：
ioRatio（默认 100）与maxTaskProcessingQuantumNs共同限制单轮任务执行时间，防止任务风暴饿死 IO。 - handler 阻塞 = loop 阻塞：handler 里的
Thread.sleep会卡住整个runIo()阶段，同 loop 上其他连接的就绪事件得不到处理。真实观察：一条连接的 handler sleep 300ms，另一条连接的可见延迟被拉到 200ms 以上（U67 loopBlocking=PASS reads=2 elapsedMs=228 cause=Thread.sleep-in-handler）。
结论：业务 handler 里只做 CPU 微秒级的事；数据库、HTTP 调用、文件 IO 一律交给业务线程池，回调再 ctx.executor().execute(...) 回到 loop。
pipeline：入站从 head 到 tail
pipeline 是 handler 的双向链表：入站事件从 head 向 tail 传播，出站事件从 tail 向 head。真实观察：两个 handler 依次收到消息，顺序是 first;second;（U67 pipelineOrder=PASS order=first;second;）。handler 可以被动态移除，移除后不再收消息（U67 handlerRemoval=PASS delivered=1 stopped=true）。
编解码：半包、粘包与累积缓冲
长度域分帧
TCP 是字节流，没有消息边界。长度域分帧（LengthFieldBasedFrameDecoder）的规则：前 4 字节大端长度 + 负载；不完整就等待，多条粘在一起就逐条切。
真实观察：
- 粘包：三个帧一次到达，解码出三条（
U67 framerSticky=PASS frames=3）。 - 半包：帧头 + 一半负载先到，解码器不输出；补齐后输出一条（
U67 partialFrame=PASS halfWaited=true completed=true）。 - 逐字节到达：16 个字节分 16 次写入，自定义
ByteToMessageDecoder累积到完整帧才输出（U67 decoderAccumulate=PASS chunks=16 frames=1）。
累积语义的固定源码
ByteToMessageDecoder.callDecode 的核心循环（固定源码，ByteToMessageDecoder.java，464–506 行，节选）：
protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
try {
while (in.isReadable()) {
final int outSize = out.size();
if (outSize > 0) {
fireChannelRead(ctx, out, outSize);
out.clear();
// Check if this handler was removed before continuing with decoding.
// If it was removed, it is not safe to continue to operate on the buffer.
//
// See:
// - https://github.com/netty/netty/issues/4635
if (ctx.isRemoved()) {
break;
}
}
int oldInputLength = in.readableBytes();
decodeRemovalReentryProtection(ctx, in, out);
// Check if this handler was removed before continuing the loop.
// If it was removed, it is not safe to continue to operate on the buffer.
//
// See https://github.com/netty/netty/issues/1664
if (ctx.isRemoved()) {
break;
}
if (out.isEmpty()) {
if (oldInputLength == in.readableBytes()) {
break;
} else {
continue;
}
}
if (oldInputLength == in.readableBytes()) {
throw new DecoderException(
StringUtil.simpleClassName(getClass()) +
".decode() did not read anything but decoded a message.");
}四个要点：
- 循环条件
in.isReadable()：有字节就继续尝试解码；解码器返回但没消费字节（oldInputLength == in.readableBytes()）就break——这是"半包等待"的实现。 - "没读字节却产出消息"是错误：
decode()什么都不读却往 out 里放消息会抛DecoderException。这条校验防止解码器死循环。 - 累积缓冲由基类维护：
decode拿到的in是累积了多次读事件的复合缓冲，不是单次 read 的数据——所以你的解码器必须能处理"半个帧头"。 - handler 移除后不再操作缓冲：
ctx.isRemoved()检查避免在解码过程中被移除的 handler 继续操作 in。
工程含义：解码器里永远不要假设"一次 decode 拿到完整消息"；半包是常态，不是异常。
ByteBuf 与引用计数
计数语义
ByteBuf 是引用计数的：refCnt() 初始为 1，release() 减 1，归零时释放内存。真实观察：
- 释放后写入抛
IllegalReferenceCountException（U67 refCount=PASS released=0 useAfterRelease=rejected）。 - 二次释放同样被拒绝（
U67 doubleRelease=PASS rejected=true）。 retainedDuplicate()共享底层存储、计数 +1；释放 duplicate 后原 buffer 仍可读（U67 retainedDuplicate=PASS sharedStorage=true refCnt=2to1）。
释放的固定源码
AbstractReferenceCountedByteBuf 的 release 路径（固定源码，82–101 行）：
public boolean release() {
return handleRelease(RefCnt.release(refCnt));
}
@Override
public boolean release(int decrement) {
return handleRelease(RefCnt.release(refCnt, decrement));
}
private boolean handleRelease(boolean result) {
if (result) {
deallocate();
}
return result;
}
/**
* Called once {@link #refCnt()} is equals 0.
*/
protected abstract void deallocate();要点：release() 返回布尔值表示"是否归零"；归零才调用 deallocate()。没有"强制释放"——漏掉的 release 只能由 ResourceLeakDetector 在 GC 时报告（本实验未开启运行时检测，只基于 refCnt 语义断言：U67 leakSignal=PASS refCntBeforeRelease=1 detectorRole=refCnt）。
谁负责释放
规则简单但容易错：谁最后持有谁释放。
- 入站消息到达 tail 的最后一个 handler 后，由它负责
release()（或传给下游前retain()）。 SimpleChannelInboundHandler会自动 release；ChannelInboundHandlerAdapter不会——用错基类是头号泄漏来源。- 出站消息的释放责任在写出完成后（
writeAndFlush的 future 完成或失败）。
背压：谁读不动，谁就该被反压
现象
接收方停止读取时，事件走向是：内核接收缓冲填满 → TCP 窗口降为 0 → 发送方 socket 发送缓冲填满 → write() 阻塞。真实观察：接收方 autoRead=false、永不读取；客户端在独立线程里写 64MB，500ms 后仍阻塞在 write 上，只写出约 320–640KB 就卡住（U67 senderBackpressure=PASS writerBlockedAfter500ms=true bytesWritten=327680 readerNeverRead=true）。
Netty 侧的控制点
Channel.config().setAutoRead(false)：暂停读事件订阅，接收缓冲不再被消费。真实观察：关闭后服务端不再收到数据（U67 autoRead=PASS readsBeforeOff=1 readsAfterOff=1）。channel.isWritable()/channelWritabilityChanged：出站方向的背压信号——写队列超过高水位时为 false，业务据此暂停生产（不是 Netty 自动替你限流）。- 手动
read()：autoRead=false时显式拉取，实现"处理完一条再读下一条"的按需读取。
背压的边界
Netty 只提供"停止读"与"可写性"两个原语；是否反压到上游（暂停生产、降级、拒绝）是业务责任。把 autoRead=false 当作"自动限流"是误读：它只停止消费，队列与内存压力仍在（内核缓冲 + 出站写队列），最终仍要靠业务侧预算。
关闭与取消
关闭的三种路径
- 对端关闭：触发一次
channelInactive（U67 peerClose=PASS inactive=1）。 - 本端关闭：
close().sync()；关闭后的写入不抛异常，返回的 future 以失败完成，cause是StacklessClosedChannelException（U67 writeAfterClose=PASS closed=true failureType=StacklessClosedChannelException）。 - 进程退出：EventLoopGroup 的
shutdownGracefully有界等待任务完成。
关闭后写入的语义
写失败以 future 完成，不是同步异常——这是 Netty 的异步契约。业务不能依赖 try/catch 捕获写失败，必须挂 listener：
channel.writeAndFlush(msg).addListener(future -> {
if (!future.isSuccess()) {
// 处理失败：重试、记录、关闭
}
});观测与排障
几个关键观测点
- 事件循环线程名与数量（
nioEventLoopGroup-N-M）——handler 里打印线程名，确认没有跑错线程。 - 队列水位：
ChannelOutboundBuffer的待写字节数、isWritable变化。 - 引用计数：
refCnt异常与ResourceLeakDetector（生产开启 paranoid 级别）。 - 解码异常：
DecoderException、TooLongFrameException的堆栈。 - 关闭事件：
channelInactive的次数与顺序。
常见误读
- "Netty 是异步的所以快"：快来自单线程串行与零拷贝缓冲管理，不来自"异步"这个词；handler 阻塞照样慢。
- "用了池化 allocator 就没有 GC 压力"：池化减少分配，泄漏的 ByteBuf 仍然是泄漏。
- "autoRead=false 就是背压"：它只是停止读取；上游是否感知与处理是业务责任。
本页证明边界
| 已验证（真实 loopback / 进程内） | 未运行 |
|---|---|
| 事件循环线程归属、loop 阻塞连带延迟、pipeline 顺序与移除、粘包/半包/逐字节解码、refCnt 四种语义、池化 allocator、写阻塞背压、autoRead 关闭、对端关闭、关闭后写失败 | 真实远端网络/TLS/代理、epoll 与 kqueue 差异、吞吐与延迟压测、ResourceLeakDetector 运行时报告、出站编码器链与异常传播专项 |
本页证明的是 Netty 的机制与失败形态；容量与生产调优需要真实负载，不在本票。
实验读法
python3 labs/netty-core/run.pyCodecAndRefCountTest（8）：粘包/半包/逐字节解码、refCnt 释放/二次释放/retainedDuplicate/池化/泄漏信号。EventLoopTest（6）：线程归属、loop 阻塞、pipeline 顺序、背压、autoRead、channelInactive。CancelAndShutdownTest（3）：对端关闭、关闭后写失败、handler 移除。
闭卷验收
本页 21 道闭卷题收进了下方自测面板：先回忆，再展开参考答案与评分点，最后给三档自评。