Netty的引用计数对象

作者: 大风过岗 | 来源:发表于2020-01-11 18:35 被阅读0次

引用计数对象

从Netty4开始，对象的生命周期由它们的引用计数负责管理，这样，一旦它们不被使用的时候，Netty就可以把他们放入对象池中。
垃圾回收和引用队列无法提供高效的实时的不可达保证，然而，引用计数却可以通过牺牲些许便利性，做到这一点。

ByteBuf就是其中最显著的一种数据类型，它利用引用计数实现了高性能的内存分配和内存释放。本节将解释一下使用ByteBuf时，引用计数的内部机制。

引用计数的基本概念

一个新的引用计数对象的初始引用数是 1 ：

ByteBuf buf = ctx.alloc().directBuffer();
assert buf.refCnt() == 1;

当你释放引用计数对象时，它的引用计数为减一。如果引用计数为0，该引用计数对象就会被释放或者把它放回原来的对象池中。

assert buf.refCnt() == 1;
// release() returns true only if the reference count becomes 0.
boolean destroyed = buf.release();
assert destroyed;
assert buf.refCnt() == 0;

悬挂引用

试图访问一个引用计数为0的对象将会触发一个IllegalReferenceCountException异常：

assert buf.refCnt() == 0;
try {
  buf.writeLong(0xdeadbeef);
  throw new Error("should not reach here");
} catch (IllegalReferenceCountExeception e) {
  // Expected
}

增加引用计数

只要某个引用计数对象还没有被销毁，就可以通过调用retain()方法使它的引用计数增加。

ByteBuf buf = ctx.alloc().directBuffer();
assert buf.refCnt() == 1;

buf.retain();
assert buf.refCnt() == 2;

boolean destroyed = buf.release();
assert !destroyed;
assert buf.refCnt() == 1;

谁负责销毁？

通常的经验法则是：谁最后访问引用计数对象，谁负责销毁。更特殊的是：

如果发送方把一个引用计数对象传递给另一个接收方，那么发送方通常不需要进行销毁操作，而是把销毁的工作交给接收方来做。

如果一个组件负责处理一个引用计数对象，并且确定该引用计数对象不再会被其他组件访问，那么，该组件应负责销毁它。

下面有一个简单的例子：

public ByteBuf a(ByteBuf input) {
    input.writeByte(42);
    return input;
}

public ByteBuf b(ByteBuf input) {
    try {
        output = input.alloc().directBuffer(input.readableBytes() + 1);
        output.writeBytes(input);
        output.writeByte(42);
        return output;
    } finally {
        input.release();
    }
}

public void c(ByteBuf input) {
    System.out.println(input);
    input.release();
}

public void main() {
    ...
    ByteBuf buf = ...;
    // This will print buf to System.out and destroy it.
    c(b(a(buf)));
    assert buf.refCnt() == 0;
}

Action Who should release? Who released?

main() creates buf buf→main()
main() calls a() with buf buf→a()
a() returns buf merely. buf→main()
main() calls b() with buf buf→b()
b() returns the copy of buf buf→b(), copy→main() b() releases buf
main() calls c() with copy copy→c()
c() swallows copy copy→c() c() releases copy

源生buffer

ByteBuf.duplicate(), ByteBuf.slice() 以及 ByteBuf.order(ByteOrder) 这三个方法都能创建一个衍生buffer（衍生buffer共享父级buffer的内存空间）。衍生buffer共享父buffer的引用计数，他们没有自己的引用计数。

ByteBuf parent = ctx.alloc().directBuffer();
ByteBuf derived = parent.duplicate();

// Creating a derived buffer does not increase the reference count.
assert parent.refCnt() == 1;
assert derived.refCnt() == 1;

相反，ByteBuf.copy() 和 ByteBuf.readBytes(int)则不是衍生buffer。这些方法的buffer都是有自己的内存空间，因此需要单独进行释放。

注意：
父级buffer和它的衍生buffer共享相同的引用计数，并且当创建一个衍生buffer的时候，引用计数值并不增加。因此，如果你准备把一个衍生buffer传递给其他组件时，你不得不先调用下retain()方法。

ByteBuf parent = ctx.alloc().directBuffer(512);
parent.writeBytes(...);

try {
    while (parent.isReadable(16)) {
        ByteBuf derived = parent.readSlice(16);
        derived.retain();
        process(derived);
    }
} finally {
    parent.release();
}
...

public void process(ByteBuf buf) {
    ...
    buf.release();
}

ByteBufHolder 接口

有时，ByteBuf会包含在一个buffer holder里，例如： DatagramPacket, HttpContent, and WebSocketframe。这些类型扩展了一个相同的ByteHolder接口。

