在 netty 4.1 中正确释放引用计数的 ByteBuf 对象

Question

因此，我们目前正在基于 MQTT 的消息传递后端将 netty 3.x 升级到 netty 4.1。在我们的应用程序中，我们使用自定义 MQTT 消息解码器和编码器。

对于我们的解码器，我目前使用的是ByteToMessageDecoder，如下：

public class MqttMessageDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        if (in.readableBytes() < 2) {
            return;
        }

        .....
        .....
        .....

        byte[] data = new byte[msglength];
        in.resetReaderIndex();
        in.readBytes(data);
        MessageInputStream mis = new MessageInputStream(
                new ByteArrayInputStream(data));
        Message msg = mis.readMessage();
        out.add(msg);
        ReferenceCountUtil.release(in);
    }
}

其中Message 是我们的自定义对象，它被传递给下一个ChannelHandler 的channelRead()。如您所见，一旦我从中创建了Message 对象，我就完成了传入的ByteBuf 对象in。那么，既然ByteBuf在netty中是引用计数的，那么我需要在这里调用ReferenceCountUtil.release(in)释放in对象是否正确？理想情况下，根据doc，这似乎是正确的。但是，当我这样做时，我似乎面临着异常：

Wed May 24 io.netty.channel.DefaultChannelPipeline:? WARN netty-workers-7 An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception.
io.netty.channel.ChannelPipelineException: com.bsb.hike.mqtt.MqttMessageDecoder.handlerRemoved() has thrown an exception.
    at io.netty.channel.DefaultChannelPipeline.callHandlerRemoved0(DefaultChannelPipeline.java:631) [netty-all-4.1.0.Final.jar:4.1.0.Final]
    at io.netty.channel.DefaultChannelPipeline.destroyDown(DefaultChannelPipeline.java:867) [netty-all-4.1.0.Final.jar:4.1.0.Final]
    at io.netty.channel.DefaultChannelPipeline.access$300(DefaultChannelPipeline.java:45) [netty-all-4.1.0.Final.jar:4.1.0.Final]
    at io.netty.channel.DefaultChannelPipeline$9.run(DefaultChannelPipeline.java:874) [netty-all-4.1.0.Final.jar:4.1.0.Final]
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:339) [netty-all-4.1.0.Final.jar:4.1.0.Final]
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:374) [netty-all-4.1.0.Final.jar:4.1.0.Final]
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:742) [netty-all-4.1.0.Final.jar:4.1.0.Final]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_72-internal]
Caused by: io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1
    at io.netty.buffer.AbstractReferenceCountedByteBuf.release(AbstractReferenceCountedByteBuf.java:111) ~[netty-all-4.1.0.Final.jar:4.1.0.Final]
    at io.netty.handler.codec.ByteToMessageDecoder.handlerRemoved(ByteToMessageDecoder.java:217) ~[netty-all-4.1.0.Final.jar:4.1.0.Final]
    at io.netty.channel.DefaultChannelPipeline.callHandlerRemoved0(DefaultChannelPipeline.java:626) [netty-all-4.1.0.Final.jar:4.1.0.Final]
    ... 7 common frames omitted

这告诉我，当子通道关闭时，管道中的所有处理程序都被一个接一个地删除。当此解码器处理程序关闭时，我们会显式释放附加到此解码器的ByteBuf，这会在调用以下方法时导致IllegalReferenceCountException 异常。

这是AbstractReferenceCountedByteBuf#release：

@Override
    public boolean release() {
        for (;;) {
            int refCnt = this.refCnt;
            if (refCnt == 0) {
                throw new IllegalReferenceCountException(0, -1);
            }

            if (refCntUpdater.compareAndSet(this, refCnt, refCnt - 1)) {
                if (refCnt == 1) {
                    deallocate();
                    return true;
                }
                return false;
            }
        }
    }

那么释放ByteBuf 对象的正确方法是什么，以免遇到此问题？

我正在使用PooledByteBufAllocator -

new ServerBootstrap().childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)

如果您需要有关配置的更多信息，请告诉我。

---

编辑：

作为 Ferrybig 答案的附加组件，ByteToMessageDecoder#channelRead 自行处理传入的ByteBufs 的释放。请参阅finally 块 -

@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof ByteBuf) {
            CodecOutputList out = CodecOutputList.newInstance();
            try {
                ByteBuf data = (ByteBuf) msg;
                first = cumulation == null;
                if (first) {
                    cumulation = data;
                } else {
                    cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
                }
                callDecode(ctx, cumulation, out);
            } catch (DecoderException e) {
                throw e;
            } catch (Throwable t) {
                throw new DecoderException(t);
            } finally {
                if (cumulation != null && !cumulation.isReadable()) {
                    numReads = 0;
                    cumulation.release();
                    cumulation = null;
                } else if (++ numReads >= discardAfterReads) {
                    // We did enough reads already try to discard some bytes so we not risk to see a OOME.
                    // See https://github.com/netty/netty/issues/4275
                    numReads = 0;
                    discardSomeReadBytes();
                }

                int size = out.size();
                decodeWasNull = !out.insertSinceRecycled();
                fireChannelRead(ctx, out, size);
                out.recycle();
            }
        } else {
            ctx.fireChannelRead(msg);
        }
    }

如果入站ByteBuf 正被传输到管道中的下一个通道处理程序，则此ByteBuf 的引用计数将通过ByteBuf#retain 增加，因此如果解码器之后的下一个处理程序是您的业务处理程序（通常是这种情况），您需要在那里释放 ByteBuf 对象以避免任何内存泄漏。这在docs这里也有提及。

Answer 1

并非所有处理程序都需要销毁传入的 bytebuf。 ByteToMessageDecoder 就是其中之一。

这样做的原因是这个处理程序收集了多个传入的字节缓冲区，并将它们作为 1 个连续的字节流公开给您的应用程序，以便于编码，并且不需要自己处理这些块

请记住，您仍然需要使用 readBytes 或 readSlice 手动释放您创建的任何字节缓冲区，如 javadoc 所述。

Answer 2

@Ferrybig 他的回答已经足够好了。

这里我想添加一些处理 ByteBuf 释放的简单约定。

当你有这样的 InboundHandler 时：

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = ByteBuf(msg)
...
// no fireChannelRead()
}

由于你在它之后拦截了 InboundHandler（如果有的话），并且没有人处理 ByteBuf 对象，你需要手动释放它。

您也可以调用 fireChannelRead()。 Netty 添加一个尾部处理程序，默认为您释放ByteBuf。所以调用fireChannelRead()后不需要释放它。

如果我的回答有问题，请告诉我。