Skip to content

netty eventloop pooled_direct

Jump to bottom
wangkaiwu edited this page Feb 13, 2023 · 1 revision

PooledDirectByteBuffer

构造器

仅有一个默认构造

final class PooledUnsafeDirectByteBuf extends PooledByteBuf<ByteBuffer> {
    private PooledUnsafeDirectByteBuf(Handle<PooledUnsafeDirectByteBuf> recyclerHandle, int maxCapacity) {
        super(recyclerHandle, maxCapacity);
    }
}
abstract class PooledByteBuf<T> extends AbstractReferenceCountedByteBuf {
    private final Handle<PooledByteBuf<T>> recyclerHandle;

    protected PooledByteBuf(Handle<? extends PooledByteBuf<T>> recyclerHandle, int maxCapacity) {
        super(maxCapacity);
        this.recyclerHandle = (Handle<PooledByteBuf<T>>) recyclerHandle;
    }
}
public abstract class AbstractReferenceCountedByteBuf extends AbstractByteBuf {
    protected AbstractReferenceCountedByteBuf(int maxCapacity) {
        super(maxCapacity);
    }
}
public abstract class AbstractByteBuf extends ByteBuf {
    private int maxCapacity;

    protected AbstractByteBuf(int maxCapacity) {
        this.maxCapacity = maxCapacity;
    }
}

初始化流程

final class PooledUnsafeDirectByteBuf extends PooledByteBuf<ByteBuffer> {
    private static final ObjectPool<PooledUnsafeDirectByteBuf> RECYCLER = ObjectPool.newPool(
            new ObjectCreator<PooledUnsafeDirectByteBuf>() {
                @Override
                public PooledUnsafeDirectByteBuf newObject(Handle<PooledUnsafeDirectByteBuf> handle) {
                    // 调用构造器
                    return new PooledUnsafeDirectByteBuf(handle, 0);
                }
            });
}
public abstract class ObjectPool<T> {
    private static final class RecyclerObjectPool<T> extends ObjectPool<T> {
        private final Recycler<T> recycler;

        RecyclerObjectPool(final ObjectCreator<T> creator) {
            recycler = new Recycler<T>() {
                @Override
                protected T newObject(Handle<T> handle) {
                    // 调用了上面的方法
                    return creator.newObject(handle);
                }
            };
        }
    }
}
public abstract class Recycler<T> {
    private final int maxCapacityPerThread;
    private final FastThreadLocal<Stack<T>> threadLocal = new FastThreadLocal<Stack<T>>() {
    };

    public final T get() {
        if (maxCapacityPerThread == 0) {
            // 单个线程的最大容量
            return newObject((Handle<T>) NOOP_HANDLE);
        }
        // 线程局部变量里的栈结构
        Stack<T> stack = threadLocal.get();
        DefaultHandle<T> handle = stack.pop();
        if (handle == null) {
            // 说明没有，生成一个
            handle = stack.newHandle();
            handle.value = newObject(handle);
        }
        return (T) handle.value;
    }
}
public abstract class ObjectPool<T> {
    private static final class RecyclerObjectPool<T> extends ObjectPool<T> {
        @Override
        public T get() {
            // 调用上面的 get 方法
            return recycler.get();
        }
    }
}
final class PooledUnsafeDirectByteBuf extends PooledByteBuf<ByteBuffer> {
    static PooledUnsafeDirectByteBuf newInstance(int maxCapacity) {
        // 产生一个 buffer
        PooledUnsafeDirectByteBuf buf = RECYCLER.get();
        // ......
        return buf;
    }
}
abstract class PoolArena<T> implements PoolArenaMetric {
    static final class DirectArena extends PoolArena<ByteBuffer> {
        @Override
        protected PooledByteBuf<ByteBuffer> newByteBuf(int maxCapacity) {
            if (HAS_UNSAFE) {
                // 调用上面的方法
                return PooledUnsafeDirectByteBuf.newInstance(maxCapacity);
            } else {
                return PooledDirectByteBuf.newInstance(maxCapacity);
            }
        }
    }
}
abstract class PoolArena<T> implements PoolArenaMetric {
    PooledByteBuf<T> allocate(PoolThreadCache cache, int reqCapacity, int maxCapacity) {
        PooledByteBuf<T> buf = newByteBuf(maxCapacity);
        // TODO：
        allocate(cache, buf, reqCapacity);
        return buf;
    }
}
public class PooledByteBufAllocator extends AbstractByteBufAllocator implements ByteBufAllocatorMetricProvider {
    private final PoolThreadLocalCache threadCache;

