ByteBufAllocator 源码分析（三十四）

àì夳堔傛蜴生んèń 2022-10-29 14:23 140阅读 0赞

今天进行 ByteBufAllocator 源码分析：

**1、ByteBufAllocator**

ByteBufAllocator  是  Netty  内存分配最顶层的抽象，负责分配所有类型的内存。

ByteBuf buffer(); // 直接分配一块内存，是使用  direct  还是  heap  取决于子类实现。

ByteBuf ioBuffer(); // 尽可能地分配一块堆外直接内存，如果系统不支持则分配堆内内存 ByteBuf heapBuffer();// heap  内存分配。

ByteBuf directBuffer();// direct  内存分配。

在  ByteBufAllocator  中还指明  Netty  中的缺省  ByteBufAllocator ，默认情况下，在操作系统非 Android  的情况下，都是使用的  PooledByteBufAllocator ，当然可以使用参数

\-Dio.netty.allocator.type  进行修改。

ByteBufAllocator  的直接实现子类  PreferHeapByteBufAllocator  有个 @UnstableApi  注解，表明这个实现类是不稳定的，我们不做研究。

***2、AbstractByteBufAllocator***

AbstractByteBufAllocator  是  ByteBufAllocator  的骨架实现（类似于  AbstractByteBuf  和 ByteBuf 的关系），它提供了两个重要的抽象方法供子类扩展，以实现堆内和堆外内存的创建。

/**
     * Create a heap {@link ByteBuf} with the given initialCapacity and maxCapacity.
     */
    protected abstract ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity);
    
    /**
     * Create a direct {@link ByteBuf} with the given initialCapacity and maxCapacity.
     */
    protected abstract ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity);

而且 AbstractByteBufAllocator 对 ByteBufAllocator 中方法的实现，基本上也都可以归结到对这两个抽象方法的调用上。

***3、UnpooledByteBufAllocator***

从名字可知， UnpooledByteBufAllocator  是非池化分配器，我们先来看  newHeapBuffer 的实现逻辑。

@Override
    protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
        return PlatformDependent.hasUnsafe() ?
                new InstrumentedUnpooledUnsafeHeapByteBuf(this, initialCapacity, maxCapacity) :
                new InstrumentedUnpooledHeapByteBuf(this, initialCapacity, maxCapacity);
    }

Netty 会首先判断当前 JDK 中能否拿到 Jdk 底层去实现的 unsafe 对象，然后分情况处理，但是不管哪种情况，处理逻辑非常类似。 先看 InstrumentedUnpooledHeapByteBuf

private static final class InstrumentedUnpooledHeapByteBuf extends UnpooledHeapByteBuf {
        InstrumentedUnpooledHeapByteBuf(UnpooledByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
            super(alloc, initialCapacity, maxCapacity);
        }
    
        @Override
        protected byte[] allocateArray(int initialCapacity) {
            byte[] bytes = super.allocateArray(initialCapacity);
            ((UnpooledByteBufAllocator) alloc()).incrementHeap(bytes.length);
            return bytes;
        }
    
        @Override
        protected void freeArray(byte[] array) {
            int length = array.length;
            super.freeArray(array);
            ((UnpooledByteBufAllocator) alloc()).decrementHeap(length);
        }
    }

继续：

protected byte[] allocateArray(int initialCapacity) {
        return new byte[initialCapacity];
    }

可以看到本质上其实是分配了一个 byte  数组，并保存在  UnpooledHeapByteBuf  的  array 成员变量中。而  InstrumentedUnpooledUnsafeHeapByteBuf  本质上也是分配了一个  byte  数组，但是直接通过 JDK  中的  UNSAFE  对象来进行内存的分配和访问的。

private static final class InstrumentedUnpooledUnsafeHeapByteBuf extends UnpooledUnsafeHeapByteBuf {
        InstrumentedUnpooledUnsafeHeapByteBuf(UnpooledByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
            super(alloc, initialCapacity, maxCapacity);
        }
    
        @Override
        protected byte[] allocateArray(int initialCapacity) {
            byte[] bytes = super.allocateArray(initialCapacity);
            ((UnpooledByteBufAllocator) alloc()).incrementHeap(bytes.length);
            return bytes;
        }
    
        @Override
        protected void freeArray(byte[] array) {
            int length = array.length;
            super.freeArray(array);
            ((UnpooledByteBufAllocator) alloc()).decrementHeap(length);
        }
    }

点击 UnpooledUnsafeHeapByteBuf

@Override
    public byte getByte(int index) {
        checkIndex(index);
        return _getByte(index);
    }

继续：

@Override
    protected byte _getByte(int index) {
        return UnsafeByteBufUtil.getByte(array, index);
    }

继续：

static byte getByte(byte[] array, int index) {
        return PlatformDependent.getByte(array, index);
    }

继续：

public static byte getByte(byte[] data, int index) {
        return PlatformDependent0.getByte(data, index);
    }

继续：

static byte getByte(byte[] data, int index) {
        return UNSAFE.getByte(data, BYTE_ARRAY_BASE_OFFSET + index);
    }

newDirectBuffer 在实现逻辑上和 newHeapBuffer 高度相似，只不过底层实现上，是直接使用的 JDK 中的 ByteBuffer 来进行内存的分配的，同样的也区分是否使用 JDK 中的 UNSAFE 对象来进行。

@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);
        }
        return disableLeakDetector ? buf : toLeakAwareBuffer(buf);
    }

至于这个里面的 noCleaner 与 JDK 中对直接内存的回收有关。对直接内存的回收，JDK 提供了两种方式， 在 DirectByteBuffer 中提供了 Cleaner 用来主动释放内存，同时还有 Unsafe 的 freeMemory 方法。

***4、PooledByteBufAllocator***

从  PooledByteBufAllocator  对  newHeapBuffer  和  newDirectBuffer  的实现来看，逻辑很像：

@Override
    protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
        PoolThreadCache cache = threadCache.get();
        PoolArena<byte[]> heapArena = cache.heapArena;
    
        final ByteBuf buf;
        if (heapArena != null) {
            buf = heapArena.allocate(cache, initialCapacity, maxCapacity);
        } else {
            buf = PlatformDependent.hasUnsafe() ?
                    new UnpooledUnsafeHeapByteBuf(this, initialCapacity, maxCapacity) :
                    new UnpooledHeapByteBuf(this, initialCapacity, maxCapacity);
        }
    
        return toLeakAwareBuffer(buf);
    }
    
    @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 {
            buf = PlatformDependent.hasUnsafe() ?
                    UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity) :
                    new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
        }
    
        return toLeakAwareBuffer(buf);
    }

两者都不是直接去系统申请，而是先从 threadCache 对象获取一个线程局部缓存对象，再获取其中的 PoolArena 对象，由 PoolArena 对象来为 ByteBuf 分配内存。具体的 PoolArena 等等相关类在讲到 PooledByteBuf 时会有描述。

今天分享 ByteBufAllocator 源码到此结束，下篇分享PooledByteBuf 源码，敬请期待！