Netty之内存分配器ByteBufAllocator

柔光的暖阳◎ 2023-06-30 04:26 157阅读 0赞

**目录**

内存分配器继承体系

ByteBufAllocator

AbstractByteBufAllocator

UnpooledByteBufAllocator

newHeapBuffer

newDirectBuffer

PooledByteBufAllocator

PoolThreadLocalCache

PoolArena

arena分配内存流程

--------------------

## 内存分配器继承体系 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEyMTIzOTQ_size_16_color_FFFFFF_t_70][]

--------------------

## ByteBufAllocator ##

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

public interface ByteBufAllocator {
    
        ByteBufAllocator DEFAULT = ByteBufUtil.DEFAULT_ALLOCATOR;
    
        // 直接分配一块内存，是使用direct还是heap取决于子类实现
        ByteBuf buffer();
    
        // 重载方法，增加初始容量
        ByteBuf buffer(int initialCapacity);
    
        // 重载方法，增加初始容量与最大容量
        ByteBuf buffer(int initialCapacity, int maxCapacity);
    
        // 更倾向于direct方式分配
        ByteBuf ioBuffer();
    
        ByteBuf ioBuffer(int initialCapacity);
    
        ByteBuf ioBuffer(int initialCapacity, int maxCapacity);
    
        // heap内存分配
        ByteBuf heapBuffer();
    
        ByteBuf heapBuffer(int initialCapacity);
    
        ByteBuf heapBuffer(int initialCapacity, int maxCapacity);
    
        // direct内存分配
        ByteBuf directBuffer();
    
        ByteBuf directBuffer(int initialCapacity);
    
        ByteBuf directBuffer(int initialCapacity, int maxCapacity);
        ...
    }

--------------------

## AbstractByteBufAllocator ##

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

// 创建堆内内存
        protected abstract ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity);
        // 创建堆外内存
        protected abstract ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity);

--------------------

## UnpooledByteBufAllocator ##

UnpooledByteBufAllocator是非池化分配器，这里着重分析一下newHeapBuffer方法与newDirectBuffer方法的实现逻辑。

@Override
        protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
            // netty会优先使用unsafe方式
            return PlatformDependent.hasUnsafe() ? new UnpooledUnsafeHeapByteBuf(this, initialCapacity, maxCapacity)
                    : new UnpooledHeapByteBuf(this, initialCapacity, maxCapacity);
        }
    
        @Override
        protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
            // netty会优先使用unsafe方式
            ByteBuf buf = PlatformDependent.hasUnsafe() ?
                    UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity) :
                    new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
    
            return disableLeakDetector ? buf : toLeakAwareBuffer(buf);
        }

--------------------

### newHeapBuffer ###

在newHeapBuffer方法中，会根据hasUnsafe的值来确定ByteBuf的类型是UnpooledUnsafeHeapByteBuf还是UnpooledHeapByteBuf。

UnpooledUnsafeHeapByteBuf是UnpooledHeapByteBuf的子类，子类构造函数调用父类构造函数，走了同一套实例化的逻辑，通过new的方式创建了指定容量的byte数组。

final class UnpooledUnsafeHeapByteBuf extends UnpooledHeapByteBuf {
        UnpooledUnsafeHeapByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
            super(alloc, initialCapacity, maxCapacity);
        }
        ...
    }
    
    public class UnpooledHeapByteBuf extends AbstractReferenceCountedByteBuf {
        protected UnpooledHeapByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
            this(alloc, new byte[initialCapacity], 0, 0, maxCapacity);
        }
        ...
    
        private UnpooledHeapByteBuf(
                ByteBufAllocator alloc, byte[] initialArray, int readerIndex, int writerIndex, int maxCapacity) {
    
            super(maxCapacity);
    
            if (alloc == null) {
                throw new NullPointerException("alloc");
            }
            if (initialArray == null) {
                throw new NullPointerException("initialArray");
            }
            if (initialArray.length > maxCapacity) {
                throw new IllegalArgumentException(String.format(
                        "initialCapacity(%d) > maxCapacity(%d)", initialArray.length, maxCapacity));
            }
    
            this.alloc = alloc;
            setArray(initialArray);
            setIndex(readerIndex, writerIndex);
        }
    
        private void setArray(byte[] initialArray) {
            array = initialArray;
            tmpNioBuf = null;
        }
    }

既然UnpooledUnsafeHeapByteBuf与UnpooledHeapByteBuf的实例化逻辑完全一样，那它们的区别在哪？

我们知道在父类AbstractByteBuf中有很多下划线开头的抽象方法，这些方法都是需要子类去实现的，两者的区别就是在\_getByte方法的实现上。

UnpooledHeapByteBuf的\_getByte方法是直接拿到初始化new的byte数组，通过index获取数据。

@Override
        protected byte _getByte(int index) {
            return HeapByteBufUtil.getByte(array, index);
        }
        ...
        static byte getByte(byte[] memory, int index) {
            return memory[index];
        }

而UnpooledUnsafeHeapByteBuf的\_getByte方法则是调用了unsafe对象的native API，根据数组对象与偏移量来获取数据，效率相对而言较高，这也是Netty优先选择unsafe方式的原因。

@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 ###

在newDirectBuffer方法中，同样会根据hasUnsafe的值来ByteBuf的类型是否是Unsafe。

首先是UnpooledDirectByteBuf，它构造函数的逻辑很简单，通过堆外内存创建ByteBuffer。