    @Override
    protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
        PoolThreadCache cache = threadCache.get();
        PoolArena<ByteBuffer> directArena = cache.directArena;
        final ByteBuf buf;
        if (directArena != null) {
            buf = directArena.allocate(cache, initialCapacity, maxCapacity);
        } else {
            // ...........
        }
        // TODO:
        return toLeakAwareBuffer(buf);
    }
}
public abstract class AbstractByteBufAllocator implements ByteBufAllocator {
    @Override
    public ByteBuf directBuffer(int initialCapacity, int maxCapacity) {
        return newDirectBuffer(initialCapacity, maxCapacity);
    }
}
public abstract class AbstractByteBufAllocator implements ByteBufAllocator {
    static final int DEFAULT_MAX_CAPACITY = Integer.MAX_VALUE;

    @Override
    public ByteBuf directBuffer(int initialCapacity) {
        return directBuffer(initialCapacity, DEFAULT_MAX_CAPACITY);
    }
}
public abstract class AbstractByteBufAllocator implements ByteBufAllocator {
    @Override
    public ByteBuf ioBuffer(int initialCapacity) {
        if (PlatformDependent.hasUnsafe() || isDirectBufferPooled()) {
            // 支持 unsafe 且启用了 pool
            return directBuffer(initialCapacity);
        }
        return heapBuffer(initialCapacity);
    }
}
public abstract class DefaultMaxMessagesRecvByteBufAllocator implements MaxMessagesRecvByteBufAllocator {
    @Override
    public ByteBuf allocate(ByteBufAllocator alloc) {
        // TODO:
        return alloc.ioBuffer(guess());
    }
}

通过下面的位置，确认了当 channel 收到新消息到时候，就会先分配一个 byteBuffer，大小由算法决定。

protected class NioByteUnsafe extends AbstractNioUnsafe {
    @Override
    public final void read() {
        // ........
        ByteBuf byteBuf = null;
        // .........
        byteBuf = allocHandle.allocate(allocator);
        // .........
    }
}

归还流程

public abstract class Recycler<T> {
    private static final class DefaultHandle<T> implements Handle<T> {
        Stack<?> stack;

        DefaultHandle(Stack<?> stack) {
            this.stack = stack;
        }

        @Override
        public void recycle(Object object) {
            Stack<?> stack = this.stack;
            stack.push(this);
        }
    }
}
abstract class PooledByteBuf<T> extends AbstractReferenceCountedByteBuf {
    private void recycle() {
        recyclerHandle.recycle(this);
    }
}
abstract class PooledByteBuf<T> extends AbstractReferenceCountedByteBuf {
    protected PoolChunk<T> chunk;
    protected long handle;
    protected T memory;

    @Override
    protected final void deallocate() {
        if (handle >= 0) {
            final long handle = this.handle;
            this.handle = -1;
            memory = null;
            chunk.arena.free(chunk, tmpNioBuf, handle, maxLength, cache);
            chunk = null;
            recycle();
        }
    }
}
public abstract class AbstractReferenceCountedByteBuf extends AbstractByteBuf {
    private boolean handleRelease(boolean result) {
        if (result) {
            deallocate();
        }
        return result;
    }
}
public abstract class AbstractReferenceCountedByteBuf extends AbstractByteBuf {
    @Override
    public boolean release() {
        // TODO:
        return handleRelease(updater.release(this));
    }
}
public final class ReferenceCountUtil {
    public static boolean release(Object msg) {
        if (msg instanceof ReferenceCounted) {
            return ((ReferenceCounted) msg).release();
        }
        return false;
    }
}
public final class ReferenceCountUtil {
    public static void safeRelease(Object msg) {
        release(msg);
    }
}

