Netty 入站消息未释放导致 OOM

01结论先行：OOM 的根因

OutOfMemoryHandler 继承的是 ChannelInboundHandlerAdapter（而非 SimpleChannelInboundHandler），它重写了 channelRead，处理完入站消息后 既没有调用 ReferenceCountUtil.release(msg) 释放引用，也没有 ctx.fireChannelRead(msg) 把消息传下去。

而 Netty 的 ByteBuf / BinaryWebSocketFrame 都是引用计数（Reference Counted）对象： 引用计数永远不归零，底层缓冲就永远不被回收。同时该处理器还用 PooledByteBufAllocator(preferDirect = true) 申请堆外直接内存， 泄漏的直接内存不受堆 GC 管理，最终触发 OutOfMemoryError。

02问题背景：Netty 的引用计数模型

Netty 高性能的关键之一是零拷贝 + 引用计数。它的缓冲区 ByteBuf 以及包装它的 WebSocketFrame 都实现了 ReferenceCounted 接口，遵循这样一套规则：

• retain()：引用计数 refCnt + 1，表示「多一个使用者」。
• release()：引用计数 refCnt - 1，表示「我用完了」。
• 当 refCnt 归零，底层内存才真正被回收——池化分配器下归还给内存池复用，非池化则 free。

那么谁来负责 release？Netty 的通用约定是：

还需要区分两类内存：

本案例里两种泄漏叠加：入站消息本身泄漏（从注释堆栈看是堆内解压缓冲），处理器自己又申请了直接内存、异常时同样泄漏。

03详细内容：两段代码对比

3.1 第一段：SimpleChannelInboundHandler 的「正确姿势」

这是 Netty 框架源码 SimpleChannelInboundHandler.channelRead，它是释放逻辑的标准范式：

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    boolean release = true;
    try {
        if (acceptInboundMessage(msg)) {
            I imsg = (I) msg;
            channelRead0(ctx, imsg);        // 业务逻辑
        } else {
            release = false;                  // 不匹配 → 交给下游，下游负责释放
            ctx.fireChannelRead(msg);
        }
    } finally {
        if (autoRelease && release) {
            ReferenceCountUtil.release(msg);  // 关键：自动释放
        }
    }
}

要点：

• acceptInboundMessage(msg) 判断消息类型是否匹配泛型 I。
• 匹配 → 调 channelRead0 交给业务；不匹配 → fireChannelRead 下传，并把 release 置 false（因为下游接管了释放责任）。
• finally 里根据 autoRelease && release 决定是否 ReferenceCountUtil.release(msg)。autoRelease 是该类的字段，默认 true。

也就是说，只要你继承 SimpleChannelInboundHandler，用完消息后框架会自动帮你 release，无需手写。

3.2 第二段：OutOfMemoryHandler 的问题

再看出问题的自定义 handler（关键部分）：

public class OutOfMemoryHandler extends ChannelInboundHandlerAdapter {
    PooledByteBufAllocator allocator = new PooledByteBufAllocator(true);  // preferDirect = true

    private Object getBinaryWebSocketFrame(BinaryWebSocketFrame msg) {
        ByteBuf byteBuf = null;
        try {
            byte[] serviceTimeBytes = ("_" + System.currentTimeMillis()).getBytes();
            int capacity = msg.content().readableBytes() + serviceTimeBytes.length;
            byteBuf = allocator.directBuffer(capacity);     // 申请堆外直接内存
            byteBuf.writeBytes(msg.content());
            byteBuf.writeBytes(serviceTimeBytes);
            msg.content().resetReaderIndex();
            return new BinaryWebSocketFrame(byteBuf);
        } catch (Exception e) {
            return null;                             // ← 异常时 byteBuf 没释放！
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof BinaryWebSocketFrame) {
            BinaryWebSocketFrame frame = (BinaryWebSocketFrame) msg;
            BinaryUtils.toString(msg);
            ctx.channel().writeAndFlush(this.getBinaryWebSocketFrame(frame));
            // ← 既没有 ReferenceCountUtil.release(msg)，也没有 ctx.fireChannelRead(msg)
        }
    }
}

对比第一段代码，能清楚地看到它缺了什么：

注意：ChannelInboundHandlerAdapter.channelRead 的默认实现是 ctx.fireChannelRead(msg)（把消息往下传）。一旦你重写了它又不主动下传， 就切断了消息往下游流动的路径，而它又不会自动释放——于是消息就「卡」在这个 handler 里，泄漏了。

04OOM 是如何一步步形成的

把上面的问题串起来，泄漏链是这样的：

每条入站 BinaryWebSocketFrame
  └─ 到达 OutOfMemoryHandler.channelRead
       ├─ 被处理（toString + 回显一个带时间戳的新帧）
       ├─ 新帧 writeAndFlush 出去（出站方向 Netty 会在写完后释放，正常）
       └─ 入站帧 msg：
            ├─ 没有 release()  →  refCnt 始终 ≥ 1，永不归零
            ├─ 没有 fireChannelRead()  →  下游也无法代为释放
            └─ 底层 ByteBuf 永远回不到池 / 不被 free

高频消息（例如 EmbeddedChannel 压测循环 writeInbound）
  └─ 每个帧都泄漏一份缓冲 → 内存只增不减 → OOM

具体有两个层面的后果：

