日常推荐：
Netty PoolChunk原理探究
Netty-PoolSubpage原理探究
这两篇文章我觉得可以很好的帮助理解池化内存一些相关内容。

这个池化内存我们是如何使用的呢？

1.首先来到NioSocketChannle 读取数据部分：

if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {unsafe.read();}//NioByteUnsafepublic final void read() {   final ChannelPipeline pipeline = pipeline();final ByteBufAllocator allocator = config.getAllocator();final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();allocHandle.reset(config);ByteBuf byteBuf = null;boolean close = false;try {do {// 分配池化内存入口：byteBuf = allocHandle.allocate(allocator);……   } while (allocHandle.continueReading());……} catch (Throwable t) {handleReadException(pipeline, byteBuf, t, close, allocHandle);} finally {}}

我们直接看类：PooledByteBufAllocator，这个类来控制ByteBuf的分配：

@Overrideprotected 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是PooledByteBufAllocator类中已经初始化好的PoolThreadLocalCache的实例。关于PoolThreadLocalCache的一些分析我们在：Netty中自研的FastThreadLocal、FastThreadLocalThread的使用
此时我们获取到与线程相关的PoolThreadCache。

PoolSubpage的分配过程分享文章中很清楚了，我分享下分配内存之后的一些过程

PoolSubpage 是如何被使用的

final PoolSubpage<T> head = table[tableIdx];synchronized (head) {final PoolSubpage<T> s = head.next;if (s != head) {assert s.doNotDestroy && s.elemSize == normCapacity;long handle = s.allocate();assert handle >= 0;s.chunk.initBufWithSubpage(buf, null, handle, reqCapacity);incTinySmallAllocation(tiny);return;}}

我们这里分析下为什么是s != head 和 s.chunk.initBufWithSubpage(buf, null, handle, reqCapacity);的过程
这里先看smallSubpagePools中head。这里以smallSubpagePools为例

1.SubpagePools中head

for (int i = 0; i < smallSubpagePools.length; i ++) {smallSubpagePools[i] = newSubpagePoolHead(pageSize);
}private PoolSubpage<T> newSubpagePoolHead(int pageSize) {PoolSubpage<T> head = new PoolSubpage<T>(pageSize);head.prev = head;head.next = head;return head;
}