像衍生buffer一样，buffer holder和它所包含的buffer共享相同的引用计数。

ChannelHandler中的引用计数

入站消息

当evetLoop把数据读进ByteBuf时，会触发一个相应的channelRead()方法。此时，由相应管道中的ChannelHandler负责释放此buffer。因此，对接收到的数据进行处理的handler，应该在它的channelRead()方法中调动release()方法释放相应数据。

public void channelRead(ChannelHandlerContext ctx, Object msg) {
    ByteBuf buf = (ByteBuf) msg;
    try {
        ...
    } finally {
        buf.release();
    }
}

正如在上面讲述的“谁负责销毁”，如果你的handler需要把一个buffer传递到另一个handler中的话，此时，你不需要释放它：

public void channelRead(ChannelHandlerContext ctx, Object msg) {
    ByteBuf buf = (ByteBuf) msg;
    ...
    ctx.fireChannelRead(buf);
}

注意： ByteBuf并不是Netty中唯一的引用计数类型，如果你处理的是由解码器产生的消息时，很有可能该消息也是引用计数对象：

// Assuming your handler is placed next to `HttpRequestDecoder`
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    if (msg instanceof HttpRequest) {
        HttpRequest req = (HttpRequest) msg;
        ...
    }
    if (msg instanceof HttpContent) {
        HttpContent content = (HttpContent) msg;
        try {
            ...
        } finally {
            content.release();
        }
    }
}

如果你心存怀疑的话，你可以很轻易地使用 ReferenceCountUtil.release()释放此消息：

public void channelRead(ChannelHandlerContext ctx, Object msg) {
    try {
        ...
    } finally {
        ReferenceCountUtil.release(msg);
    }
}

或者，你可以选择继承SimpleChannelHandler，它会为你接收到所有消息调用ReferenceCountUtil.release(msg)。

出站消息

和入站消息不同的是，出站消息是由你的应用创建的，所以，Netty在把他们写出去之后，会负责释放这些消息。然而，那些拦截你写请求的handler要确保释放所有的中间对象(e.g: 解码器)


// Simple-pass through
public void write(ChannelHandlerContext ctx, Object message, ChannelPromise promise) {
    System.err.println("Writing: " + message);
    ctx.write(message, promise);
}

// Transformation
public void write(ChannelHandlerContext ctx, Object message, ChannelPromise promise) {
    if (message instanceof HttpContent) {
        // Transform HttpContent to ByteBuf.
        HttpContent content = (HttpContent) message;
        try {
            ByteBuf transformed = ctx.alloc().buffer();
            ....
            ctx.write(transformed, promise);
        } finally {
            content.release();
        }
    } else {
        // Pass non-HttpContent through.
        ctx.write(message, promise);
    }
}

缓冲区泄漏的解决

引用计数的缺点是，它很容易泄漏引用对象。因为JVM并不认识Netty实现的引用计数对象，在它们变得不可达时，JVM会自动释放它们，即使它们的引用计数不是0。一旦对象被垃圾回收之后，就无法再令他们复活，因此，也就无法被放入对象池中。故而会造成内存泄漏。

不幸的是，尽管寻找内存泄漏很困难，Netty默认会抽样1%的缓冲区分配，从而检查他们是否存在内存泄漏。当发生泄漏时，你会得到如下日志信息：

LEAK: ByteBuf.release() was not called before it's garbage-collected. Enable advanced leak reporting to find out where the leak occurred. To enable advanced leak reporting, specify the JVM option '-Dio.netty.leakDetectionLevel=advanced' or call ResourceLeakDetector.setLevel()

按照上面的提示修改你的JVM配置，重新运行你的应用，之后，你会看到泄漏缓存区发生的位置。下面的输出就展示了我们的单元测试(XmlFrameDecoderTest.testDecodeWithXml())中的内存泄漏:

Running io.netty.handler.codec.xml.XmlFrameDecoderTest
15:03:36.886 [main] ERROR io.netty.util.ResourceLeakDetector - LEAK: ByteBuf.release() was not called before it's garbage-collected.
Recent access records: 1
#1:
    io.netty.buffer.AdvancedLeakAwareByteBuf.toString(AdvancedLeakAwareByteBuf.java:697)
    io.netty.handler.codec.xml.XmlFrameDecoderTest.testDecodeWithXml(XmlFrameDecoderTest.java:157)
    io.netty.handler.codec.xml.XmlFrameDecoderTest.testDecodeWithTwoMessages(XmlFrameDecoderTest.java:133)
    ...