• 入站帧泄漏：每条消息的 ByteBuf 引用计数不归零，对象无法被释放； 如果是直接内存，还不受堆 GC 管理，MaxDirectMemorySize 很快被耗尽。
• 异常路径二次泄漏：getBinaryWebSocketFrame 里 allocator.directBuffer(capacity) 申请的直接内存，一旦中间抛异常，catch 直接 return null，刚申请还没写入帧的 byteBuf 也没被释放。

05心跳为何也被拖垮：IM 消息异常

除了内存泄漏，这个 handler 还会直接影响客户端心跳，导致 IM 消息异常。 根子还是同一个：重写 channelRead 后，只处理了 BinaryWebSocketFrame， 却没有 else 分支把其他帧往下传。于是它成了 pipeline 里的一个「黑洞」：

PingWebSocketFrame / PongWebSocketFrame / TextWebSocketFrame / CloseWebSocketFrame
  └─ 进入 OutOfMemoryHandler.channelRead
       └─ 不是 BinaryWebSocketFrame → 什么都不做
            ├─ 不 fireChannelRead → 下游（心跳/业务）永远收不到
            └─ 不 release           → 这些帧同样泄漏，雪上加霜

BinaryWebSocketFrame（真实业务消息 / 应用层心跳）
  └─ 被截住 → 不回传给业务，而是回显一帧「原内容 + _时间戳」

IM 系统里常见的两类心跳，都被这条代码破坏：

心跳一断，紧接着就是一串 IM 消息异常：

• 频繁断线重连：客户端因心跳超时反复断开、重连，连接状态抖动。
• 消息丢失：真实 IM 二进制消息被截住、不回传业务层，服务端根本没处理。
• 消息污染/异常：每条消息还被回显一帧「原内容 + _时间戳」，客户端可能解析失败、出现重复或乱序。
• 雪上加霜：被丢弃的 ping/pong 等帧同样没释放，与 OOM 泄漏叠加，服务端 GC 停顿又进一步拖慢心跳处理。

06泄漏检测报告解读

代码注释里贴的那段「Recent access records: / Created at:」堆栈， 正是 Netty ResourceLeakDetector（资源泄漏检测器）的典型报告。它会在某个 ByteBuf 被 GC 回收、却从未 release 时，打印出这个缓冲的创建点，帮助定位泄漏来源。

这段堆栈的关键几行：

PooledByteBufAllocator.newHeapBuffer(...)            // 泄漏的缓冲是「堆内」池化缓冲
AbstractByteBufAllocator.heapBuffer(...)
ZlibDecoder.prepareDecompressBuffer(...)       // 解压时申请的解压缓冲
JdkZlibDecoder.decode(...)
PerMessageDeflateDecoder.decode(...)           // WebSocket PerMessageDeflate 扩展
MessageToMessageDecoder.channelRead(...)
...
EmbeddedChannel.writeInbound(...)              // 测试/压测里写入入站消息

它说明了三件事：

• 泄漏发生在 WebSocket PerMessageDeflate（消息压缩扩展）的解压路径上——压缩消息解压产生的缓冲没有被释放。
• 这个被泄漏的缓冲正是入站帧的内容，最终流入 OutOfMemoryHandler 后被「截住」不再释放。
• 堆栈末尾的 EmbeddedChannel.writeInbound 说明是在单元测试 / 压测场景里用 EmbeddedChannel 反复写入入站消息触发的——这也解释了为什么消息量大到足以 OOM。

07修复方案

按「改动最小、最稳妥」排序，推荐下面三种，任选其一：

方案一：改成继承 SimpleChannelInboundHandler（最推荐）

public class OutOfMemoryHandler extends SimpleChannelInboundHandler<BinaryWebSocketFrame> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, BinaryWebSocketFrame msg) throws Exception {
        BinaryUtils.toString(msg);
        ctx.channel().writeAndFlush(this.getBinaryWebSocketFrame(msg));
        // 框架会在 channelRead0 返回后自动 release(msg)，无需手写
    }
}

方案二：手动 release（仍继承 Adapter）

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    boolean release = true;
    try {
        if (msg instanceof BinaryWebSocketFrame) {
            BinaryWebSocketFrame frame = (BinaryWebSocketFrame) msg;
            BinaryUtils.toString(msg);
            ctx.channel().writeAndFlush(this.getBinaryWebSocketFrame(frame));
        } else {
            release = false;
            ctx.fireChannelRead(msg);
        }
    } finally {
        if (release) {
            ReferenceCountUtil.release(msg);
        }
    }
}

方案三：修复 getBinaryWebSocketFrame 的异常泄漏

private Object getBinaryWebSocketFrame(BinaryWebSocketFrame msg) {
    ByteBuf byteBuf = null;
    try {
        byte[] serviceTimeBytes = ("_" + System.currentTimeMillis()).getBytes();
        int capacity = msg.content().readableBytes() + serviceTimeBytes.length;
        byteBuf = allocator.directBuffer(capacity);
        byteBuf.writeBytes(msg.content());
        byteBuf.writeBytes(serviceTimeBytes);
        msg.content().resetReaderIndex();
        return new BinaryWebSocketFrame(byteBuf);
    } catch (Exception e) {
        ReferenceCountUtil.release(byteBuf);   // 异常时也要释放已申请的直接内存
        return null;
    }
}

08总结归纳