这里可以看到head.next是它自己。同时我看到了PoolSubpage在使用中是一种环形链表
所以当smallSubpagePools[i]中只有head时，是不满足{ s!=head }的。此时会进入：

 synchronized (this) {allocateNormal(buf, reqCapacity, normCapacity);}private void allocateNormal(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) {if (q050.allocate(buf, reqCapacity, normCapacity) || q025.allocate(buf, reqCapacity, normCapacity) ||q000.allocate(buf, reqCapacity, normCapacity) || qInit.allocate(buf, reqCapacity, normCapacity) ||q075.allocate(buf, reqCapacity, normCapacity)) {return;}// Add a new chunk.PoolChunk<T> c = newChunk(pageSize, maxOrder, pageShifts, chunkSize);boolean success = c.allocate(buf, reqCapacity, normCapacity);assert success;qInit.add(c);}

可以看到当q050等chunk链表中如果有合适的chunk时，会直接使用。否则创建一个新的chunk。

boolean allocate(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) {final long handle;if ((normCapacity & subpageOverflowMask) != 0) { // >= pageSizehandle =  allocateRun(normCapacity);} else {handle = allocateSubpage(normCapacity);}if (handle < 0) {return false;}ByteBuffer nioBuffer = cachedNioBuffers != null ? cachedNioBuffers.pollLast() : null;initBuf(buf, nioBuffer, handle, reqCapacity);return true;}
private long allocateSubpage(int normCapacity) {// Obtain the head of the PoolSubPage pool that is owned by the PoolArena and synchronize on it.// This is need as we may add it back and so alter the linked-list structure.PoolSubpage<T> head = arena.findSubpagePoolHead(normCapacity);int d = maxOrder; // subpages are only be allocated from pages i.e., leavessynchronized (head) {int id = allocateNode(d);if (id < 0) {return id;}final PoolSubpage<T>[] subpages = this.subpages;final int pageSize = this.pageSize;freeBytes -= pageSize;int subpageIdx = subpageIdx(id);PoolSubpage<T> subpage = subpages[subpageIdx];if (subpage == null) {subpage = new PoolSubpage<T>(head, this, id, runOffset(id), pageSize, normCapacity);subpages[subpageIdx] = subpage;} else {subpage.init(head, normCapacity); //subpages重新使用？}return subpage.allocate();}}

在进行allocateSubpage过程时，我们有看到smallSubpagePools中的head节点。同时又看到了PoolSubpage构造函数的使用。和subpage重新init的过程。head与PoolChunk中subpages数组中subpage的关联就可以在 subpage.init()中体现：

private void addToPool(PoolSubpage<T> head) {assert prev == null && next == null;prev = head;next = head.next;next.prev = this;head.next = this;}

把新分配的subpage加入到了head的下一个节点。
最后subpage.allocate()会返回一个handle值。接着回道PoolChunk中的allocate()方法中的initBuf:

void initBuf(PooledByteBuf<T> buf, ByteBuffer nioBuffer, long handle, int reqCapacity) {int memoryMapIdx = memoryMapIdx(handle);int bitmapIdx = bitmapIdx(handle); // bitmapIdx包含bitmap[]数组对应的下标和对应的long的二进制位数 if (bitmapIdx == 0) {byte val = value(memoryMapIdx);assert val == unusable : String.valueOf(val);buf.init(this, nioBuffer, handle, runOffset(memoryMapIdx) + offset,reqCapacity, runLength(memoryMapIdx), arena.parent.threadCache());} else {initBufWithSubpage(buf, nioBuffer, handle, bitmapIdx, reqCapacity);}}

当bitmapIdx！=0时我们才会进入initBuf WithSubpage()方法。这里是印证为什么subpage.tohandle时需要：0x4000000000000000L，二进制位：1……0（62个0）
而initBufWithSubpage则是对PooledByteBuf进行下内存分配设置：对应的chunk和offset等。

此时新的subPage已经与head进行关联。同时我们在这个过程也看到了chunk中subpage分配内存的过程。
当再有subPage需要分配内存时，则可以通过上文所提到的{ s!=head }的判断：

final PoolSubpage<T> head = table[tableIdx];synchronized (head) {final PoolSubpage<T> s = head.next;if (s != head) {assert s.doNotDestroy && s.elemSize == normCapacity;long handle = s.allocate();assert handle >= 0;s.chunk.initBufWithSubpage(buf, null, handle, reqCapacity);incTinySmallAllocation(tiny);return;}}

这个分配过程，我们又看到了s.allocate()和initBufWithSubpage。但是此时我有一个疑问，subpage内存完全分配后此subPage是如何处理的，我们再看下s.allocate()：

2.PoolSubPage内存用尽后的操作

long allocate() {if (elemSize == 0) {return toHandle(0);}if (numAvail == 0 || !doNotDestroy) {return -1;}final int bitmapIdx = getNextAvail();int q = bitmapIdx >>> 6;//第几位long ，相当于findNextAvail()方法的逆过程。int r = bitmapIdx & 63; //放到这个long第几位，所以用的是右移符号assert (bitmap[q] >>> r & 1) == 0;bitmap[q] |= 1L << r; //这个lang的相关位置置为1，表示已使用 if (-- numAvail == 0) { // subPage内存被分配完毕后，从subPage链表中移除。removeFromPool();}return toHandle(bitmapIdx);}

我们看到numAvail为0后会执行removeFromPool()：

private void removeFromPool() {assert prev != null && next != null;prev.next = next;next.prev = prev;next = null;prev = null;}

此时subpage则会从head链表中移除。下一次head中在分配subPage时要么在head中寻找下一个subPage，要么从chunk中分配新的subpage。

PoolSubPage重新加入head

当PooledByteBuf 释放内存后，subpage会重新加入到head中，会调用subpage的free方法：

boolean free(PoolSubpage<T> head, int bitmapIdx) {if (elemSize == 0) {return true;}int q = bitmapIdx >>> 6;int r = bitmapIdx & 63;assert (bitmap[q] >>> r & 1) != 0;bitmap[q] ^= 1L << r;setNextAvail(bitmapIdx);if (numAvail ++ == 0) {addToPool(head);return true;}if (numAvail != maxNumElems) {return true;} else {// Subpage not in use (numAvail == maxNumElems)if (prev == next) {// Do not remove if this subpage is the only one left in the pool.return true;}// Remove this subpage from the pool if there are other subpages left in the pool.doNotDestroy = false;removeFromPool();return false;}}

PoolChunk内存分配时机

这个问题的疑问是，Pool Area中会预分配一份PoolChunk吗，以加速PoolChunk中内存的分配，但是从PoolChunk的构造函数调用情况来看，只有在使用时，才会调用对应的newChunk方法。