在io.netty.buffer.AbstractByteBufAllocator類中有如下程式碼

//裝飾器模式,用SimpleLeakAwareByteBuf或AdvancedLeakAwareByteBuf來包裝原始的ByteBuf
//兩個包裝類均通過呼叫ResourceLeak的record方法來記錄ByteBuf的方法呼叫堆疊,區別在於後者比前者記錄更多的內容
protected static ByteBuf toLeakAwareBuffer(ByteBuf buf) {
    ResourceLeak leak;
    //根據設定的Level來選擇使用何種包裝器
    switch (ResourceLeakDetector.getLevel()) {
        case SIMPLE:
            //建立用於跟蹤和表示內容洩露的ResourcLeak物件
            leak = AbstractByteBuf.leakDetector.open(buf);
            if (leak != null) {
                //只在ByteBuf.order方法中呼叫ResourceLeak.record
                buf = new SimpleLeakAwareByteBuf(buf, leak);
            }
            break;
        case ADVANCED:
        case PARANOID:
            leak = AbstractByteBuf.leakDetector.open(buf);
            if (leak != null) {
                //在ByteBuf幾乎所有方法中呼叫ResourceLeak.record  
                buf = new AdvancedLeakAwareByteBuf(buf, leak);
            }
            break;
    }
    return buf;
}
 

下圖展示了該方法被呼叫的時機.可見Netty只對PooledByteBuf和DirectByteBuf監控記憶體洩露.

記憶體洩露檢測

下面觀察上述程式碼中的AbstractByteBuf.leakDetector.open(buf);

//建立用於跟蹤和表示內容洩露的ResourcLeak物件
public ResourceLeak open(T obj) {
    Level level = ResourceLeakDetector.level;
    if (level == Level.DISABLED) {//禁用記憶體跟蹤
        return null;
    }
    if (level.ordinal() < Level.PARANOID.ordinal()) {
        //如果監控級別低於PARANOID,在一定的取樣頻率下報告記憶體洩露
        if (leakCheckCnt ++ % samplingInterval == 0) {
            reportLeak(level);
            return new DefaultResourceLeak(obj);
        } else {
            return null;
        }
    } else {
        //每次需要分配 ByteBuf 時,報告記憶體洩露情況
        reportLeak(level);
        return new DefaultResourceLeak(obj);
    }
}
 

其中reportLeak方法中完成對記憶體洩露的檢測和報告,如下面程式碼所示.

private void reportLeak(Level level) {
    //......

    // 報告生成了太多的活躍資源
    int samplingInterval = level == Level.PARANOID? 1 : this.samplingInterval;
    if (active * samplingInterval > maxActive && loggedTooManyActive.compareAndSet(false, true)) {
        logger.error("LEAK: You are creating too many " + resourceType + " instances.  " +
                resourceType + " is a shared resource that must be reused across the JVM," +
                "so that only a few instances are created.");
    }

    // 檢測並報告之前發生的記憶體洩露
    for (;;) {
        @SuppressWarnings("unchecked")
        //檢查引用佇列(為什麼通過檢查該佇列,可以判斷是否存在記憶體洩露)
        DefaultResourceLeak ref = (DefaultResourceLeak) refQueue.poll();
        if (ref == null) {//佇列為空,沒有未報告的記憶體洩露或者從未發生記憶體洩露
            break;
        }
<span style="white-space:pre">	</span>//清理引用
        ref.clear();

        if (!ref.close()) {
            continue;
        }
        //通過錯誤日誌列印資源的方法呼叫記錄,並將其儲存在reportedLeaks中
        String records = ref.toString();
        if (reportedLeaks.putIfAbsent(records, Boolean.TRUE) == null) {
            if (records.isEmpty()) {
                logger.error("LEAK: {}.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 '-D{}={}' or call {}.setLevel()",
                        resourceType, PROP_LEVEL, Level.ADVANCED.name().toLowerCase(), simpleClassName(this));
            } else {
                logger.error(
                        "LEAK: {}.release() was not called before it's garbage-collected.{}",
                        resourceType, records);
            }
        }
    }
}

DefaultResourceLeak的宣告如下

private final class DefaultResourceLeak extends PhantomReference<Object> implements ResourceLeak{
    //......

    public DefaultResourceLeak(Object referent) {
        //使用一個靜態的引用佇列(refQueue)初始化
        //refQueue是ResourceLeakDecetor的成員變數並由其初始化
        super(referent, referent != null? refQueue : null);
        //......
    }

    //......
}