Netty內(nèi)存池泄漏問題以解決方案

更新時間：2023年12月28日 09:36:28 作者：你的豆腐在這

這篇文章主要介紹了Netty內(nèi)存池泄漏問題以解決方案,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教

為了提升消息接收和發(fā)送性能，Netty針對ByteBuf的申請和釋放采用池化技術(shù)，通過PooledByteBufAllocator可以創(chuàng)建基于內(nèi)存池分配的ByteBuf對象，這樣就避免了每次消息讀寫都申請和釋放ByteBuf。由于ByteBuf涉及byte[]數(shù)組的創(chuàng)建和銷毀，對于性能要求苛刻的系統(tǒng)而言，重用ByteBuf帶來的性能收益是非?？捎^的。

內(nèi)存池是一把雙刃劍，如果使用不當，很容易帶來內(nèi)存泄漏和內(nèi)存非法引用等問題，另外，除了內(nèi)存池，Netty同時也支持非池化的ByteBuf，多種類型的ByteBuf功能存在一些差異，使用不當很容易帶來各種問題。

業(yè)務(wù)路由分發(fā)模塊使用Netty作為通信框架，負責協(xié)議消息的接入和路由轉(zhuǎn)發(fā)，在功能測試時沒有發(fā)現(xiàn)問題，轉(zhuǎn)性能測試之后，運行一段時間就發(fā)現(xiàn)內(nèi)存分配異常，服務(wù)端無法接收請求消息，系統(tǒng)吞吐量降為0。

1 路由轉(zhuǎn)發(fā)服務(wù)代碼

作為案例示例，對業(yè)務(wù)服務(wù)路由轉(zhuǎn)發(fā)代碼進行簡化，以方便分析：

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(()->{
            //解析請求消息，做路由轉(zhuǎn)發(fā)，代碼省略
            //轉(zhuǎn)發(fā)成功，返回響應(yīng)給客戶端
            ByteBuf respMsg = allocator.heapBuffer(body.length);
            respMsg.writeBytes(body);//作為示例，簡化處理，將請求返回
            ctx.writeAndFlush(respMsg);
        });
    }
}

進行一段時間的性能測試之后，日志中出現(xiàn)異常，進程內(nèi)存不斷飆升，懷疑存在內(nèi)存泄漏問題，如下圖所示。

2 響應(yīng)消息內(nèi)存釋放玄機

對業(yè)務(wù)ByteBuf申請相關(guān)代碼進行排查，發(fā)現(xiàn)響應(yīng)消息由業(yè)務(wù)線程創(chuàng)建，但是卻沒有主動釋放，因此懷疑是響應(yīng)消息沒有釋放導(dǎo)致的內(nèi)存泄漏。

因為響應(yīng)消息使用的是PooledHeapByteBuf，如果發(fā)生內(nèi)存泄漏，利用堆內(nèi)存監(jiān)控就可以找到泄漏點，通過Java VisualVM工具觀察堆內(nèi)存占用趨勢，并沒有發(fā)現(xiàn)堆內(nèi)存發(fā)生泄漏，如下圖所示。

對內(nèi)存做快照，查看在性能壓測過程中響應(yīng)消息PooledUnsafeHeapByteBuf的實例個數(shù)，如下圖所示，響應(yīng)消息對象個數(shù)和內(nèi)存占用都很少，排除內(nèi)存泄漏嫌疑。

業(yè)務(wù)從內(nèi)存池中申請了ByteBuf，但是卻沒有主動釋放它，最后也沒有發(fā)生內(nèi)存泄漏，這究竟是什么原因呢？

通過對Netty源碼的分析，我們破解了其中的玄機。

原來調(diào)用ctx.writeAndFlush(respMsg)方法時，當消息發(fā)送完成，Netty框架會主動幫助應(yīng)用釋放內(nèi)存，內(nèi)存的釋放分為如下兩種場景。

（1）如果是堆內(nèi)存（PooledHeapByteBuf），則將HeapByteBuffer轉(zhuǎn)換成DirectByteBuffer，并釋放PooledHeapByteBuf到內(nèi)存池，代碼如下（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源碼進行斷點調(diào)試，驗證上述分析。

斷點1：在響應(yīng)消息發(fā)送處設(shè)置斷點，獲取到的PooledUnsafeHeapByteBuf實例的ID為1506，如下圖所示。

斷點2：在HeapByteBuffer轉(zhuǎn)換成DirectByteBuffer處設(shè)置斷點，發(fā)現(xiàn)實例ID為1506的PooledUnsafeHeapByteBuf被釋放，如下圖所示。