当buffer 写出成功后，会触发 release

public final class ChannelOutboundBuffer {
    private Entry flushedEntry;

    public boolean remove() {
        Entry e = flushedEntry;
        //........
        Object msg = e.msg;
        //.........
        if (!e.cancelled) {
            ReferenceCountUtil.safeRelease(msg);
            // ........
        }
        //.......
        return true;
    }
}

读数据

public abstract class AbstractByteBuf extends ByteBuf {
    int readerIndex;
    @Override
    public byte readByte() {
        int i = readerIndex;
        byte b = _getByte(i);
        readerIndex = i + 1;
        return b;
    }
}
final class PooledUnsafeDirectByteBuf extends PooledByteBuf<ByteBuffer> {
    private long memoryAddress;
    @Override
    protected byte _getByte(int index) {
        return UnsafeByteBufUtil.getByte(addr(index));
    }
    private long addr(int index) {
        // 存在一个内存基址
        return memoryAddress + index;
    }
}
final class UnsafeByteBufUtil {
    static byte getByte(long address) {
        return PlatformDependent.getByte(address);
    }
}
public final class PlatformDependent {
    public static byte getByte(long address) {
        return PlatformDependent0.getByte(address);
    }
}
final class PlatformDependent0 {
    static final Unsafe UNSAFE;

    static byte getByte(long address) {
        return UNSAFE.getByte(address);
    }
}

可以看到能获取到数据的核心是 memoryAddress，接下来看下 PoolArena#allocate() 方法，会对 memoryAddress 进行赋值

PoolArena#allocate()

abstract class PoolArena<T> implements PoolArenaMetric {
    PooledByteBuf<T> allocate(PoolThreadCache cache, int reqCapacity, int maxCapacity) {
        PooledByteBuf<T> buf = newByteBuf(maxCapacity);
        // 在此处调用了 allocate，是在分配新的 ByteBuf 之后
        allocate(cache, buf, reqCapacity);
        return buf;
    }
}
abstract class PoolArena<T> implements PoolArenaMetric {
    // 默认值是： -8192。 0x1111_1111_1111_1111_1110_0000_0000_0000
    final int subpageOverflowMask;

    private void allocate(PoolThreadCache cache, PooledByteBuf<T> buf, final int reqCapacity) {
        // 计算实际的大小
        final int normCapacity = normalizeCapacity(reqCapacity);
        if (isTinyOrSmall(normCapacity)) {
            // 容量小于页大小
            int tableIdx;
            PoolSubpage<T>[] table;
            boolean tiny = isTiny(normCapacity);
            if (tiny) { // < 512
                // TODO:
                if (cache.allocateTiny(this, buf, reqCapacity, normCapacity)) {
                    return;
                }
                tableIdx = tinyIdx(normCapacity);
                table = tinySubpagePools;
            } else {
                // .........
            }
            final PoolSubpage<T> head = table[tableIdx];
            synchronized (this) {
                // 调用了 memory 的赋值
                allocateNormal(buf, reqCapacity, normCapacity, cache);
            }
            incTinySmallAllocation(tiny);
            return;
        }
        //.........
    }

    boolean isTinyOrSmall(int normCapacity) {
        // 范围大概是 0 < ? < 8192。
        return (normCapacity & subpageOverflowMask) == 0;
    }

    // ? < 512
    static boolean isTiny(int normCapacity) {
        return (normCapacity & 0xFFFFFE00) == 0;
    }
}
abstract class PoolArena<T> implements PoolArenaMetric {
    private final PoolChunkList<T> qInit;

