Netty 内存共享与释放机制

01结论先行：案例与核心结论

这个案例（WebSocketFrameHandler）完整展示了 Netty 的两大内存机制——内存共享（零拷贝） 与 释放机制（引用计数），以及一个典型的释放 bug。

在向多个接收方转发一条消息时，代码先调用 newFrame(msg) 深拷贝出一帧新消息 （分配了一块新的直接内存，refCnt = 1），紧接着却用 retainedDuplicate(msg) 把变量覆盖掉。于是那块刚分配的新 buffer 既没写入、也没释放——每转发一条消息、 每个接收方就泄漏一块直接内存。

顺带梳理出 Netty 完整的释放责任链，方便对照自查：

02问题背景：引用计数与内存共享

Netty 高性能的两大支柱是零拷贝与池化内存，两者都建立在同一个机制上： 引用计数（Reference Counting）。

ByteBuf 以及包装它的 ByteBufHolder（如 WebSocketFrame）都实现了 ReferenceCounted 接口：retain() 让计数 +1，release() 让计数 -1， 计数归零时才触发真正的释放。

「内存共享」意味着多个使用者共享同一块底层内存（零拷贝，不复制），代价是每个使用者都必须负责一次 release——这就是「谁最后使用谁释放」的由来。反之「深拷贝」（如 newFrame） 各自持有独立内存、互不干扰，但要额外分配和复制。

03释放机制：自动释放点与手动释放

3.1 引用计数模型与释放流程

一次 release() 把计数减到 0 后，底层内存的归宿取决于分配方式：

• 池化内存：归还到 PoolThreadCache（线程私有缓存，下次复用最快），再根据使用情况回收至 Chunk。
• huge 内存：直接彻底销毁。

而 Chunk 本身也有生命周期：释放时根据使用率把 Chunk 移动到合适的 PoolChunkList，使用率低于阈值时销毁整个 Chunk，把内存还给操作系统。

3.2 读方向：TailContext 兜底释放

入站消息沿 Head → Tail 依次经过 ChannelInboundHandler。如果消息一直没人处理、走到 pipeline 尾部，TailContext 会兜底释放它：

final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        onUnhandledInboundMessage(msg);
    }
}

protected void onUnhandledInboundMessage(Object msg) {
    try {
        logger.debug("Discarded inbound message {} that reached at the tail of the pipeline.", msg);
    } finally {
        ReferenceCountUtil.release(msg);   // 未处理消息在此释放
    }
}

3.3 写方向：HeadContext flush 后释放

出站消息沿 Tail → Head 依次经过 ChannelOutboundHandler。写请求最终到达 HeadContext，真正把数据刷到 socket 后释放 buffer：

private void invokeWriteAndFlush(Object msg, ChannelPromise promise) {
    if (invokeHandler()) {
        invokeWrite0(msg, promise);
        // invokeFlush0() 这里释放过程 refCnt → 0
        invokeFlush0();
    }
}

private void invokeFlush0() {
    ((ChannelOutboundHandler) handler()).flush(this);
}

@Override
public void flush(ChannelHandlerContext ctx) throws Exception {
    unsafe.flush();   // unsafe flush 后被释放
}

3.4 SimpleChannelInboundHandler 自动释放

前面的文档已详述：继承 SimpleChannelInboundHandler 后，channelRead0 返回时框架会 在 finally 里自动 release 入站消息（autoRelease = true）。 本案例的 WebSocketFrameHandler 正是这么做的，所以入站帧本身不会泄漏——泄漏点在它内部手动创建的那块新 buffer。

3.5 手动释放：ReferenceCountUtil

public final class ReferenceCountUtil {
    public static boolean release(Object msg, int decrement) {
        if (msg instanceof ReferenceCounted) {
            return ((ReferenceCounted) msg).release(decrement);
        }
        return false;
    }

    public static void safeRelease(Object msg) {
        try {
            release(msg);
        } catch (Throwable t) {
            logger.warn("Failed to release a message: {}", msg, t);
        }
    }
}

ReferenceCounted 对象初始化时引用计数默认是 1——表示「当前持有者」占用一份引用， 用完必须 release 一次把这份引用还掉。

04内存共享：零拷贝 vs 深拷贝

发送一份消息给多个接收方时，有两条路可选，取舍完全不同：

// 零拷贝共享：复制的是「视图」，不是内存
private static Object retainedDuplicate(Object message) {
    if (message instanceof ByteBuf) {
        return ((ByteBuf) message).retainedDuplicate();
    } else {
        return message instanceof ByteBufHolder
            ? ((ByteBufHolder) message).retainedDuplicate()   // refCnt + 1
            : ReferenceCountUtil.retain(message);
    }
}