斷點3：轉(zhuǎn)換之后待發(fā)送的響應(yīng)消息PooledUnsafeDirectByteBuf實例的ID為1527，如下圖所示。

斷點4：在響應(yīng)消息發(fā)送完成后，實例ID為1527的PooledUnsafeDirectByteBuf被釋放到內(nèi)存池中，如下圖所示。

（2）如果是DirectByteBuffer，則不需要轉(zhuǎn)換，在消息發(fā)送完成后，由ChannelOutboundBuffer的remove()負責釋放。

通過源碼解讀、調(diào)試及堆內(nèi)存的監(jiān)控分析，可以確認不是響應(yīng)消息沒有主動釋放導(dǎo)致的內(nèi)存泄漏，需要Dump內(nèi)存做進一步定位。

3 采集堆內(nèi)存快照分析

執(zhí)行jmap命令，Dump應(yīng)用內(nèi)存堆棧，如圖8所示。

通過MemoryAnalyzer工具對內(nèi)存堆棧進行分析，尋找內(nèi)存泄漏點，如圖9所示。

從下圖可以看出，內(nèi)存泄漏點是Netty內(nèi)存池對象PoolChunk，由于請求和響應(yīng)消息內(nèi)存分配都來自PoolChunk，暫時還不確認是請求還是響應(yīng)消息導(dǎo)致的問題。

進一步對代碼進行分析，發(fā)現(xiàn)響應(yīng)消息使用的是堆內(nèi)存HeapByteBuffer，請求消息使用的是DirectByteBuffer，由于Dump出來的是堆內(nèi)存，如果是堆內(nèi)存泄漏，Dump出來的內(nèi)存文件應(yīng)該包含大量的PooledHeapByteBuf，實際上并沒有，因此可以確認系統(tǒng)發(fā)生了堆外內(nèi)存泄漏，即請求消息沒有被釋放或者沒有被及時釋放導(dǎo)致的內(nèi)存泄漏。

對請求消息的內(nèi)存分配進行分析，發(fā)現(xiàn)在NioByteUnsafe的read方法中申請了內(nèi)存，代碼如下（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; 
//代碼省略

繼續(xù)對allocate方法進行分析，發(fā)現(xiàn)調(diào)用的是DefaultMaxMessagesRecvByteBuf- Allocator$MaxMessageHandle的allocate方法，代碼如下（DefaultMaxMessagesRecvByteBuf- Allocator）：

 public ByteBuf allocate(ByteBufAllocator alloc) { 
     return alloc.ioBuffer(guess()); 
 }

alloc.ioBuffer方法最終會調(diào)用PooledByteBufAllocator的newDirectBuffer方法創(chuàng)建PooledDirectByteBuf對象。

請求ByteBuf的創(chuàng)建分析完，繼續(xù)分析它的釋放操作，由于業(yè)務(wù)的RouterServerHandler繼承自ChannelInboundHandlerAdapter，它的channelRead(ChannelHandlerContext ctx, Object msg)方法執(zhí)行完成，ChannelHandler的執(zhí)行就結(jié)束了，代碼示例如下：

@Override 
public void channelRead(ChannelHandlerContext ctx, Object msg) { 
     ByteBuf reqMsg = (ByteBuf)msg; 
     byte [] body = new byte[reqMsg.readableBytes()]; 
     executorService.execute(()-> { 
         //解析請求消息，做路由轉(zhuǎn)發(fā)，代碼省略 
         //轉(zhuǎn)發(fā)成功，返回響應(yīng)給客戶端 
        ByteBuf respMsg = allocator.heapBuffer(body.length); 
        respMsg.writeBytes(body);//作為示例，簡化處理，將請求返回 
        ctx.writeAndFlush(respMsg); 
     }); 
}

通過代碼分析發(fā)現(xiàn)，請求ByteBuf被Netty框架申請后竟然沒有被釋放，為了驗證分析，在業(yè)務(wù)代碼中調(diào)用ReferenceCountUtil的release方法進行內(nèi)存釋放操作，代碼修改如下：