    private void allocateNormal(PooledByteBuf<T> buf, int reqCapacity, int normCapacity, PoolThreadCache threadCache) {
        //.......
        // TODO:
        PoolChunk<T> c = newChunk(pageSize, maxOrder, pageShifts, chunkSize);
        boolean success = c.allocate(buf, reqCapacity, normCapacity, threadCache);
        qInit.add(c);
    }
}
final class PoolChunk<T> implements PoolChunkMetric {
    boolean allocate(PooledByteBuf<T> buf, int reqCapacity, int normCapacity, PoolThreadCache threadCache) {
        final long handle;
        if ((normCapacity & subpageOverflowMask) != 0) {
            //.........
        } else {
            handle = allocateSubpage(normCapacity); 
        }
        // nioBuffer = null
        initBuf(buf, nioBuffer, handle, reqCapacity, threadCache);
        return true;
    }
}
final class PoolChunk<T> implements PoolChunkMetric {
    void initBuf(PooledByteBuf<T> buf, ByteBuffer nioBuffer, long handle, int reqCapacity,
                 PoolThreadCache threadCache) {
        int memoryMapIdx = memoryMapIdx(handle);
        int bitmapIdx = bitmapIdx(handle);
        if (bitmapIdx == 0) {
            // .......
        } else {
            initBufWithSubpage(buf, nioBuffer, handle, bitmapIdx, reqCapacity, threadCache);
        }
    }
}
final class PoolChunk<T> implements PoolChunkMetric {
    private void initBufWithSubpage(PooledByteBuf<T> buf, ByteBuffer nioBuffer,
                                    long handle, int bitmapIdx, int reqCapacity, PoolThreadCache threadCache) {
        int memoryMapIdx = memoryMapIdx(handle);
        PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];
        buf.init(
                this, nioBuffer, handle,
                runOffset(memoryMapIdx) + (bitmapIdx & 0x3FFFFFFF) * subpage.elemSize + offset,
                reqCapacity, subpage.elemSize, threadCache);
    }
}
final class PooledUnsafeDirectByteBuf extends PooledByteBuf<ByteBuffer> {
    private long memoryAddress;
    @Override
    void init(PoolChunk<ByteBuffer> chunk, ByteBuffer nioBuffer,
              long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
        super.init(chunk, nioBuffer, handle, offset, length, maxLength, cache);
        initMemoryAddress();
    }
    private void initMemoryAddress() {
        // 最终在这里完成了对 memoryAddress 的赋值
        memoryAddress = PlatformDependent.directBufferAddress(memory) + offset;
    }
}

推导 memory 和 offset 的赋值

先看赋值的方法

abstract class PooledByteBuf<T> extends AbstractReferenceCountedByteBuf {
    protected T memory;
    protected int offset;
    
    private void init0(PoolChunk<T> chunk, ByteBuffer nioBuffer,
                       long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
        memory = chunk.memory;
        this.offset = offset;
    }
}

offset 赋值

final class PoolChunk<T> implements PoolChunkMetric {
    final int offset;
    private void initBufWithSubpage(PooledByteBuf<T> buf, ByteBuffer nioBuffer,
                                    long handle, int bitmapIdx, int reqCapacity, PoolThreadCache threadCache) {
        int memoryMapIdx = memoryMapIdx(handle);
        PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];
        buf.init(
                this, nioBuffer, handle,
                // 赋值位置，太复杂了，就不看了
                runOffset(memoryMapIdx) + (bitmapIdx & 0x3FFFFFFF) * subpage.elemSize + offset,
                reqCapacity, subpage.elemSize, threadCache);
    }
}

chunk 赋值

abstract class PoolArena<T> implements PoolArenaMetric {
    final int chunkSize;

    private void allocateNormal(PooledByteBuf<T> buf, int reqCapacity, int normCapacity, PoolThreadCache threadCache) {
        //.........
        // newChunk
        PoolChunk<T> c = newChunk(pageSize, maxOrder, pageShifts, chunkSize);
        boolean success = c.allocate(buf, reqCapacity, normCapacity, threadCache);
        //.........
    }
}
abstract class PoolArena<T> implements PoolArenaMetric {
    static final class DirectArena extends PoolArena<ByteBuffer> {
        @Override
        protected PoolChunk<ByteBuffer> newChunk(int pageSize, int maxOrder,
                                                 int pageShifts, int chunkSize) {
            if (directMemoryCacheAlignment == 0) {
                // chunk 的构造器在下面介绍
                return new PoolChunk<ByteBuffer>(this,
                        // 重点看下 allocateDirect
                        allocateDirect(chunkSize), pageSize, maxOrder,
                        pageShifts, chunkSize, 0);
            }
            // .........
        }
    }
}
abstract class PoolArena<T> implements PoolArenaMetric {
    static final class DirectArena extends PoolArena<ByteBuffer> {
        private static ByteBuffer allocateDirect(int capacity) {
            return PlatformDependent.useDirectBufferNoCleaner() ?
                    PlatformDependent.allocateDirectNoCleaner(capacity) 
                    // 由于条件是false，最终调用这里
                    : ByteBuffer.allocateDirect(capacity);
        }
    }
}

