Netty内存池泄漏问题以解决方案
为了提升消息接收和发送性能，Netty针对ByteBuf的申请和释放采用池化技术，通过PooledByteBufAllocator可以创建基于内存池分配的ByteBuf对象，这样就避免了每次消息读写都申请和释放ByteBuf。由于ByteBuf涉及byte[]数组的创建和销毁，对于性能要求苛刻的系统而言，重用ByteBuf带来的性能收益是非常可观的。
内存池是一把双刃剑，如果使用不当，很容易带来内存泄漏和内存非法引用等问题，另外，除了内存池，Netty同时也支持非池化的ByteBuf，多种类型的ByteBuf功能存在一些差异，使用不当很容易带来各种问题。
业务路由分发模块使用Netty作为通信框架，负责协议消息的接入和路由转发，在功能测试时没有发现问题，转性能测试之后，运行一段时间就发现内存分配异常，服务端无法接收请求消息，系统吞吐量降为0。
1 路由转发服务代码
作为案例示例，对业务服务路由转发代码进行简化，以方便分析：
public class RouterServerHandler extends ChannelInboundHandlerAdapter {
static ExecutorService executorService = Executors.newSingleThreadExecutor();
PooledByteBufAllocator allocator = new PooledByteBufAllocator(false);
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf reqMsg = (ByteBuf)msg;
byte [] body = new byte[reqMsg.readableBytes()];
executorService.execute(()->{
//解析请求消息，做路由转发，代码省略
//转发成功，返回响应给客户端
ByteBuf respMsg = allocator.heapBuffer(body.length);
respMsg.writeBytes(body);//作为示例，简化处理，将请求返回
ctx.writeAndFlush(respMsg);
});
}
}进行一段时间的性能测试之后，日志中出现异常，进程内存不断飙升，怀疑存在内存泄漏问题，如下图所示。
2 响应消息内存释放玄机
对业务ByteBuf申请相关代码进行排查，发现响应消息由业务线程创建，但是却没有主动释放，因此怀疑是响应消息没有释放导致的内存泄漏。
因为响应消息使用的是PooledHeapByteBuf，如果发生内存泄漏，利用堆内存监控就可以找到泄漏点，通过Java VisualVM工具观察堆内存占用趋势，并没有发现堆内存发生泄漏，如下图所示。
对内存做快照，查看在性能压测过程中响应消息PooledUnsafeHeapByteBuf的实例个数，如下图所示，响应消息对象个数和内存占用都很少，排除内存泄漏嫌疑。
业务从内存池中申请了ByteBuf，但是却没有主动释放它，最后也没有发生内存泄漏，这究竟是什么原因呢？
通过对Netty源码的分析，我们破解了其中的玄机。
原来调用ctx.writeAndFlush(respMsg)方法时，当消息发送完成，Netty框架会主动帮助应用释放内存，内存的释放分为如下两种场景。
（1）如果是堆内存（PooledHeapByteBuf），则将HeapByteBuffer转换成DirectByteBuffer，并释放PooledHeapByteBuf到内存池，代码如下（AbstractNioChannel类）：
protected final ByteBuf newDirectBuffer(ByteBuf buf) {
final int readableBytes = buf.readableBytes();
if (readableBytes == 0) {
ReferenceCountUtil.safeRelease(buf);
return Unpooled.EMPTY_BUFFER;
}
final ByteBufAllocator alloc = alloc();
if (alloc.isDirectBufferPooled()) {
ByteBuf directBuf = alloc.directBuffer(readableBytes);
directBuf.writeBytes(buf, buf.readerIndex(), readableBytes);
ReferenceCountUtil.safeRelease(buf);
return directBuf;
}
} 如果消息完整地被写到SocketChannel中，则释放DirectByteBuffer，代码如下（ChannelOutboundBuffer）：
public boolean remove() {
Entry e = flushedEntry;
if (e == null) {
clearNioBuffers();
return false;
}
Object msg = e.msg;
ChannelPromise promise = e.promise;
int size = e.pendingSize;
removeEntry(e);
if (!e.cancelled) {
ReferenceCountUtil.safeRelease(msg);
safeSuccess(promise);
decrementPendingOutboundBytes(size, false, true);
}
} 对Netty源码进行断点调试，验证上述分析。
断点1：在响应消息发送处设置断点，获取到的PooledUnsafeHeapByteBuf实例的ID为1506，如下图所示。
断点2：在HeapByteBuffer转换成DirectByteBuffer处设置断点，发现实例ID为1506的PooledUnsafeHeapByteBuf被释放，如下图所示。
断点3：转换之后待发送的响应消息PooledUnsafeDirectByteBuf实例的ID为1527，如下图所示。
断点4：在响应消息发送完成后，实例ID为1527的PooledUnsafeDirectByteBuf被释放到内存池中，如下图所示。
（2）如果是DirectByteBuffer，则不需要转换，在消息发送完成后，由ChannelOutboundBuffer的remove()负责释放。
通过源码解读、调试及堆内存的监控分析，可以确认不是响应消息没有主动释放导致的内存泄漏，需要Dump内存做进一步定位。
3 采集堆内存快照分析
执行jmap命令，Dump应用内存堆栈，如图8所示。
通过MemoryAnalyzer工具对内存堆栈进行分析，寻找内存泄漏点，如图9所示。
从下图可以看出，内存泄漏点是Netty内存池对象PoolChunk，由于请求和响应消息内存分配都来自PoolChunk，暂时还不确认是请求还是响应消息导致的问题。
进一步对代码进行分析，发现响应消息使用的是堆内存HeapByteBuffer，请求消息使用的是DirectByteBuffer，由于Dump出来的是堆内存，如果是堆内存泄漏，Dump出来的内存文件应该包含大量的PooledHeapByteBuf，实际上并没有，因此可以确认系统发生了堆外内存泄漏，即请求消息没有被释放或者没有被及时释放导致的内存泄漏。
对请求消息的内存分配进行分析，发现在NioByteUnsafe的read方法中申请了内存，代码如下（NioByteUnsafe）：
public final void read() {
final ChannelConfig config = config();
if (shouldBreakReadReady(config)) {
clearReadPending();
return;
}
final ChannelPipeline pipeline = pipeline();
final ByteBufAllocator allocator = config.getAllocator();
final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
allocHandle.reset(config);
ByteBuf byteBuf = null;
boolean close = false;
//代码省略 继续对allocate方法进行分析，发现调用的是DefaultMaxMessagesRecvByteBuf- Allocator$MaxMessageHandle的allocate方法，代码如下（DefaultMaxMessagesRecvByteBuf- Allocator）：
public ByteBuf allocate(ByteBufAllocator alloc) {
return alloc.ioBuffer(guess());
} alloc.ioBuffer方法最终会调用PooledByteBufAllocator的newDirectBuffer方法创建PooledDirectByteBuf对象。
请求ByteBuf的创建分析完，继续分析它的释放操作，由于业务的RouterServerHandler继承自ChannelInboundHandlerAdapter，它的channelRead(ChannelHandlerContext ctx, Object msg)方法执行完成，ChannelHandler的执行就结束了，代码示例如下：
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf reqMsg = (ByteBuf)msg;
byte [] body = new byte[reqMsg.readableBytes()];
executorService.execute(()-> {
//解析请求消息，做路由转发，代码省略
//转发成功，返回响应给客户端
ByteBuf respMsg = allocator.heapBuffer(body.length);
respMsg.writeBytes(body);//作为示例，简化处理，将请求返回
ctx.writeAndFlush(respMsg);
});
} 通过代码分析发现，请求ByteBuf被Netty框架申请后竟然没有被释放，为了验证分析，在业务代码中调用ReferenceCountUtil的release方法进行内存释放操作，代码修改如下：
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf reqMsg = (ByteBuf)msg;byte [] body = new byte[reqMsg.readableBytes()];
ReferenceCountUtil.release(reqMsg);
//后续代码省略 修改之后继续进行压测，发现系统运行平稳，没有发生OOM异常。对内存活动对象进行排序，没有再发现大量的PoolChunk对象，内存泄漏问题解决，问题修复之后的内存快照如下图所示。
4 ByteBuf申请和释放的理解误区
有一种说法认为Netty框架分配的ByteBuf框架会自动释放，业务不需要释放；业务创建的ByteBuf则需要自己释放，Netty框架不会释放。
通过前面的案例分析和验证，我们可以看出这个观点是错误的。为了在实际项目中更好地管理ByteBuf，下面我们分4种场景进行说明。
1.基于内存池的请求ByteBuf
这类ByteBuf主要包括PooledDirectByteBuf和PooledHeapByteBuf，它由Netty的NioEventLoop线程在处理Channel的读操作时分配，需要在业务ChannelInboundHandler处理完请求消息之后释放（通常在解码之后），它的释放有两种策略。
策略1 业务ChannelInboundHandler继承自SimpleChannelInboundHandler，实现它的抽象方法channelRead0(ChannelHandlerContext ctx, I msg)，ByteBuf的释放业务不用关心，由SimpleChannelInboundHandler负责释放，相关代码如下（SimpleChannelInboundHandler）：
@Override
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 && re lease) {
ReferenceCountUtil.release(msg);
}
}
} 如果当前业务ChannelInboundHandler需要执行，则调用channelRead0之后执行ReferenceCountUtil.release(msg)释放当前请求消息。如果没有匹配上需要继续执行后续的ChannelInboundHandler，则不释放当前请求消息，调用ctx.fireChannelRead(msg)驱动ChannelPipeline继续执行。
对案例中的问题代码进行修改，继承自SimpleChannelInboundHandler，即便业务不释放请求的ByteBuf对象，依然不会发生内存泄漏，修改之后的代码如下（RouterServerHandlerV2）：
public class RouterServerHandlerV2 extends SimpleChannelInboundHandler <ByteBuf> {
@Override
public void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
byte [] body = new byte[msg.readableBytes()];
executorService.execute(()-> {
//解析请求消息，做路由转发，代码省略
//转发成功，返回响应给客户端
ByteBuf respMsg = allocator.heapBuffer(body.length);
respMsg.writeBytes(body);//作为示例，简化处理，将请求返回
ctx.writeAndFlush(respMsg);
});
} 对修改之后的代码做性能测试，发现内存占用平稳，无内存泄漏问题，验证了之前的分析结论。
策略2 在业务ChannelInboundHandler中调用ctx.fireChannelRead(msg)方法，让请求消息继续向后执行，直到调用DefaultChannelPipeline的内部类TailContext，由它来负责释放请求消息，代码如下（TailContext）：
protected void onUnhandledInboundMessage(Object msg) {
try {
logger.debug( "Discarded inbound message {} that reached at the tail of thpipeline." +
"Please check your pipeline configuration.", msg);
} finally {
ReferenceCountUtil.release(msg);
}
} 2.基于非内存池的请求ByteBuf
如果业务使用非内存池模式覆盖Netty默认的内存池模式创建请求ByteBuf，例如通过如下代码修改内存申请策略为Unpooled：
//代码省略
childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
ch.config().setAllocator(UnpooledByteBufAllocator.DEFAULT);
p.addLast(new RouterServerHandler());
}
} 也需要按照内存池的方式释放内存。
3.基于内存池的响应ByteBuf
根据之前的分析，只要调用了writeAndFlush或者flush方法，在消息发送完成后都会由Netty框架进行内存释放，业务不需要主动释放内存。
4.基于非内存池的响应ByteBuf
无论是基于内存池还是非内存池分配的ByteBuf，如果是堆内存，则将堆内存转换成堆外内存，然后释放HeapByteBuffer，待消息发送完成，再释放转换后的DirectByteBuf；如果是DirectByteBuffer，则不需要转换，待消息发送完成之后释放。因此对于需要发送的响应ByteBuf，由业务创建，但是不需要由业务来释放
总结
以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。
相关文章
这篇文章主要介绍了Easycode自动化springboot的curd，围绕主题的相关资料展开详细内容，具有一定的参考价值，需要的小伙伴可以参考一下，希望给对你有所帮助2022-01-01
这篇文章主要介绍了java利用Socket实现聊天室功能实例，具有一定的参考价值，感兴趣的小伙伴们可以参考一下。2017-02-02
详解spring cloud整合Swagger2构建RESTful服务的APIs
这篇文章主要介绍了详解spring cloud整合Swagger2构建RESTful服务的APIs，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧2018-01-01
Fastjson是一个Java语言编写的高性能功能完善的JSON库。接下来通过本文给大家分享Fastjson 常用API介绍及下载地址，感兴趣的朋友一起看看吧2017-11-11
这篇文章主要介绍了IDEA检查项目的jdk版本需要看的地方,文中通过图文介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧2024-06-06
这篇文章主要介绍了Java使用Tessdata做OCR图片文字识别的详细思路,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下2021-07-07
这篇文章主要介绍了Java排序算法之选择排序,文中有非常详细的代码示例,对正在学习java的小伙伴们有非常好的帮助,需要的朋友可以参考下2021-05-05
Java自定义过滤器和拦截器实现ThreadLocal线程封闭
本文主要介绍了Java自定义过滤器和拦截器实现ThreadLocal线程封闭，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2022-08-08
这篇文章主要为大家介绍了elasticsearch之metric聚合实现示例详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪2023-01-01
MyBatis是一款优秀的数据库持久层框架,相比Hibernate我更喜欢使用MyBatis,看的到SQL还是让人更安心点,这篇文章主要给大家介绍了关于基于springboot创建mybatis的完整步骤,需要的朋友可以参考下2024-03-03
最新评论