05案例逐行分析：泄漏点定位

关键代码（channelRead0 的二进制分支与 writeAndFlush）：

private BinaryWebSocketFrame newFrame(BinaryWebSocketFrame msg) {
    ByteBuf byteBuf = null;
    try {
        byte[] serviceTimeBytes = ("_" + System.currentTimeMillis()).getBytes();
        int capacity = msg.content().readableBytes() + serviceTimeBytes.length;
        byteBuf = ByteBufAllocator.DEFAULT.directBuffer(capacity);   // 深拷贝：新分配直接内存
        byteBuf.writeBytes(msg.content());
        byteBuf.writeBytes(serviceTimeBytes);
        msg.content().resetReaderIndex();
        return new BinaryWebSocketFrame(byteBuf);                    // refCnt = 1
    } catch (Exception e) {
        if (byteBuf != null) ReferenceCountUtil.release(byteBuf);       // 异常路径已修复
        throw e;
    }
}

private void writeAndFlush(..., BinaryWebSocketFrame msg, List<Channel> channels) {
    for (Channel channel : channels) {
        ...
        BinaryWebSocketFrame webSocketFrame = this.newFrame(msg);   // 分配新 buffer，refCnt=1
        webSocketFrame = (BinaryWebSocketFrame) retainedDuplicate(msg);   // 覆盖！新 buffer 被丢弃 → 泄漏
        channel.writeAndFlush(webSocketFrame, promise);                    // 共享帧写出，flush 后释放
    }
}

引用计数的完整流转如下：

入站：WebSocketFrameHandler（SimpleChannelInboundHandler<WebSocketFrame>）
  channelRead0 收到 BinaryWebSocketFrame msg，msg.refCnt = 1

  循环每个接收方 channel（共 N 个）：
    ├─ newFrame(msg)            → 新分配独立 direct buffer，refCnt=1  ← 被覆盖，永不释放 ✗
    ├─ retainedDuplicate(msg)   → 共享 msg 底层内存，msg.refCnt + 1
    └─ channel.writeAndFlush()  → flush 后 Netty 释放该共享帧，msg.refCnt - 1

  channelRead0 返回 → SimpleChannelInboundHandler 自动 release(msg)，msg.refCnt - 1

最终：
  msg 的共享内存：1 + N - N - 1 = 0  ✓ 正确释放
  newFrame 的独立内存：每个 channel 泄漏 1 块  ✗ 永不释放（泄漏量 = 消息数 × N）

正确写法

要么用「拷贝」、要么用「共享」，二选一，别混用。这里因为要追加时间戳，用拷贝即可：

for (Channel channel : channels) {
    if (channel == null || !channel.isOpen() || !channel.isActive()) {
        if (chatType == Chat.CHAT_TYPE_1_2_1) {
            ctx.channel().writeAndFlush(new TextWebSocketFrame(Instruction.OFFLINE));
        }
        continue;
    }
    // 每个接收方一份独立拷贝，写出去后所有权移交出站管线，flush 后由 HeadContext 释放
    BinaryWebSocketFrame webSocketFrame = this.newFrame(msg);
    channel.writeAndFlush(webSocketFrame, promise);
}

这样 newFrame 返回的 refCnt=1 新帧被 writeAndFlush 写出去， 释放责任移交给出站管线（HeadContext flush 后释放），既不用手动 release， 也不需要 retainedDuplicate。

06泄露监控：ResourceLeakDetector

Netty 用 ResourceLeakDetector 检测内存泄漏，但要注意它的定位：只告警、不释放。

• 跟踪机制：创建 ByteBuf 时，Netty 会把它包装成 LeakAwareByteBuf （SimpleLeakAwareByteBuf / AdvancedLeakAwareByteBuf），并创建一个 WeakReference 指向原始 ByteBuf，注册到引用链中跟踪。
• 判定泄露：对象被 GC 回收时，若从未调用 release()（refCnt 未归零）， 该弱引用会进入 ReferenceQueue，从而判定为泄漏。
• 只记录不释放：clearRefQueue() 轮询队列并调用 ref.dispose()， 这一步只是记录泄漏堆栈（就是我们看到的那份「Recent access records / Created at」报告）， 并不会释放底层直接内存。

private static final class DefaultResourceLeak<T>
        extends WeakReference<Object> implements ResourceLeakTracker<T>, ResourceLeak {

private void clearRefQueue() {
    for (;;) {
        DefaultResourceLeak ref = (DefaultResourceLeak) refQueue.poll();
        if (ref == null) break;
        ref.dispose();   // 仅记录泄漏堆栈，不释放底层内存
    }
}

07总结归纳