可以看到，最终还是依赖了 jdk 的 DirectByteBuffer，大小是

public abstract class ByteBuffer
        extends Buffer
        implements Comparable<ByteBuffer> {
    public static ByteBuffer allocateDirect(int capacity) {
        return new DirectByteBuffer(capacity);
    }
}

chunk 构造

final class PoolChunk<T> implements PoolChunkMetric {
    final T memory;

    PoolChunk(PoolArena<T> arena, T memory, int pageSize, int maxOrder, int pageShifts, int chunkSize, int offset) {
        //.......
        this.memory = memory;
    }
}

chunkSize 赋值

abstract class PoolArena<T> implements PoolArenaMetric {
    final int chunkSize;
    protected PoolArena(PooledByteBufAllocator parent, int pageSize,
                        int maxOrder, int pageShifts, int chunkSize, int cacheAlignment) {
        this.chunkSize = chunkSize;
    }
}
abstract class PoolArena<T> implements PoolArenaMetric {
    static final class DirectArena extends PoolArena<ByteBuffer> {
        DirectArena(PooledByteBufAllocator parent, int pageSize, int maxOrder,
                    int pageShifts, int chunkSize, int directMemoryCacheAlignment) {
            super(parent, pageSize, maxOrder, pageShifts, chunkSize,
                    directMemoryCacheAlignment);
        }
    }
}
public class PooledByteBufAllocator extends AbstractByteBufAllocator implements ByteBufAllocatorMetricProvider {
    private final PoolArena<ByteBuffer>[] directArenas;
    public PooledByteBufAllocator(boolean preferDirect, int nHeapArena, int nDirectArena, int pageSize, int maxOrder,
                                  int tinyCacheSize, int smallCacheSize, int normalCacheSize,
                                  boolean useCacheForAllThreads, int directMemoryCacheAlignment) {
        if (nDirectArena > 0) {
            // 默认是 cpu 核数的 2倍大小
            directArenas = newArenaArray(nDirectArena);
            for (int i = 0; i < directArenas.length; i ++) {
                PoolArena.DirectArena arena = new PoolArena.DirectArena(
                        this, pageSize, maxOrder, pageShifts, chunkSize, directMemoryCacheAlignment);
                directArenas[i] = arena;
                metrics.add(arena);
            }
        }
    }
}
public class PooledByteBufAllocator extends AbstractByteBufAllocator implements ByteBufAllocatorMetricProvider {
    public PooledByteBufAllocator(boolean preferDirect, int nHeapArena, int nDirectArena, int pageSize, int maxOrder,
                                  int tinyCacheSize, int smallCacheSize, int normalCacheSize,
                                  boolean useCacheForAllThreads, int directMemoryCacheAlignment) {
        chunkSize = validateAndCalculateChunkSize(pageSize, maxOrder);
    }
}
public class PooledByteBufAllocator extends AbstractByteBufAllocator implements ByteBufAllocatorMetricProvider {
    public PooledByteBufAllocator(boolean preferDirect, int nHeapArena, int nDirectArena, int pageSize, int maxOrder,
                                  int tinyCacheSize, int smallCacheSize, int normalCacheSize) {
        this(preferDirect, nHeapArena, nDirectArena, pageSize, maxOrder, tinyCacheSize, smallCacheSize,
                normalCacheSize, DEFAULT_USE_CACHE_FOR_ALL_THREADS, DEFAULT_DIRECT_MEMORY_CACHE_ALIGNMENT);
    }

