Netty 如何自动探测内存泄露的发生

public final class UnpooledByteBufAllocator {    @Override    protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {        final ByteBuf buf;        if (PlatformDependent.hasUnsafe()) {            buf = noCleaner ? new InstrumentedUnpooledUnsafeNoCleanerDirectByteBuf(this, initialCapacity, maxCapacity) :                    new InstrumentedUnpooledUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);        } else {            buf = new InstrumentedUnpooledDirectByteBuf(this, initialCapacity, maxCapacity);        }        // 是否启动内存泄露探测，如果启动则额外用 LeakAwareByteBuf 进行包装返回        return disableLeakDetector ? buf : toLeakAwareBuffer(buf);    }}

public class PooledByteBufAllocator {    // 线程本地缓存    private final PoolThreadLocalCache threadCache;
    @Override    protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {        // 获取线程本地缓存，线程第一次申请内存的时候会在这里与 PoolArena 进行绑定        PoolThreadCache cache = threadCache.get();        // 获取与当前线程绑定的 PoolArena        PoolArena<ByteBuffer> directArena = cache.directArena;
        final ByteBuf buf;        if (directArena != null) {            // 从固定的 PoolArena 中申请内存            buf = directArena.allocate(cache, initialCapacity, maxCapacity);        } else {            // 申请非池化内存            buf = PlatformDependent.hasUnsafe() ?                    UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity) :                    new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);        }        // 如果内存泄露探测开启，则用 LeakAwareByteBuf 包装 PooledByteBuf 返回        return toLeakAwareBuffer(buf);    }}

public class Cleaner extends PhantomReference<Object>

        Thread handler = new ReferenceHandler(tg, "Reference Handler");        // 设置 ReferenceHandler 线程的优先级为最高优先级        handler.setPriority(Thread.MAX_PRIORITY);        handler.setDaemon(true);

  private static class Deallocator implements Runnable {        public void run() {                // 底层调用 free 来释放 native memory            UNSAFE.freeMemory(address);        }  }

    private static final class FreeOnFinalize {        // 待释放的 PoolThreadCache        private volatile PoolThreadCache cache;
        private FreeOnFinalize(PoolThreadCache cache) {            this.cache = cache;        }
        @Override        protected void finalize() throws Throwable {            try {                super.finalize();            } finally {                PoolThreadCache cache = this.cache;                this.cache = null;                // 当 FreeOnFinalize 实例要被回收的时候，触发 PoolThreadCache 的释放                if (cache != null) {                    cache.free(true);                }            }        }    }

final class Finalizer extends FinalReference<Object> { 
    private static ReferenceQueue<Object> queue = new ReferenceQueue<>();
    private Finalizer(Object finalizee) {        // 这里的 finalizee 就是 FreeOnFinalize 对象，被 FinalReference 引用        super(finalizee, queue);              ......    }}

   Reference(T referent, ReferenceQueue<? super T> queue) {        // FreeOnFinalize 对象        this.referent = referent;        //  Finalizer 中的 ReferenceQueue 实例（全局）        this.queue = (queue == null) ? ReferenceQueue.NULL : queue;    }

        Thread finalizer = new FinalizerThread(tg);        finalizer.setPriority(Thread.MAX_PRIORITY - 2);        finalizer.setDaemon(true);        finalizer.start();

final class DefaultResourceLeak<T> extends WeakReference<Object>

public class ResourceLeakDetector<T> {    private final ReferenceQueue<Object> refQueue = new ReferenceQueue<Object>();}

public abstract class AbstractReferenceCountedByteBuf extends AbstractByteBuf {   // 引用计数   private volatile int refCnt;}

public interface ReferenceCounted {     int refCnt();     ReferenceCounted retain();     boolean release();}

public class Cleaner extends PhantomReference<Object>{    private static Cleaner first = null;
    private Cleaner next = null, prev = null;}

final class Finalizer extends FinalReference<Object> {     // 双向链表，保存 JVM 堆中所有的 Finalizer 对象，防止 Finalizer 被 GC 掉    private static Finalizer unfinalized = null;    private Finalizer next, prev;}

public class ResourceLeakDetector<T> {    private final Set<DefaultResourceLeak<?>> allLeaks =            Collections.newSetFromMap(new ConcurrentHashMap<DefaultResourceLeak<?>, Boolean>());}

    public enum Level {        DISABLED,        SIMPLE,        ADVANCED,        PARANOID;    }

public class ResourceLeakDetector<T> {  // 采样间隔，默认 128  static final int SAMPLING_INTERVAL;
  private static final String PROP_SAMPLING_INTERVAL = "io.netty.leakDetection.samplingInterval";
  private static final int DEFAULT_SAMPLING_INTERVAL = 128;
  SAMPLING_INTERVAL = SystemPropertyUtil.getInt(PROP_SAMPLING_INTERVAL, DEFAULT_SAMPLING_INTERVAL);}

PlatformDependent.threadLocalRandom().nextInt(samplingInterval) == 0

private static final class DefaultResourceLeak<T> {    // 栈顶指针    private volatile TraceRecord head; // 栈结构，存放对应 ByteBuf 的访问堆栈}
private static class TraceRecord extends Throwable {  // 栈底  private static final TraceRecord BOTTOM = new TraceRecord()}

public class ResourceLeakDetector<T> {    // ByteBuf 访问堆栈记录个数限制，默认为 4    private static final int TARGET_RECORDS;
    private static final String PROP_TARGET_RECORDS = "io.netty.leakDetection.targetRecords";
    private static final int DEFAULT_TARGET_RECORDS = 4;
    TARGET_RECORDS = SystemPropertyUtil.getInt(PROP_TARGET_RECORDS, DEFAULT_TARGET_RECORDS);}

public class ResourceLeakDetector<T> {    // 探测级别    private static Level level;    // 未被释放的 ByteBuf 对应的弱引用 DefaultResourceLeak 集合    private final Set<DefaultResourceLeak<?>> allLeaks =            Collections.newSetFromMap(new ConcurrentHashMap<DefaultResourceLeak<?>, Boolean>());    // 用于接收 ByteBuf 被回收的通知    private final ReferenceQueue<Object> refQueue = new ReferenceQueue<Object>();    // 探测的资源类型，这里是 ByteBuf    private final String resourceType;    // 采样间隔    private final int samplingInterval;    // 内存泄露监听器，一旦探测到内存泄露，Netty 就会回调 LeakListener    private volatile LeakListener leakListener;}

    public interface LeakListener {
        /**         * Will be called once a leak is detected.         */        void onLeak(String resourceType, String records);    }

public final class ByteBufUtil {
    public static void setLeakListener(ResourceLeakDetector.LeakListener leakListener) {        AbstractByteBuf.leakDetector.setLeakListener(leakListener);    }}

public abstract class AbstractByteBuf extends ByteBuf {    // 全局 ResourceLeakDetector 实例    static final ResourceLeakDetector<ByteBuf> leakDetector =            ResourceLeakDetectorFactory.instance().newResourceLeakDetector(ByteBuf.class);}

public class ResourceLeakDetector<T> {
    public final ResourceLeakTracker<T> track(T obj) {        return track0(obj, false);    }
    // 采样频率，默认 128    private final int samplingInterval;    // 对 obj 进行资源泄露的探测    // force 表示是否强制探测    private DefaultResourceLeak track0(T obj, boolean force) {        Level level = ResourceLeakDetector.level;        if (force ||                level == Level.PARANOID ||                (level != Level.DISABLED && PlatformDependent.threadLocalRandom().nextInt(samplingInterval) == 0)) {            // 触发内存泄露探测，如果发生内存泄露，则在日志中 report            reportLeak();            // 创建 ByteBuf (obj) 对应的弱引用 DefaultResourceLeak            // ResourceLeakDetector 中的全局 refQueue ， allLeaks 会在这里注册进去            return new DefaultResourceLeak(obj, refQueue, allLeaks, getInitialHint(resourceType));        }        return null;    }}

PlatformDependent.threadLocalRandom().nextInt(samplingInterval) == 0

    protected boolean needReport() {        return logger.isErrorEnabled();    }

    private void clearRefQueue() {        for (;;) {            // 清空 refQueue            DefaultResourceLeak ref = (DefaultResourceLeak) refQueue.poll();            if (ref == null) {                break;            }            // 将 DefaultResourceLeak 从 allLeaks 集合中删除。            ref.dispose();        }    }

DefaultResourceLeak ref = (DefaultResourceLeak) refQueue.poll();

        boolean dispose() {            // 断开 DefaultResourceLeak 与 ByteBuf 的弱引用关联            clear();            // 检查 DefaultResourceLeak 实例是否仍然存在于 allLeaks 集合中。            return allLeaks.remove(this);        }

    // resourceType 为需要探测的资源类型，这里是 ByteBuf    // records 是发生内存泄露的 ByteBuf 相关的访问堆栈    protected void reportTracedLeak(String resourceType, String records) {        logger.error(                "LEAK: {}.release() was not called before it's garbage-collected. " +                "See https://netty.io/wiki/reference-counted-objects.html for more information.{}",                resourceType, records);    }

    private void reportLeak() {        // 日志级别必须是 Error 级别        if (!needReport()) {            clearRefQueue();            return;        }
        // Detect and report previous leaks.        for (;;) {            // 对应的 ByteBuf 必须已经被 GC 回收，才会触发内存泄露的探测            DefaultResourceLeak ref = (DefaultResourceLeak) refQueue.poll();            if (ref == null) {                break;            }            // 检查 ByteBuf 对应的 DefaultResourceLeak 是否仍然停留在 allLeaks 集合中            if (!ref.dispose()) {                // 如果不存在，则说明 ByteBuf 已经被及时的释放了，不存在内存泄露                continue;            }            // 当探测到 ByteBuf 发生内存泄露之后，这里会获取 ByteBuf 相关的访问堆栈             String records = ref.getReportAndClearRecords();            if (reportedLeaks.add(records)) { // 去重泄露日志                // 打印泄露的堆栈路径                if (records.isEmpty()) {                    reportUntracedLeak(resourceType);                } else {                    reportTracedLeak(resourceType, records);                }                // 回调 LeakListener                LeakListener listener = leakListener;                if (listener != null) {                    listener.onLeak(resourceType, records);                }            }        }    }

    private static final class DefaultResourceLeak<T>            extends WeakReference<Object> implements ResourceLeakTracker<T>, ResourceLeak {        // 栈顶指针        private volatile TraceRecord head;        // 栈中被丢弃的 TraceRecord 个数        private volatile int droppedRecords;        // 指向 ResourceLeakDetector 中的全局 allLeaks        private final Set<DefaultResourceLeak<?>> allLeaks;        // 被追踪探测的 Bytebuf 的 hash 值        private final int trackedHash;    }

        DefaultResourceLeak(                Object referent,                ReferenceQueue<Object> refQueue,                Set<DefaultResourceLeak<?>> allLeaks,                Object initialHint) {            // 弱引用关联 ByteBuf （referent）            // 注册 refQueue            super(referent, refQueue);            // 保存 Bytebuf 的 hash 值            trackedHash = System.identityHashCode(referent);            // 加入到 allLeaks 中，如果 ByteBuf 被回收之后，DefaultResourceLeak 仍然停留在 allLeaks，则表示发生内存泄露。            allLeaks.add(this);            // 创建第一个 TraceRecord，记录 ByteBuf 的创建位置堆栈，保存在栈底            headUpdater.set(this, initialHint == null ?                    new TraceRecord(TraceRecord.BOTTOM) : new TraceRecord(TraceRecord.BOTTOM, initialHint));            this.allLeaks = allLeaks;        }

        @Override        public void record() {            record0(null);        }
        @Override        public void record(Object hint) {            record0(hint);        }

// numElements 为当前栈中的 TraceRecord 个数final int backOffFactor = Math.min(numElements - TARGET_RECORDS, 30)dropped = PlatformDependent.threadLocalRandom().nextInt(1 << backOffFactor) != 0

        private void record0(Object hint) {            if (TARGET_RECORDS > 0) {                TraceRecord oldHead;                TraceRecord prevHead;                TraceRecord newHead;                boolean dropped;                do {                    // 获取栈顶 TraceRecord，也就是 ByteBuf 最近一次的访问堆栈                    if ((prevHead = oldHead = headUpdater.get(this)) == null) {                        // 栈顶为 null ，表示 ByteBuf 已经被释放，对应的泄露探测已经关闭。                        return;                    }                    // 获取当前栈中的 TraceRecord 个数                    final int numElements = oldHead.pos + 1;                    // 如果达到 TARGET_RECORDS 的限制，就开始概率性的丢弃当前栈顶                    // 然后用新的 TraceRecord 作为栈顶                    if (numElements >= TARGET_RECORDS) {                        final int backOffFactor = Math.min(numElements - TARGET_RECORDS, 30);                        // numElements 超出 TARGET_RECORDS 的限制越多，当前栈顶就越容易被 drop                        if (dropped = PlatformDependent.threadLocalRandom().nextInt(1 << backOffFactor) != 0) {                            // 命中丢弃的概率，则将当前栈顶 TraceRecord 丢弃                            prevHead = oldHead.next;                        }                    } else {                        // 保留当前栈顶，这样栈中的 TraceRecord 个数就会超过 TARGET_RECORDS 的限制                        // 但 ByteBuf 中间的访问链路堆栈就会被概率性的保留下来                        dropped = false;                    }                    // 创建的新的 TraceRecord（记录 ByteBuf 的当前访问堆栈）                    // 并作为新的栈顶元素                    newHead = hint != null ? new TraceRecord(prevHead, hint) : new TraceRecord(prevHead);                } while (!headUpdater.compareAndSet(this, oldHead, newHead));
                if (dropped) {                     // 统计被丢弃的 TraceRecord 个数                    droppedRecordsUpdater.incrementAndGet(this);                }            }        }

        String getReportAndClearRecords() {            // 获取栈顶 TraceRecord            TraceRecord oldHead = headUpdater.getAndSet(this, null);            // 由近及远的输出 ByteBuf 相关的 TraceRecords            return generateReport(oldHead);        }

        private String generateReport(TraceRecord oldHead) {            // 当前 DefaultResourceLeak 栈中一共有多少个 TraceRecord            int present = oldHead.pos + 1;            // 每个 TraceRecord 分配 2K 大小的内存            StringBuilder buf = new StringBuilder(present * 2048).append(NEWLINE);            buf.append("Recent access records: ").append(NEWLINE);            int i = 1;            // 防重集合            Set<String> seen = new HashSet<String>(present);            // 从栈顶开始生成泄露堆栈            for (; oldHead != TraceRecord.BOTTOM; oldHead = oldHead.next) {                // 获取 TraceRecord 记录的堆栈信息                String s = oldHead.toString();                if (seen.add(s)) {                    if (oldHead.next == TraceRecord.BOTTOM) {                        // 栈底 TraceRecord 记录了 Buffer 的创建位置                        buf.append("Created at:").append(NEWLINE).append(s);                    } else {                        buf.append('#').append(i++).append(':').append(NEWLINE).append(s);                    }                } else {                    // 重复的 TraceRecord 个数                    duped++;                }            }            // 生成泄露堆栈，并返回            buf.setLength(buf.length() - NEWLINE.length());            return buf.toString();        }

    private static class TraceRecord extends Throwable {        // 空实现，用来标识栈底位置        private static final TraceRecord BOTTOM = new TraceRecord() {            @Override            public Throwable fillInStackTrace() {                return this;            }        };        // 出现在日志中的自定义 Hint 提示信息         private final String hintString;        // 栈中下一个 TraceRecord        private final TraceRecord next;        // 当前 TraceRecord 在栈中的位置        private final int pos;    }

        @Override        public String toString() {            // 每个 TraceRecord 堆栈信息占用 2K 内存            StringBuilder buf = new StringBuilder(2048);            if (hintString != null) {                // 日志中显示我们自定义的提示信息 tHint                buf.append("\tHint: ").append(hintString).append(NEWLINE);            }
            // 获取 TraceRecord 记录的堆栈信息            StackTraceElement[] array = getStackTrace();            // Skip the first three elements.            out: for (int i = 3; i < array.length; i++) {                StackTraceElement element = array[i];
                ....... 清理一些没用的堆栈信息 ......
                // 生成有效的堆栈信息                buf.append('\t');                buf.append(element.toString());                buf.append(NEWLINE);            }            return buf.toString();        }

public final class UnpooledByteBufAllocator {    @Override    protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {        final ByteBuf buf;
        ....... 分配 UnpooledByteBuf .....
        // 是否启动内存泄露探测，如果启动则额外用 LeakAwareByteBuf 进行包装返回        return disableLeakDetector ? buf : toLeakAwareBuffer(buf);    }}
public class PooledByteBufAllocator {    @Override    protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
         ....... 分配 PooledByteBuf .....
        // 如果内存泄露探测开启，则用 LeakAwareByteBuf 包装 PooledByteBuf 返回        return toLeakAwareBuffer(buf);    }}

    protected static ByteBuf toLeakAwareBuffer(ByteBuf buf) {        // DefaultResourceLeak 用于追踪 ByteBuf 的泄露路径        ResourceLeakTracker<ByteBuf> leak;        switch (ResourceLeakDetector.getLevel()) {            case SIMPLE:                // 触发内存泄露采样探测，如果命中采样频率                // 则为 ByteBuf 创建 DefaultResourceLeak（弱引用）                leak = AbstractByteBuf.leakDetector.track(buf); // 本文 3.1 小节内容                if (leak != null) {                    // SIMPLE 级别对应的是 SimpleLeakAwareByteBuf                    buf = new SimpleLeakAwareByteBuf(buf, leak);                }                break;            case ADVANCED:            case PARANOID:                // 触发内存泄露采样探测                leak = AbstractByteBuf.leakDetector.track(buf); // 本文 3.1 小节内容                if (leak != null) {                    // ADVANCED , PARANOID 级别对应的是  AdvancedLeakAwareByteBuf                    buf = new AdvancedLeakAwareByteBuf(buf, leak);                }                break;            default:                break;        }        // 如果命中采样频率，则用 LeakAwareByteBuf 包装返回        // 如果没有命中采样频率，则原样返回        return buf;    }

class SimpleLeakAwareByteBuf extends WrappedByteBuf {   // 需要被探测的普通  ByteBuf   private final ByteBuf trackedByteBuf;   // ByteBuf 的弱引用 DefaultResourceLeak   final ResourceLeakTracker<ByteBuf> leak;
   SimpleLeakAwareByteBuf(ByteBuf wrapped, ResourceLeakTracker<ByteBuf> leak) {        this(wrapped, wrapped, leak);    }
   SimpleLeakAwareByteBuf(ByteBuf wrapped, ByteBuf trackedByteBuf, ResourceLeakTracker<ByteBuf> leak) {        super(wrapped);        this.trackedByteBuf = ObjectUtil.checkNotNull(trackedByteBuf, "trackedByteBuf");        this.leak = ObjectUtil.checkNotNull(leak, "leak");    }}

class SimpleLeakAwareByteBuf extends WrappedByteBuf {    @Override    public byte readByte() {        return trackedByteBuf.readByte();    }
    @Override    public ByteBuf writeByte(int value) {        trackedByteBuf.writeByte(value);        return this;    }}

    @Override    public boolean release() {        // 引用计数为 0         if (super.release()) {            // 关闭内存泄露的探测            closeLeak();            return true;        }        return false;    }
   private void closeLeak() {        boolean closed = leak.close(trackedByteBuf);    }

    public void clear() {        this.referent = null;    }

private static final class DefaultResourceLeak<T>            extends WeakReference<Object> implements ResourceLeakTracker<T>, ResourceLeak {
        @Override        public boolean close() {            // 将 DefaultResourceLeak 从 allLeaks 集合中删除            if (allLeaks.remove(this)) {                // 断开 DefaultResourceLeak 与 trackedByteBuf 的弱引用关联                clear();                // 清空 DefaultResourceLeak 栈                headUpdater.set(this, null);                return true;            }            return false;        }

public class ResourceLeakDetector<T> {
    private void reportLeak() {        // Detect and report previous leaks.        for (;;) {            DefaultResourceLeak ref = (DefaultResourceLeak) refQueue.poll();            if (ref == null) {                break;            }            // 检查 ByteBuf 对应的 DefaultResourceLeak 是否仍然停留在 allLeaks 集合中            if (!ref.dispose()) {                // 如果不存在，则说明 ByteBuf 已经被及时的释放了，不存在内存泄露                continue;            }            // 当探测到 ByteBuf 发生内存泄露之后，这里会获取 ByteBuf 相关的访问堆栈             String records = ref.getReportAndClearRecords();            // 打印泄露的堆栈路径            reportTracedLeak(resourceType, records);        }    }}

final class AdvancedLeakAwareByteBuf extends SimpleLeakAwareByteBuf {
    AdvancedLeakAwareByteBuf(ByteBuf buf, ResourceLeakTracker<ByteBuf> leak) {        super(buf, leak);    }
    @Override    public byte readByte() {        // 记录当前访问的堆栈信息        recordLeakNonRefCountingOperation(leak);        return super.readByte();    }
    @Override    public ByteBuf writeByte(int value) {         // 记录当前访问的堆栈信息        recordLeakNonRefCountingOperation(leak);        return super.writeByte(value);    }
    static void recordLeakNonRefCountingOperation(ResourceLeakTracker<ByteBuf> leak) {        if (!ACQUIRE_AND_RELEASE_ONLY) {            // 向 DefaultResourceLeak 添加新的堆栈            leak.record();        }    }}

private static final String PROP_ACQUIRE_AND_RELEASE_ONLY = "io.netty.leakDetection.acquireAndReleaseOnly";
ACQUIRE_AND_RELEASE_ONLY = SystemPropertyUtil.getBoolean(PROP_ACQUIRE_AND_RELEASE_ONLY, false);

    @Override    public ByteBuf touch() {        leak.record();        return this;    }
    @Override    public ByteBuf touch(Object hint) {        leak.record(hint);        return this;    }
    @Override    public ByteBuf retain() {        leak.record();        return super.retain();    }
    @Override    public boolean release() {        leak.record();        return super.release();    }

class SimpleLeakAwareByteBuf extends WrappedByteBuf {    @Override    public ByteBuf touch() {        return this;    }
    @Override    public ByteBuf touch(Object hint) {        return this;    }}