Netty内存池设计之内存申请代码实现

引言

前文我们通过引导，设问的方式一步步思考和总结出了一套管理连续内存的申请与分配的方法。同时也分析了相关算法思想下，Netty中算法的变种表现形式。今天，我们接着上文，来为大家梳理下Netty中针对这段算法的代码实现方式。

基础属性

用于实现内存分配算法的类是io.netty.buffer.PoolChunk。这是一个被final修饰的具备泛型的类。使用泛型的主要原因在于，用于分配的内存有两种形式：堆内内存，也就是byte[]和直接内存DirectByteBuffer。

这里有一个基础概念先明确下，进行内存空间的分配的基本是内存块，一个内存块在Netty中默认是64M大小。内存池则是有多个内存块构成。PoolChunk代表的概念是内存块。接着我们来看下PoolChunk的重要属性，如下

final class PoolChunk < T > implements PoolChunkMetric
{
    final T memory;//连续内存空间的底层载体，可能是byte[]，也可能是DirectByteBuffer
    final boolean unpooled; //该内存块是否属于内存池的一部分
    final int offset; //当memory是DirectByteBuffer时，offer代表着其内存空间地址的起始位置
    private final byte[] memoryMap;//存储管理内存区域的二叉树的节点的值；完全二叉树可以使用数组表现。
    private final byte[] depthMap;//存储管理内存区域的二叉树的节点的初始值
    private final int pageSize;//内存块是由连续的内存页构成，pageSize是内存页的大小
    private final int pageShifts;//pageSize = 1 << pageShifts
    private final int maxOrder; //二叉树的最大深度
    private final int chunkSize;//内存块的大小
    private final int log2ChunkSize;//log2ChunkSize=log2(chunkSize)
    private final byte unusable;//unusable=maxOrder+1,当一个节点的值等于unusable时，意味着该节点不可分配了
    private int freeBytes;//当前还剩余可分配大小
}

上述的属性中，我们提到了一个之前未曾提到的概念，内存页。实际上，这是一个虚拟的概念。一个内存块由连续的2^maxOrder个内存页构成。也就是说给定pageSize和maxOrder，chunkSize就被确定了。

二叉树管理的内存块可以形象的表达为

页的大小是pageSize，页的个数是2^maxOrder,而chunkSize等于pageSize * 2^maxOrder。为了方便于计算，pageSize必须是2的次方幂，默认情况下，取值为8k，maxOrder默认取值为11，也就是默认情况下，一个Chunk中有2048个内存页。

树的节点的深度取值从根节点到叶子节点依次为0到maxOrder。显然，根节点管理的内存大小是chunkSize，叶子节点管理的内存大小是chunkSize/2^maxOrder。据此，可以得到节点管理内存大小的计算公式为chunkSize/2^h,这里h是节点的深度。位移计算比除法运算更快，且chunkSize也是2的次方幂，因此可以将这里的除法计算转化为位移运算即为chunkSize >> h。而chunkSize可以看成是1 << log2ChunkSize,因此节点管理的内存大小可以计算为1 << (log2ChunkSize-h)，这个公式也就是Netty中计算节点管理内存大小的公式。

到这里，我们已经梳理完pageSize，chunkSize，log2ChunkSize,maxOrder几个属性之间的数学关系了。Netty在这内存申请和释放的相关方法中，大量的使用了位运算，因此这几个属性之间的数学关系和运算转换是一个必须要理解的基础前提。

构造方法

看过了属性的解释，接着我们来看下PoolChunk的构造方法，如下

PoolChunk(PoolArena < T > arena, T memory, int pageSize, int maxOrder, int pageShifts, int chunkSize, int offset)
{
     unpooled = false;
     this.arena = arena;
     this.memory = memory;
     this.pageSize = pageSize;
     this.pageShifts = pageShifts;
     this.maxOrder = maxOrder;
     this.chunkSize = chunkSize;
     this.offset = offset;
     unusable = (byte)(maxOrder + 1);
     log2ChunkSize = log2(chunkSize);
     subpageOverflowMask = ~(pageSize - 1);//与小内存管理相关，先忽略
     freeBytes = chunkSize;
     maxSubpageAllocs = 1 << maxOrder;//与小内存管理相关，先忽略
     p.
     memoryMap = new byte[maxSubpageAllocs << 1];
     depthMap = new byte[memoryMap.length];
     int memoryMapIndex = 1;
     for(int d = 0; d <= maxOrder; ++d)
     {
         int depth = 1 << d;
         for(int p = 0; p < depth; ++p)
         {
             memoryMap[memoryMapIndex] = (byte) d;
             depthMap[memoryMapIndex] = (byte) d;
             memoryMapIndex++;
         }
     }
     subpages = newSubpageArray(maxSubpageAllocs);//与小内存管理相关，先忽略
     cachedNioBuffers = new ArrayDeque < ByteBuffer > (8);
}

大部分属性的含义都在《基础属性》章节说明过，这边我们来说明下memoryMap和depthMap这两个数组。memoryMap用于存储二叉树每个节点的当前值，depthMap用于存储二叉树每个节点的初始值。两个数组的长度是相同的，都是1<<(maxOrder+1),因为内存页的数量，也就是二叉树叶子节点的数量是1<<maxOrder。这意味着二叉树总节点个数为（1<<maxOrder）*2-1。用数组可以表达这样的完全二叉树，其长度应该为1<<(maxOrder+1),其中下标0的元素不使用。

代码

int memoryMapIndex = 1;
for(int d = 0; d <= maxOrder; ++d)
{
int depth = 1 << d;
for(int p = 0; p < depth; ++p)
{
memoryMap[memoryMapIndex] = (byte) d;
depthMap[memoryMapIndex] = (byte) d;
memoryMapIndex++;
}
}

通过两个for循环来为二叉树每一个深度的节点来赋值，每一个节点的初始值都等于该节点所处的深度。

申请空间

在PoolChunk上申请空间的代码如下

boolean allocate(PooledByteBuf < T > buf, int reqCapacity, int normCapacity)
{
    final long handle;
    if((normCapacity & subpageOverflowMask) != 0)
    {
        handle = allocateRun(normCapacity);
    }
    else
    {
        handle = allocateSubpage(normCapacity);//当申请的大小小于pageSize时走小内存申请模式，在这里暂时忽略
    }
    if(handle < 0)
    {
        return false;
    }
    ByteBuffer nioBuffer = cachedNioBuffers != null ? cachedNioBuffers.pollLast() : null;
    initBuf(buf