Created at:
    io.netty.buffer.UnpooledByteBufAllocator.newDirectBuffer(UnpooledByteBufAllocator.java:55)
    io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:155)
    io.netty.buffer.UnpooledUnsafeDirectByteBuf.copy(UnpooledUnsafeDirectByteBuf.java:465)
    io.netty.buffer.WrappedByteBuf.copy(WrappedByteBuf.java:697)
    io.netty.buffer.AdvancedLeakAwareByteBuf.copy(AdvancedLeakAwareByteBuf.java:656)
    io.netty.handler.codec.xml.XmlFrameDecoder.extractFrame(XmlFrameDecoder.java:198)
    io.netty.handler.codec.xml.XmlFrameDecoder.decode(XmlFrameDecoder.java:174)
    io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:227)
    io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:140)
    io.netty.channel.ChannelHandlerInvokerUtil.invokeChannelReadNow(ChannelHandlerInvokerUtil.java:74)
    io.netty.channel.embedded.EmbeddedEventLoop.invokeChannelRead(EmbeddedEventLoop.java:142)
    io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:317)
    io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:846)
    io.netty.channel.embedded.EmbeddedChannel.writeInbound(EmbeddedChannel.java:176)
    io.netty.handler.codec.xml.XmlFrameDecoderTest.testDecodeWithXml(XmlFrameDecoderTest.java:147)
    io.netty.handler.codec.xml.XmlFrameDecoderTest.testDecodeWithTwoMessages(XmlFrameDecoderTest.java:133)
    ...

如果你使用的是Netty 5或更高的版本，会提示额外的信息帮助你定位到哪个handler最后一次处理了内存泄漏。
下面的案例，展示了内存泄漏是由名字叫EchoServerHandler#0的handler处理的然后被垃圾回收，这即意味着：很有可能是 EchoServerHandler#0 忘记了释放内存。

12:05:24.374 [nioEventLoop-1-1] ERROR io.netty.util.ResourceLeakDetector - LEAK: ByteBuf.release() was not called before it's garbage-collected.
Recent access records: 2
#2:
    Hint: 'EchoServerHandler#0' will handle the message from this point.
    io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:329)
    io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:846)
    io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:133)
    io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:485)
    io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:452)
    io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:346)
    io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:794)
    java.lang.Thread.run(Thread.java:744)
#1:
    io.netty.buffer.AdvancedLeakAwareByteBuf.writeBytes(AdvancedLeakAwareByteBuf.java:589)
    io.netty.channel.socket.nio.NioSocketChannel.doReadBytes(NioSocketChannel.java:208)
    io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:125)
    io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:485)
    io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:452)
    io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:346)
    io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:794)
    java.lang.Thread.run(Thread.java:744)
Created at:
    io.netty.buffer.UnpooledByteBufAllocator.newDirectBuffer(UnpooledByteBufAllocator.java:55)
    io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:155)
    io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:146)
    io.netty.buffer.AbstractByteBufAllocator.ioBuffer(AbstractByteBufAllocator.java:107)
    io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:123)
    io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:485)
    io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:452)
    io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:346)
    io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:794)
    java.lang.Thread.run(Thread.java:744)

泄漏检测的级别

当前有4个泄漏检测的级别：

DISABLED - 完全禁用内存泄漏检测，不推荐。
SIMPLE - 抽样1%的buffer，并诊断释放有内存泄漏. 默认级别.
ADVANCED - 抽样1%的buffer，诊断出那些地方访问了这些内存泄漏.
PARANOID -和ADVANCED相同，不同的是，它是针对的每个单一的buffer. 自动测试时，这样很有用。当构建输出中包含‘LEAK’字样时，你的构建将会失败。

你也可以通过JVM 配置来指定内存泄漏级别：

java -Dio.netty.leakDetection.level=advanced ...

注意：This property used to be called io.netty.leakDetectionLevel.

避免内存泄漏的最佳实践

运行你的单元测试，并开启PARANOID内存泄漏级别的检测。
在将应用程序以简单的方式扩展到整个集群之前，请在相当长的一段时间内对应用程序进行检测，以确定是否存在泄漏。
如果有内存泄漏，在ADVANCED级别进行金丝雀测试获取更多的提示信息。
别把一个有内存泄漏的应用部署到整个集群。

在单元测试中修复缓冲区泄漏

在单元测试中，很容易忘记对buffer或消息进行释放。这样会产生一个内存泄漏的警告，但是这并不意味着你的应用就一定存在内存泄漏。
另外，除了在try-finally块中进行所有的buffer外，还可以使用ReferenceCountUtil.releaseLater()来办到这一点。

import static io.netty.util.ReferenceCountUtil.*;

@Test
public void testSomething() throws Exception {
    // ReferenceCountUtil.releaseLater() will keep the reference of buf,
    // and then release it when the test thread is terminated.
    ByteBuf buf = releaseLater(Unpooled.directBuffer(512));
    ...
}