@Override 
public void channelRead(ChannelHandlerContext ctx, Object msg) { 
     ByteBuf reqMsg = (ByteBuf)msg;byte [] body = new byte[reqMsg.readableBytes()]; 
     ReferenceCountUtil.release(reqMsg); 
//后續(xù)代碼省略

修改之后繼續(xù)進行壓測，發(fā)現(xiàn)系統(tǒng)運行平穩(wěn)，沒有發(fā)生OOM異常。對內(nèi)存活動對象進行排序，沒有再發(fā)現(xiàn)大量的PoolChunk對象，內(nèi)存泄漏問題解決，問題修復(fù)之后的內(nèi)存快照如下圖所示。

4 ByteBuf申請和釋放的理解誤區(qū)

有一種說法認為Netty框架分配的ByteBuf框架會自動釋放，業(yè)務(wù)不需要釋放；業(yè)務(wù)創(chuàng)建的ByteBuf則需要自己釋放，Netty框架不會釋放。

通過前面的案例分析和驗證，我們可以看出這個觀點是錯誤的。為了在實際項目中更好地管理ByteBuf，下面我們分4種場景進行說明。

1.基于內(nèi)存池的請求ByteBuf

這類ByteBuf主要包括PooledDirectByteBuf和PooledHeapByteBuf，它由Netty的NioEventLoop線程在處理Channel的讀操作時分配，需要在業(yè)務(wù)ChannelInboundHandler處理完請求消息之后釋放（通常在解碼之后），它的釋放有兩種策略。

策略1 業(yè)務(wù)ChannelInboundHandler繼承自SimpleChannelInboundHandler，實現(xiàn)它的抽象方法channelRead0(ChannelHandlerContext ctx, I msg)，ByteBuf的釋放業(yè)務(wù)不用關(guān)心，由SimpleChannelInboundHandler負責釋放，相關(guān)代碼如下（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); 
        } 
    } 
}

如果當前業(yè)務(wù)ChannelInboundHandler需要執(zhí)行，則調(diào)用channelRead0之后執(zhí)行ReferenceCountUtil.release(msg)釋放當前請求消息。如果沒有匹配上需要繼續(xù)執(zhí)行后續(xù)的ChannelInboundHandler，則不釋放當前請求消息，調(diào)用ctx.fireChannelRead(msg)驅(qū)動ChannelPipeline繼續(xù)執(zhí)行。

對案例中的問題代碼進行修改，繼承自SimpleChannelInboundHandler，即便業(yè)務(wù)不釋放請求的ByteBuf對象，依然不會發(fā)生內(nèi)存泄漏，修改之后的代碼如下（RouterServerHandlerV2）：

public class RouterServerHandlerV2 extends SimpleChannelInboundHandler <ByteBuf> { 
 
@Override 
public void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) { 
    byte [] body = new byte[msg.readableBytes()]; 
    executorService.execute(()-> { 
        //解析請求消息，做路由轉(zhuǎn)發(fā)，代碼省略
        //轉(zhuǎn)發(fā)成功，返回響應(yīng)給客戶端 
        ByteBuf respMsg = allocator.heapBuffer(body.length); 
        respMsg.writeBytes(body);//作為示例，簡化處理，將請求返回 
        ctx.writeAndFlush(respMsg); 
    }); 
}

對修改之后的代碼做性能測試，發(fā)現(xiàn)內(nèi)存占用平穩(wěn)，無內(nèi)存泄漏問題，驗證了之前的分析結(jié)論。

策略2 在業(yè)務(wù)ChannelInboundHandler中調(diào)用ctx.fireChannelRead(msg)方法，讓請求消息繼續(xù)向后執(zhí)行，直到調(diào)用DefaultChannelPipeline的內(nèi)部類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.基于非內(nèi)存池的請求ByteBuf

如果業(yè)務(wù)使用非內(nèi)存池模式覆蓋Netty默認的內(nèi)存池模式創(chuàng)建請求ByteBuf，例如通過如下代碼修改內(nèi)存申請策略為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()); 
    } 
}

也需要按照內(nèi)存池的方式釋放內(nèi)存。

3.基于內(nèi)存池的響應(yīng)ByteBuf

根據(jù)之前的分析，只要調(diào)用了writeAndFlush或者flush方法，在消息發(fā)送完成后都會由Netty框架進行內(nèi)存釋放，業(yè)務(wù)不需要主動釋放內(nèi)存。

4.基于非內(nèi)存池的響應(yīng)ByteBuf

無論是基于內(nèi)存池還是非內(nèi)存池分配的ByteBuf，如果是堆內(nèi)存，則將堆內(nèi)存轉(zhuǎn)換成堆外內(nèi)存，然后釋放HeapByteBuffer，待消息發(fā)送完成，再釋放轉(zhuǎn)換后的DirectByteBuf；如果是DirectByteBuffer，則不需要轉(zhuǎn)換，待消息發(fā)送完成之后釋放。因此對于需要發(fā)送的響應(yīng)ByteBuf，由業(yè)務(wù)創(chuàng)建，但是不需要由業(yè)務(wù)來釋放