protected UnpooledDirectByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
            super(maxCapacity);
            if (alloc == null) {
                throw new NullPointerException("alloc");
            }
            if (initialCapacity < 0) {
                throw new IllegalArgumentException("initialCapacity: " + initialCapacity);
            }
            if (maxCapacity < 0) {
                throw new IllegalArgumentException("maxCapacity: " + maxCapacity);
            }
            if (initialCapacity > maxCapacity) {
                throw new IllegalArgumentException(String.format(
                        "initialCapacity(%d) > maxCapacity(%d)", initialCapacity, maxCapacity));
            }
    
            this.alloc = alloc;
            // ByteBuffer.allocateDirect(initialCapacity)通过堆外内存创建ByteBuffer
            setByteBuffer(ByteBuffer.allocateDirect(initialCapacity));
        }
    
        private void setByteBuffer(ByteBuffer buffer) {
            ByteBuffer oldBuffer = this.buffer;
            if (oldBuffer != null) {
                if (doNotFree) {
                    doNotFree = false;
                } else {
                    freeDirect(oldBuffer);
                }
            }
    
            this.buffer = buffer;
            tmpNioBuf = null;
            capacity = buffer.remaining();
        }

再看一下UnpooledUnsafeDirectByteBuf构造函数的逻辑。

protected UnpooledUnsafeDirectByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
            super(maxCapacity);
            if (alloc == null) {
                throw new NullPointerException("alloc");
            }
            if (initialCapacity < 0) {
                throw new IllegalArgumentException("initialCapacity: " + initialCapacity);
            }
            if (maxCapacity < 0) {
                throw new IllegalArgumentException("maxCapacity: " + maxCapacity);
            }
            if (initialCapacity > maxCapacity) {
                throw new IllegalArgumentException(String.format(
                        "initialCapacity(%d) > maxCapacity(%d)", initialCapacity, maxCapacity));
            }
    
            this.alloc = alloc;
            setByteBuffer(allocateDirect(initialCapacity), false);
        }
    
        final void setByteBuffer(ByteBuffer buffer, boolean tryFree) {
            if (tryFree) {
                ByteBuffer oldBuffer = this.buffer;
                if (oldBuffer != null) {
                    if (doNotFree) {
                        doNotFree = false;
                    } else {
                        freeDirect(oldBuffer);
                    }
                }
            }
            this.buffer = buffer;
            // 计算ByteBuffer的内存地址
            memoryAddress = PlatformDependent.directBufferAddress(buffer);
            tmpNioBuf = null;
            capacity = buffer.remaining();
        }

可以看到，方法整体逻辑与非unsafe方式是一样的，只不过在setByteBuffer方法中额外计算了ByteBuffer的内存地址。

再来看一下两个类对于\_getByte方法的实现有什么不同：

UnpooledDirectByteBuf直接通过index在ByteBuffer中获取数据。

@Override
        protected byte _getByte(int index) {
            return buffer.get(index);
        }

而UnpooledUnsafeDirectByteBuf则是通过前面保存的内存地址memoryAddress利用unsafe获取数据。

@Override
        protected byte _getByte(int index) {
            return UnsafeByteBufUtil.getByte(addr(index));
        }
    
        long addr(int index) {
            return memoryAddress + index;
        }
        ...
        static byte getByte(long address) {
            return UNSAFE.getByte(address);
        }

--------------------

## PooledByteBufAllocator ##

PooledByteBufAllocator是池化分配器，首先看一下它对于newHeapBuffer方法与newDirectBuffer方法的实现逻辑。

private final PoolThreadLocalCache threadCache;
        ...
        @Override
        protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
            // 线程局部缓存，在多线程情况下，每个线程有一份独立的缓存
            PoolThreadCache cache = threadCache.get();
            PoolArena<byte[]> heapArena = cache.heapArena;
    
            ByteBuf buf;
            if (heapArena != null) {
                buf = heapArena.allocate(cache, initialCapacity, maxCapacity);
            } else {
                buf = 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;
    
            ByteBuf buf;
            if (directArena != null) {
                buf = directArena.allocate(cache, initialCapacity, maxCapacity);
            } else {
                if (PlatformDependent.hasUnsafe()) {
                    buf = UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);
                } else {
                    buf = new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
                }
            }
    
            return toLeakAwareBuffer(buf);
        }

可以看到，与unpooled模式对比，pooled模式申请内存的方式不是直接去系统申请，而是先从threadCache对象获取一个线程局部缓存对象，再获取其中的PoolArena对象，由PoolArena对象来为ByteBuf分配内存。

--------------------

### PoolThreadLocalCache ###

threadCache的对象类型是PoolThreadLocalCache，它可以看做是一个增强版的ThreadLocal，这样每个线程都会有一个自己的cache，而这个cache又维护了两类内存，分别是heap（堆内）和direct（堆外）内存。

--------------------

### PoolArena ###

PoolArena可以理解为内存分配的竞技场。在PooledByteBufAllocator初始化时，会分别创建heapArenas与directArenas两个arena数组，数组大小默认是CPU核数 \* 2，与NioEvenLoop的默认数量是一致的，目的是保证每个线程都有一个自己的arena，避免竞争。当有新的线程请求时，PoolThreadCache会拿到对应的PoolArena对象，保存到ThreadLocal成员变量中，这样就完成了线程与arena的绑定。

--------------------

### arena分配内存流程 ###

1.  从Recycler（对象回收池）获取PooledByteBuf进行复用。
2.  优先从之前分配过的缓存中分配内存。
3.  如果没有命中缓存则从内存堆中分配内存。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEyMTIzOTQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2020011511023426.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTEyMTIzOTQ=,size_16,color_FFFFFF,t_70