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

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

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

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

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

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

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

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

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

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

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

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

通過(guò)對(duì)Netty源碼的分析，我們破解了其中的玄機(jī)。

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

（1）如果是堆內(nèi)存（PooledHeapByteBuf），則將HeapByteBuffer轉(zhuǎn)換成DirectByteBuffer，并釋放PooledHeapByteBuf到內(nèi)存池，代碼如下（AbstractNioChannel類(lèi)）：

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; 
    } 
}  

如果消息完整地被寫(xiě)到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); 
    } 
} 

對(duì)Netty源碼進(jìn)行斷點(diǎn)調(diào)試，驗(yàn)證上述分析。

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

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

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

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

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

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

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

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

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

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

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

對(duì)請(qǐng)求消息的內(nèi)存分配進(jìn)行分析，發(fā)現(xiàn)在NioByteUnsafe的read方法中申請(qǐng)了內(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ù)對(duì)allocate方法進(jìn)行分析，發(fā)現(xiàn)調(diào)用的是DefaultMaxMessagesRecvByteBuf- Allocator$MaxMessageHandle的allocate方法，代碼如下（DefaultMaxMessagesRecvByteBuf- Allocator）：

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

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

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

通過(guò)代碼分析發(fā)現(xiàn)，請(qǐng)求ByteBuf被Netty框架申請(qǐng)后竟然沒(méi)有被釋放，為了驗(yàn)證分析，在業(yè)務(wù)代碼中調(diào)用ReferenceCountUtil的release方法進(jìn)行內(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ù)進(jìn)行壓測(cè)，發(fā)現(xiàn)系統(tǒng)運(yùn)行平穩(wěn)，沒(méi)有發(fā)生OOM異常。對(duì)內(nèi)存活動(dòng)對(duì)象進(jìn)行排序，沒(méi)有再發(fā)現(xiàn)大量的PoolChunk對(duì)象，內(nèi)存泄漏問(wèn)題解決，問(wèn)題修復(fù)之后的內(nèi)存快照如下圖所示。

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

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

通過(guò)前面的案例分析和驗(yàn)證，我們可以看出這個(gè)觀點(diǎn)是錯(cuò)誤的。為了在實(shí)際項(xiàng)目中更好地管理ByteBuf，下面我們分4種場(chǎng)景進(jìn)行說(shuō)明。

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

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

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

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

對(duì)案例中的問(wèn)題代碼進(jìn)行修改，繼承自SimpleChannelInboundHandler，即便業(yè)務(wù)不釋放請(qǐng)求的ByteBuf對(duì)象，依然不會(huì)發(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(()-> { 
        //解析請(qǐng)求消息，做路由轉(zhuǎn)發(fā)，代碼省略
        //轉(zhuǎn)發(fā)成功，返回響應(yīng)給客戶(hù)端 
        ByteBuf respMsg = allocator.heapBuffer(body.length); 
        respMsg.writeBytes(body);//作為示例，簡(jiǎn)化處理，將請(qǐng)求返回 
        ctx.writeAndFlush(respMsg); 
    }); 
} 

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

策略2 在業(yè)務(wù)ChannelInboundHandler中調(diào)用ctx.fireChannelRead(msg)方法，讓請(qǐng)求消息繼續(xù)向后執(zhí)行，直到調(diào)用DefaultChannelPipeline的內(nèi)部類(lèi)TailContext，由它來(lái)負(fù)責(zé)釋放請(qǐng)求消息，代碼如下（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)存池的請(qǐng)求ByteBuf

如果業(yè)務(wù)使用非內(nèi)存池模式覆蓋Netty默認(rèn)的內(nèi)存池模式創(chuàng)建請(qǐng)求ByteBuf，例如通過(guò)如下代碼修改內(nèi)存申請(qǐng)策略為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ā)送完成后都會(huì)由Netty框架進(jìn)行內(nèi)存釋放，業(yè)務(wù)不需要主動(dòng)釋放內(nèi)存。

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

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

總結(jié)

以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。