    public PooledByteBufAllocator(boolean preferDirect, int nHeapArena, int nDirectArena, int pageSize, int maxOrder) {
        this(preferDirect, nHeapArena, nDirectArena, pageSize, maxOrder,
                DEFAULT_TINY_CACHE_SIZE, DEFAULT_SMALL_CACHE_SIZE, DEFAULT_NORMAL_CACHE_SIZE);
    }
    
    public PooledByteBufAllocator(boolean preferDirect) {
        this(preferDirect, DEFAULT_NUM_HEAP_ARENA, DEFAULT_NUM_DIRECT_ARENA, DEFAULT_PAGE_SIZE, DEFAULT_MAX_ORDER);
    }
}

DEFAULT_PAGE_SIZE 与 DEFAULT_MAX_ORDER， DEFAULT_NUM_DIRECT_ARENA

public class PooledByteBufAllocator extends AbstractByteBufAllocator implements ByteBufAllocatorMetricProvider {
    static {
        int defaultPageSize = SystemPropertyUtil.getInt("io.netty.allocator.pageSize", 8192);
        validateAndCalculatePageShifts(defaultPageSize);
        DEFAULT_PAGE_SIZE = defaultPageSize;
        int defaultMaxOrder = SystemPropertyUtil.getInt("io.netty.allocator.maxOrder", 11);
        validateAndCalculateChunkSize(DEFAULT_PAGE_SIZE, defaultMaxOrder);
        DEFAULT_MAX_ORDER = defaultMaxOrder;
        
        // cpu 核数的2倍
        final int defaultMinNumArena = NettyRuntime.availableProcessors() * 2;
        // 16mb
        final int defaultChunkSize = DEFAULT_PAGE_SIZE << DEFAULT_MAX_ORDER;
        // 如果 cpu 核数 小于 21，以 cpu 核数为准
        DEFAULT_NUM_DIRECT_ARENA = Math.max(0,
                SystemPropertyUtil.getInt(
                        "io.netty.allocator.numDirectArenas",
                        (int) Math.min(
                                defaultMinNumArena,
                                // 计算后的结果是 42
                                PlatformDependent.maxDirectMemory() / defaultChunkSize / 2 / 3)));
    }
}

计算 chunkSize 过程， validateAndCalculateChunkSize，根据默认值最终结果是 16777216 = 16384kb = 16Mb

public class PooledByteBufAllocator extends AbstractByteBufAllocator implements ByteBufAllocatorMetricProvider {
    // INT最大值的一半
    private static final int MAX_CHUNK_SIZE = (int) (((long) Integer.MAX_VALUE + 1) / 2);
    private static int validateAndCalculateChunkSize(int pageSize, int maxOrder) {
        if (maxOrder > 14) {
            throw new IllegalArgumentException("maxOrder: " + maxOrder + " (expected: 0-14)");
        }
        int chunkSize = pageSize;
        // 每一个 order 就会导致 chunkSize * 2
        for (int i = maxOrder; i > 0; i --) {
            if (chunkSize > MAX_CHUNK_SIZE / 2) {
                throw new IllegalArgumentException(String.format(
                        "pageSize (%d) << maxOrder (%d) must not exceed %d", pageSize, maxOrder, MAX_CHUNK_SIZE));
            }
            chunkSize <<= 1;
        }
        return chunkSize;
    }
}

maxDirectMemory 赋值

默认的 nDirectArena 是 cpu 核数的2倍，但 cpu 核数超过某个值和 maxDirectMemory 有关

public final class PlatformDependent {
    private static final Pattern MAX_DIRECT_MEMORY_SIZE_ARG_PATTERN = Pattern.compile(
            "\\s*-XX:MaxDirectMemorySize\\s*=\\s*([0-9]+)\\s*([kKmMgG]?)\\s*$");
    private static final long MAX_DIRECT_MEMORY = maxDirectMemory0();
    private static final long DIRECT_MEMORY_LIMIT;

    static {
        long maxDirectMemory = SystemPropertyUtil.getLong("io.netty.maxDirectMemory", -1);
        DIRECT_MEMORY_LIMIT = maxDirectMemory >= 1 ? maxDirectMemory : MAX_DIRECT_MEMORY;
    }

    private static long maxDirectMemory0() {
        List<String> vmArgs = (List<String>) runtimeClass.getDeclaredMethod("getInputArguments").invoke(runtime);
        for (int i = vmArgs.size() - 1; i >= 0; i --) {
            // 如果设置了 -XX:MaxDirectMemorySize, 则读取该值
            Matcher m = MAX_DIRECT_MEMORY_SIZE_ARG_PATTERN.matcher(vmArgs.get(i));
            if (!m.matches()) {
                continue;
            }
            maxDirectMemory = Long.parseLong(m.group(1));
            switch (m.group(2).charAt(0)) {
                case 'k': case 'K':
                    maxDirectMemory *= 1024;
                    break;
                case 'm': case 'M':
                    maxDirectMemory *= 1024 * 1024;
                    break;
                case 'g': case 'G':
                    maxDirectMemory *= 1024 * 1024 * 1024;
                    break;
            }
            break;
        }

        if (maxDirectMemory <= 0) {
            // 直接读取系统最大内存
            maxDirectMemory = Runtime.getRuntime().maxMemory();
        }
    }
}

关于 directArenas 和对象池的关系

public class PooledByteBufAllocator extends AbstractByteBufAllocator implements ByteBufAllocatorMetricProvider {
    final class PoolThreadLocalCache extends FastThreadLocal<PoolThreadCache> {
        @Override
        protected synchronized PoolThreadCache initialValue() {
            // 调用链下面的方法
            final PoolArena<ByteBuffer> directArena = leastUsedArena(directArenas);
            // 选择出来的 directArena 会存储到 cache 中
            final PoolThreadCache cache = new PoolThreadCache(
                    heapArena, directArena, tinyCacheSize, smallCacheSize, normalCacheSize,
                    DEFAULT_MAX_CACHED_BUFFER_CAPACITY, DEFAULT_CACHE_TRIM_INTERVAL);
            return cache;
        }

        private <T> PoolArena<T> leastUsedArena(PoolArena<T>[] arenas) {
            PoolArena<T> minArena = arenas[0];
            for (int i = 1; i < arenas.length; i++) {
                PoolArena<T> arena = arenas[i];
                //比较使用量，选择一个最少的
                if (arena.numThreadCaches.get() < minArena.numThreadCaches.get()) {
                    minArena = arena;
                }
            }
            return minArena;
        }
    }
    // 对象池的线程局部变量
    private final PoolThreadLocalCache threadCache;
    public PooledByteBufAllocator(boolean preferDirect, int nHeapArena, int nDirectArena, int pageSize, int maxOrder,
                                  int tinyCacheSize, int smallCacheSize, int normalCacheSize,
                                  boolean useCacheForAllThreads, int directMemoryCacheAlignment) {
        threadCache = new PoolThreadLocalCache(useCacheForAllThreads);
    }
}

PoolThreadCache 的构造器

final class PoolThreadCache {
    final PoolArena<ByteBuffer> directArena;
    private final MemoryRegionCache<ByteBuffer>[] normalDirectCaches;
    PoolThreadCache(PoolArena<byte[]> heapArena, PoolArena<ByteBuffer> directArena,
                    int tinyCacheSize, int smallCacheSize, int normalCacheSize,
                    int maxCachedBufferCapacity, int freeSweepAllocationThreshold) {
        this.directArena = directArena;
        if (directArena != null) {

            normalDirectCaches = createNormalCaches(
                    normalCacheSize, maxCachedBufferCapacity, directArena);
            
            directArena.numThreadCaches.getAndIncrement();
        }
    }
}

小结

  • 直接内存 ByteBuf 对象池存储在了线程局部变量中,线程结束的时候销毁

线程模型

CMS GC 的线程模型

G1 GC 的线程模型

G1 GC 相关

G1 垃圾收集器简介

G1 Young GC 启动条件解析

G1 GC 启动并发标记的解析

G1 Mixed GC 启动条件解析

Clone this wiki locally