IT评测·应用市场-qidao123.com

标题: 谈一谈 Netty 的内存管理 —— 且看 Netty 如何实现 Java 版的 Jemalloc [打印本页]

作者: 愛在花開的季節 时间: 2024-10-25 10:17
标题: 谈一谈 Netty 的内存管理 —— 且看 Netty 如何实现 Java 版的 Jemalloc

本文基于 Netty 4.1.112.Final 版本举行讨论

在之前的Netty 系列中，笔者是以 4.1.56.Final 版本为基础和各人讨论的，那么从本文开始，笔者将用最新版本 4.1.112.Final 对 Netty 的相关筹划展开解析，之所以这么做的原因是 Netty 的内存池筹划一直在不断地演进优化。
在 4.1.52.Final 之前 Netty 内存池是基于 jemalloc3 的筹划头脑实现的，由于在该版本的实现中，内存规格的粒度筹划的比较粗，可能会引起比较严峻的内存碎片问题。所以为了近一步降低内存碎片，Netty 在 4.1.52.Final 版本中重新基于 jemalloc4 的筹划头脑对内存池举行了重构，通过将内存规格近一步拆分成更细的粒度，以及重新筹划了内存分配算法只管将内存碎片控制在比较小的范围内。
随后在 4.1.75.Final 版本中，Netty 为了近一步降低不必要的内存消耗，将 ChunkSize 从原来的 16M 改为了 4M 。而且在默认情况下不在为普通的用户线程提供内存池的 Thread Local 缓存。在兼顾性能的前提下，将不必要的内存消耗只管控制在比较小的范围内。
Netty 在后续的版本迭代中，针对内存池这块的筹划，仍然会不断地伴随着一些小范围的优化，由于这些优化点太过微小，琐碎，笔者就不在一一列出，所以干脆直接以最新版本 4.1.112.Final 来对内存池的筹划与实现展开剖析。

1. 一步一图推演 Netty 内存池总体架构筹划

Netty 内存池的团体筹划相对来说还是有那么一点点的复杂，此中涉及到了众多概念模型，每种模型在架构层面上承担着不同的职责，模型与模型之间又有着千丝万缕的联系，在面对一个复杂的体系筹划时，我们还是按照老套路，从最简单的筹划开始，一步一步的演进，直到还原出内存池的完备样貌。
因此在本小节中，笔者的着墨重点是在总体架构筹划层面上，先把内存池涉及到的这些众多概念模型为各人梳理清晰，但并不会涉及太复杂的源码实现细节，让各人有一个团体完备的熟悉。有了这个基础，在本文后续的小节中，我们再来详细讨论源码的实现细节。
首先第一个登场的模型是 PoolArena ，它是内存池中最为重要的一个概念，整个内存管理的焦点实现就是在这里完成的。

PoolArena 有两个实现，一个是 HeapArena，负责池化堆内内存，另一个是 DirectArena，负责池化堆外内存。和上篇文章一样，本文我们的重点还是在 Direct Memory 的池化管理上，后续相关的源码实现，笔者都是以 DirectArena 举行展开。
我们可以直接把 PoolArena 当做一个内存池来对待，当线程在申请 PooledByteBuf 的时间都会到 PoolArena 中去拿。如许一来就引入一个问题，就是体系中有那么多的线程，而内存的申请又是非常频仍且重要的操作，这就导致这么多的线程频仍的去争抢这一个 PoolArena，相关锁的竞争水平会非常激烈，极大的影响了内存分配的速率。
因此 Netty 筹划了多个 PoolArena 来分摊线程的竞争，将线程与 PoolArena 举行绑定来降低锁的竞争，提高内存分配的并行度。

PoolArena 的默认个数为 availableProcessors * 2 , 因为 Netty 中的 Reactor 线程个数默认恰好也是 CPU 核数的两倍，而内存的分配与释放在 Reactor 线程中是一个非常高频的操作，所以这里将 Reactor 线程与 PoolArena 一对一绑定起来，避免 Reactor 线程之间的相互竞争。
除此之外，我们还可以通过 -Dio.netty.allocator.numHeapArenas 以及 -Dio.netty.allocator.numDirectArenas 来调团体系中 HeapArena 和 DirectArena 的个数。

public class PooledByteBufAllocator {
// 默认 HeapArena 的个数
private static final int DEFAULT_NUM_HEAP_ARENA;
// 默认 DirectArena 的个数
private static final int DEFAULT_NUM_DIRECT_ARENA;
static {
// PoolArena 的默认个数为 availableProcessors * 2
final int defaultMinNumArena = NettyRuntime.availableProcessors() * 2;
DEFAULT_NUM_HEAP_ARENA = Math.max(0,
SystemPropertyUtil.getInt(
"io.netty.allocator.numHeapArenas",
(int) Math.min(
defaultMinNumArena,
runtime.maxMemory() / defaultChunkSize / 2 / 3)));
DEFAULT_NUM_DIRECT_ARENA = Math.max(0,
SystemPropertyUtil.getInt(
"io.netty.allocator.numDirectArenas",
(int) Math.min(
defaultMinNumArena,
PlatformDependent.maxDirectMemory() / defaultChunkSize / 2 / 3)));
}
}

复制代码

但毕竟上，体系中的线程不光只有 Reactor 线程这一种，还有 FastThreadLocalThread 类型的线程，以及普通 Thread 类型的用户线程，位于 Reactor 线程之外的 FastThreadLocalThread ， UserThread 在运行起来之后会离开 Reactor 线程自己单独向 PoolArena 来申请内存。
所以无论是什么类型的线程，在它运行起来之后，当第一次向内存池申请内存的时间，都会接纳 Round-Robin 的方式与一个固定的 PoolArena 举行绑定，后续在线程整个生命周期中的内存申请以及释放等操作都只会与这个绑定的 PoolArena 举行交互。
所以线程与 PoolArena 的关系是多对一的关系，也就是说一个线程只能绑定到一个固定的 PoolArena 上，而一个 PoolArena 却可以被多个线程绑定。

如许一来固然线程与 PoolArena 产生了绑定，在很大水平上降低了竟争同一 PoolArena 的激烈水平，但仍然会存在竞争的情况。那这种微小的竞争会带来什么影响呢？
针对内存池的场景，比如现在有两个线程：Thread1 和 Thread2 ，它俩共同绑定到了同一个 PoolArena 上，Thread1 首先向 PoolArena 申请了一个内存块，并加载到运行它的 CPU1  L1 Cache 中，Thread1 使用完之后将这个内存块释放回 PoolArena。
假设此时 Thread2 向 PoolArena 申请同样尺寸的内存块，而且恰好申请到了刚刚被 Thread1 释放的内存块。注意，此时这个内存块已经在 CPU1  L1 Cache 中缓存了，运行 Thread2 的 CPU2  L1 Cache 中并没有，这就涉及到了 cacheline 的核间通信（MESI 协议相关），又要耗费几十个时钟周期。
为了极致的性能，我们能不能做到无锁化呢？近一步把 cacheline 核间通信的这部门开销省去。
这就必要引入内存池的第二个模型 —— PoolThreadCache ，作为线程的 Thread Local 缓存，它用于缓存线程从 PoolArena 中申请到的内存块，线程每次申请内存的时间首先会到 PoolThreadCache 中查看是否已经缓存了相应尺寸的内存块，如果有，则直接从 PoolThreadCache 获取，如果没有，再到 PoolArena 中去申请。同理，线程每次释放内存的时间，也是先释放到 PoolThreadCache 中，而不会直接释放回 PoolArena 。
如许一来，我们通过为每个线程引入 Thread Local 本地缓存 —— PoolThreadCache，实现了内存申请与释放的无锁化，同时也避免了 cacheline 在多核之间的通信开销，极大地提升了内存池的性能。
但是如许又会引来一个问题，就是内存消耗太大了，体系中有那么多的线程，如果每个线程在向 PoolArena 申请内存的时间，我们都为它默认创建一个 PoolThreadCache 本地缓存的话，这一部门的内存消耗将会特别大。
因此为了近一步降低内存消耗又同时兼顾内存池的性能，在 Netty 的权衡之下，默认只会为 Reactor 线程以及  FastThreadLocalThread 类型的线程创建 PoolThreadCache，而普通的用户线程在默认情况下将不再拥有本地缓存。

同时 Netty 也为此提供了一个配置选项 -Dio.netty.allocator.useCacheForAllThreads, 默认为 false 。如果我们将其配置为 true , 那么 Netty 默认将会为体系中的所有线程分配 PoolThreadCache 。

DEFAULT_USE_CACHE_FOR_ALL_THREADS = SystemPropertyUtil.getBoolean(
"io.netty.allocator.useCacheForAllThreads", false);

复制代码

好了，现在我们已经清楚了内存池的线程模型，那么接下来各人肯定很好奇这个 PoolArena 内里到底长什么样子。 PoolArena 是内存池的焦点实现，它内里管理了各种不同规格的内存块，PoolArena 的整个数据结构筹划都是围绕着这些内存块的管理展开的。所以在拆解 PoolArena 之前，我们必要知道 Netty 内存池毕竟划分了哪些规格的内存块
于是就引入了内存池的第三个模型 —— SizeClasses ，Netty 的内存池也是按照内存页 page 举行内存管理的，不外与 OS 不同的是，在 Netty 中一个 page 的大小默认为 8k，我们可以通过 -Dio.netty.allocator.pageSize 调整 page 大小，但最低只能调整到 4k，而且 pageSize 必须是 2 的次幂。

// 8k
int defaultPageSize = SystemPropertyUtil.getInt("io.netty.allocator.pageSize", 8192);
// 4K
private static final int MIN_PAGE_SIZE = 4096;

复制代码

Netty 内存池最小的管理单位是 page , 而内存池单次向 OS 申请内存的单位是 Chunk，一个 Chunk 的大小默认为 4M。Netty 用一个 PoolChunk 的结构来管理这 4M 的内存空间。我们可以通过 -Dio.netty.allocator.maxOrder 来调整 chunkSize 的大小（默认为 4M），maxOrder 的默认值为 9 ，最大值为 14。

// 9
int defaultMaxOrder = SystemPropertyUtil.getInt("io.netty.allocator.maxOrder", 9);
// 8196 << 9 = 4M
final int defaultChunkSize = DEFAULT_PAGE_SIZE << DEFAULT_MAX_ORDER;
// 1G
private static final int MAX_CHUNK_SIZE = (int) (((long) Integer.MAX_VALUE + 1) / 2);

复制代码

这里有四点焦点筹划原则必要考虑：

Netty 必要只管控制内存的消耗，尽可能用少量的 PoolChunk 满意大量的内存分配需求，避免创建新的 PoolChunk，提高每个 PoolChunk 的内存使用率。
而对于现有的 PoolChunk 来说，Netty 则必要只管避免将其回收，让它的服务周期尽可能长一些。
在此基础之上，Netty 必要兼顾内存分配的性能。
Netty 必要在内存池的整个生命周期中，从总体上做到让 PoolArena 中的这些 PoolChunk 只管平衡地承担内存分配的工作，做到雨露均沾。

那么 Netty 接纳如许的分配顺序 —— q050 > q025 > q000 > qInit > q075 ，如何保证上述四点焦点筹划原则呢？
首先前面我们已经分析过了，在内存频仍使用的场景中，内存池 PoolArena 中的 PoolChunks 大概率会集中停留在 q050 和 q075 这两个 PoolChunkList 中。由于 q050 和 q075 中集中了大量的 PoolChunks，所以我们肯定会先从这两个 PoolChunkList 查找，一下子就能找到一个 PoolChunk，保证了第三点原则 —— 内存分配的性能。
而 q075 中的 PoolChunk 内存使用率已经很高了，在 75% 到 100% 之间，很可能容量不能满意内存分配的需求导致申请内存失败，所以我们优先从 q050 开始。
由于 q050 [50% , 100%) 中同样集中了大量的 PoolChunks，优先从 q050 开始分配可以做到尽可能的使用现有的 PoolChunk，避免了这些 PoolChunk 由于长期不被使用而被释放回 OS , 保证了第二点筹划原则。
当 q050 中没有 PoolChunk 时，同样是根据第二点筹划原则，Netty 必要只管优先选择内存使用率高的 PoolChunk，所以优先从 q025 [25% , 75%) 举行分配。q025 中没有则优先从 q000 [1% , 50%) 中分配，只管避免 PoolChunk 的回收。
当 q000 中没有 PoolChunk 时，那阐明此时内存池中的内存容量已经不太够了，但是根据第一点筹划原则，在这种情况下，仍然必要避免创建新的 PoolChunk，所以下一个优先选择的 PoolChunkList 应该是 qInit [0% , 25%) ，而前面我们也介绍过了，Netty 筹划 qInit 的目的就是为了避免频仍创建不必要的 PoolChunk。
当 qInit 没有 PoolChunk 时，仍然不会贸然创建新的 PoolChunk，而是到 q075 中去寻找 PoolChunk 。之所以最后才轮到 q075，这是为了保证第四点筹划原则，因为 q075 中的内存使用率已经很高了，为了总体上保证 PoolChunk 平衡地承担内存分配的工作，所有优先到其他内存使用率相对较低的 PoolChunkList 中分配。
以上是笔者要为各人介绍的 Netty 针对 PoolChunkList 的第二个筹划，下面我们接着来看第三个筹划。各人可能注意到，PoolArena 中的这六个 PoolChunkList 在内存使用率区间的筹划上有许多重叠的部门，比如内存使用率是 30% 的 PoolChunk 既可以在 q000 中也可以在 q025 中，55% 既可以在 q025 中也可以在 q050 中，Netty 为什么要将 PoolChunkList 的内存使用率区间筹划成这么多的重叠区间？为什么不筹划成恰好连续衔接的区间呢 ？

我们可以反过来思考一下，如果 Netty 将 PoolChunkList 的内存使用率区间筹划成恰好连续衔接的区间，那么会发生什么情况？
我们现在拿 q025 和 q050 这两个 PoolChunkList 举例阐明，假设现在我们将 q025 的内存使用率区间筹划成 [25% , 50%) , q050 的内存使用率区间筹划成 [50% , 75%)，如许一来，q025 ， q050 ， q075 这三个 PoolChunkList 的内存使用率区间的上限和下限就是恰好连续衔接的了。
那么随着 PoolChunk 中内存的申请与释放，会导致 PoolChunk 的内存使用率在不断的发生变革，假设现在有一个 PoolChunk 的内存使用率是 45% ，当前停留在 q025 中，当分配内存之后，内存使用率上升至 50% ，那么该 PoolChunk 就必要立即移动到 q050 中。
当释放内存之后，这个刚刚移动到 q050 中的 PoolChunk，它的内存使用率下降到 49% ，那么又会快马加鞭地移动到 q025 ，也就是说只要这个 PoolChunk 的内存使用率在 q025 与 q050 的交界处 50% 附近来回徘徊的话，每次的内存申请与释放都会导致这个 PoolChunk 在 q025 与 q050 之间不停地来回移动。
同样的道理，只要一个 PoolChunk 的内存使用率在 75% 左右来回徘徊的话，那么每次内存的申请与释放也都会导致这个 PoolChunk 在 q050 与 q075 之间不停地来回移动，如许会造成肯定的性能下降。
但是如果各个 PoolChunkList 之间的内存使用率区间筹划成重叠区间的话，那么 PoolChunk 的可调治范围就会很广，不会频仍地在前后不同的 PoolChunkList 之间来回移动。
我们还是拿 q025 [25% , 75%) 和 q050 [50% , 100%) 来举例阐明，现在 q025 中有一个内存使用率为 45% 的 PoolChunk ，当分配内存之后，内存使用率上升至 50% ，该 PoolChunk 仍然会继续停留在 q025 中，后续随着内存分配的不断举行，当内存使用率达到 75% 的时间才会移动到 q050 中。
还是这个 PoolChunk ，当释放内存之后，PoolChunk 的使用率下降到了 70%，那么它仍然会停留在 q050 中，后续随着内存释放的不断举行，当内存使用率低于 50% 的时间才会移动到 q025 中。这种重叠区间的筹划有效的避免了 PoolChunk 频仍的在两个 PoolChunkList 之间来回移动。
好了，到现在为止，我们已经明确了内存池所有的焦点组件筹划，基于本小节中介绍的 6 个模型：PoolArena，PoolThreadCache，SizeClasses，PoolChunk ，PoolSubpage，PoolChunkList 。我们可以得出内存池的完备架构如下图所示：

2. Netty 内存池的创建与初始化

在清楚了内存池的总体架构筹划之后，本小节我们就来看一下整个内存池的骨架是如何被创建出来的，Netty 将整个内存池的实现封装在 PooledByteBufAllocator 类中。

abstract class PoolArena<T> {
enum SizeClass {
Small,
Normal
}
}

复制代码

创建内存池所必要的几个焦点参数我们必要提前了解下：
<ul>preferDirect 默认为 true , 用于指定该 Allocator 是否偏向于分配 Direct Memory，其值由 PlatformDependent.directBufferPreferred() 方法决定，相关的判断逻辑可以回看下《聊一聊 Netty 数据搬运工 ByteBuf 体系的筹划与实现》一文中的第三小节。
nHeapArena ， nDirectArena 用于指定内存池中包含的 HeapArena ， DirectArena 个数，它们分别用于池化 Heap Memory 以及 Direct Memory 。默认个数分别为 availableProcessors * 2 ，可由参数 -Dio.netty.allocator.numHeapArenas 和 -Dio.netty.allocator.numDirectArenas 指定。
pageSize 默认为 8K ，用于指定内存池中的 Page 大小。可由参数 -Dio.netty.allocator.pageSize 指定，但不能低于 4K 。

maxOrder 默认为 9 ，用于指定内存池中 PoolChunk 尺寸，默认 4M ，由 pageSize >> r & 1) == 0; // 设置内存块在 bitmap 中对应的 bit 位为 1 （已分配） bitmap[q] |= 1L > pageShifts; // 低 32 位生存 bitmapIdx return (long) runOffset RUN_OFFSET_SHIFT); }[/code]有了这个 runOffset ，我们就可以从 subpages[runOffset] 中将内存块对应的 PoolSubpage 获取到，剩下的事情就很简单了，直接将这个内存块释放回 PoolSubpage 就可以了。

struct zone {
// 伙伴系统的核心数据结构
struct free_area free_area[MAX_ORDER];
}

复制代码

随着内存块的释放，有可能会导致 PoolSubpage 变为一个 Empty PoolSubpage，也就是说 PoolSubpage 中的内存块全部空闲。对于一个 Empty PoolSubpage , Netty 会将其从 smallSubpagePools 中移除，并将 PoolSubpage 背后的内存释放回 PoolChunk。

final class PoolChunk { void free(long handle, int normCapacity, ByteBuffer nioBuffer) { // Small 规格内存块的释放 if (isSubpage(handle)) { // 获取内存块所在 PoolSubpage 的 runOffset int sIdx = runOffset(handle); PoolSubpage subpage = subpages[sIdx]; // 获取 PoolSubpage 所在 smallSubpagePools 对应规格链表头结点 PoolSubpage head = subpage.chunk.arena.smallSubpagePools[subpage.headIndex]; head.lock(); try { assert subpage.doNotDestroy; // 将内存块释放回 PoolSubpage 中 // true 表示 PoolSubpage 还是一个 Partial PoolSubpage（部门空闲） , 继续留在 smallSubpagePools 中 // false 表示 PoolSubpage 变成了一个 Empty PoolSubpage（全部空闲），从 smallSubpagePools 链表中移除 if (struct zone {
// 伙伴系统的核心数据结构
struct free_area free_area[MAX_ORDER];
}) { // the subpage is still used, do not free it return; } // Empty PoolSubpage 从 PoolChunk subpages 数组中移除 subpages[sIdx] = null; } finally { head.unlock(); } } ........ 释放 Normal 规格内存块或者 Empty PoolSubpage ...... } finally { runsAvailLock.unlock(); } }}

复制代码

内存块释放回 PoolSubpage 的逻辑也是非常简单，只必要将其 bitmapIdx 在 bitmap 中对应的 bit 位重新设置为 0 就可以了，正好和内存块的申请互为相反的操作。
那么如何通过 bitmapIdx 定位到 bitmap 中与其对应具体的 bit 呢？我们还是以上个小节的例子举行阐明，假设现在我们将 bitmapIdx 为 67 的内存块释放回 PoolSubpage 。

首先我们必要知道的是，bitmapIdx 具体是落在哪一个 bitmap 数组元素中，我们可以通过 bitmapIdx / 64 来获取，对应到上图中，bitmapIdx（67）是落在 bitmap[1] 中。

final class PoolChunk<T> implements PoolChunkMetric {
// Netty 的伙伴系统结构
private final IntPriorityQueue[] runsAvail;
}

复制代码

接下来我们就必要知道，这个 bitmapIdx 具体是 bitmap[q] 的第几个 bit ,我们可以通过 bitmapIdx & 63 来获取，对应到上图中，bitmapIdx（67）是 bitmap[1] 的第 3 个 bit （从 0 开始计数）。

abstract class PoolArena<T> {
// 管理 Small 规格内存块的核心数据结构
final PoolSubpage<T>[] smallSubpagePools;
}

复制代码

具体的 bit 定位到了，剩下的事情就很简单了，我们只必要将 bitmapIdx 对应的 bit 重新设置为 0 就可以了。

bitmap[q] ^= 1L >> 6; // long 型整数的具体第几个 bit abstract class PoolArena<T> {
// 管理 Small 规格内存块的核心数据结构
final PoolSubpage<T>[] smallSubpagePools;
} // 将内存块在 bitmap 中对应的 bit 设置为 0 bitmap[q] ^= 1L 1) { // 返回 true 表示 PoolSubpage 是一个 Partial PoolSubpage // 必要保留在 smallSubpagePools 中 return true; } } // numAvail = maxNumElems 阐明 PoolSubpage 此时变为一个 Empty PoolSubpage（全部空闲） if (numAvail != maxNumElems) { // Partial PoolSubpage 继续停留在 smallSubpagePools return true; } else { // 对于一个 Empty PoolSubpage 来说，Netty 必要将其从 smallSubpagePools 中删除，并释放 PoolSubpage 回 PoolChunk // 如果该 PoolSubpage 是 smallSubpagePools 对应规格链表中的唯一元素，那么就让它继续停留 if (prev == next) { // 始终保证 smallSubpagePools 对应规格的 PoolSubpage 链表中至少有一个 PoolSubpage return true; } // 如果对应的 PoolSubpage 链表中还有多余的 PoolSubpage // 那么就将这个 Empty PoolSubpage 释放掉 doNotDestroy = false; // 将该 Empty PoolSubpage 从 smallSubpagePools 中删除 removeFromPool(); return false; } }}

复制代码

7. PooledByteBuf 如何封装内存块

无论是从 PoolChunk 分配出来的 Normal 规格内存块，还是从 PoolSubpage 分配出来的 Small 规格内存块，内存池都会返回一个内存块的 handle 结构。

而我们拿到这个 handle 结构是无法直接使用的，因为这个 handle 并不是真正的内存，他只是用来描述内存块在 PoolChunk 中的位置信息，而真正的内存是 4M 的 PoolChunk。所以我们必要将内存块的 handle 结构转换成可以直接使用的 PooledByteBuf。

站在 PooledByteBuf 的内部视角来看，用户并不会关心 PooledByteBuf 底层的内存来自于哪里，用户只会关心 PooledByteBuf 提供的是一段从位置 0 开始，大小为 length 的内存块。在 PooledByteBuf 这个局部视角上，它的 readerIndex , writerIndex 初始均为 0 。
但我们站在整个内存池的全局视角上来看的话，PooledByteBuf 底层的内存其实是来自于 PoolChunk，笔者之前在《聊一聊 Netty 数据搬运工 ByteBuf 体系的筹划与实现》中的第 2.7 小节中介绍过 ByteBuf 视图的概念。我们可以将 PooledByteBuf 看做是 PoolChunk 的某一段局部 slice 视图。

PooledByteBuf 的本质其实是 PoolChunk 中的某一段内存地区，对于 Normal 规格的内存块来说，这段地区的起始内存地址是 memory + runOffset cache, PooledByteBuf buf, int reqCapacity) { if (cache == null) { // no cache found so just return false here return false; } // true 表示分配成功，false 表示分配失败（缓存没有了） boolean allocated = cache.allocate(buf, reqCapacity, this); // PoolThreadCache 中的 allocations 计数加 1 if (++ allocations = freeSweepAllocationThreshold) { allocations = 0; // 清理 PoolThreadCache，将缓存的内存块释放回 PoolChunk trim(); } return allocated; }[/code]同样的道理，Normal 规格内存块的申请首先也会尝试从线程本地缓存 PoolThreadCache 中去获取，如果缓存中没有，则到 PoolChunk 中申请。

abstract class PoolArena<T> {
// 分配 Page 级别的内存块
private void allocateNormal(PooledByteBuf<T> buf, int reqCapacity, int sizeIdx, PoolThreadCache threadCache) {
assert lock.isHeldByCurrentThread();
// PoolChunkList 内存分配的优先级：q050 > q025 > q000 > qInit > q075
if (q050.allocate(buf, reqCapacity, sizeIdx, threadCache) ||
q025.allocate(buf, reqCapacity, sizeIdx, threadCache) ||
q000.allocate(buf, reqCapacity, sizeIdx, threadCache) ||
qInit.allocate(buf, reqCapacity, sizeIdx, threadCache) ||
q075.allocate(buf, reqCapacity, sizeIdx, threadCache)) {
return;
}
// 5 个 PoolChunkList 中没有可用的 PoolChunk，重新向 OS 申请一个新的 PoolChunk（4M）
PoolChunk<T> c = newChunk(sizeClass.pageSize, sizeClass.nPSizes, sizeClass.pageShifts, sizeClass.chunkSize);
// 从新的 PoolChunk 中分配内存
boolean success = c.allocate(buf, reqCapacity, sizeIdx, threadCache);
assert success;
// 将刚刚创建的 PoolChunk 加入到 qInit 中
qInit.add(c);
}
}

复制代码

PoolThreadCache 中的 normalDirectCaches 是用来缓存 Normal 规格的内存块，但默认情况下只会缓存一种 Normal 规格 —— 32K ，超过 32K 还是必要到 PoolChunk 中去申请。
normalDirectCaches 数组的 index 就是对应 Normal 规格在内存规格表中的 sizeIndex - 39 , 因为第一个 Normal 规格（32K）的 sizeIndex 就是 39 。

public class PooledByteBufAllocator {
public static final PooledByteBufAllocator DEFAULT =
new PooledByteBufAllocator(PlatformDependent.directBufferPreferred());
}

复制代码

当我们获取到 Normal 规格对应的 MemoryRegionCache 之后，剩下的流程就都是一样的了，从 MemoryRegionCache 获取一个 Entry 实例，根据内里封装的内存块信息包装成 PooledByteBuf 返回。
8.3 清理 PoolThreadCache 中不经常使用的内存块

Netty 清理 PoolThreadCache 缓存有两个机遇，一个是自动清理，当 PoolThreadCache 分配内存块的次数 allocations （包括 Small 规格，Normal 规格的分配次数）达到阈值 freeSweepAllocationThreshold （8192）时 , Netty 将会把 PoolThreadCache 中缓存的所有 Small 规格以及 Normal 规格的内存块全部释放回 PoolSubpage 中。

private final PoolThreadLocalCache threadCache;
private final int smallCacheSize;
private final int normalCacheSize;
private final int chunkSize;
// 保存所有 DirectArena
private final PoolArena<ByteBuffer>[] directArenas;
public PooledByteBufAllocator(boolean preferDirect, int nHeapArena, int nDirectArena, int pageSize, int maxOrder,
int smallCacheSize, int normalCacheSize,
boolean useCacheForAllThreads, int directMemoryCacheAlignment) {
// 默认偏向于分配 Direct Memory
super(preferDirect);
// 创建 PoolThreadLocalCache ，后续用于将线程与 PoolArena 绑定
// 并为线程创建 PoolThreadCache
threadCache = new PoolThreadLocalCache(useCacheForAllThreads);
// PoolThreadCache 中，针对每一个 Small 规格的尺寸可以缓存 256 个内存块
this.smallCacheSize = smallCacheSize;
// PoolThreadCache 中，针对每一个 Normal 规格的尺寸可以缓存 64 个内存块
this.normalCacheSize = normalCacheSize;
// PoolChunk 的尺寸
// pageSize << maxOrder = 4M
chunkSize = validateAndCalculateChunkSize(pageSize, maxOrder);
// 13 ， pageSize 为 8K
int pageShifts = validateAndCalculatePageShifts(pageSize, directMemoryCacheAlignment);
// 依次创建 nDirectArena 个 DirectArena（省略 HeapArena）
if (nDirectArena > 0) {
// 创建 PoolArena 数组，个数为 2 * processors
directArenas = newArenaArray(nDirectArena);
// 划分内存规格，建立内存规格索引表
final SizeClasses sizeClasses = new SizeClasses(pageSize, pageShifts, chunkSize,
directMemoryCacheAlignment);
// 初始化 PoolArena 数组
for (int i = 0; i < directArenas.length; i ++) {
// 创建 DirectArena
PoolArena.DirectArena arena = new PoolArena.DirectArena(this, sizeClasses);
// 保存在 directArenas 数组中
directArenas[i] = arena;
}
} else {
directArenas = null;
}
}

复制代码

挨个释放 smallSubPageDirectCaches 以及 normalDirectCaches 中的 MemoryRegionCache 。

private final class PoolThreadLocalCache extends FastThreadLocal<PoolThreadCache> {
private final boolean useCacheForAllThreads;
PoolThreadLocalCache(boolean useCacheForAllThreads) {
this.useCacheForAllThreads = useCacheForAllThreads;
}
@Override
protected synchronized PoolThreadCache initialValue() {
实现线程与 PoolArena 之间的绑定
为线程创建本地缓存 PoolThreadCache
}
}

复制代码

释放缓存在 MpscQueue 中的所有内存块。

private static <T> PoolArena<T>[] newArenaArray(int size) {
return new PoolArena[size];
}

复制代码

从 MpscQueue 中获取 Entry，根据 Entry 结构中封装的内存块信息，将其释放回内存池中。

abstract class PoolArena<T> {
// Small 规格的内存块组织在这里，类似内核的 kmalloc
final PoolSubpage<T>[] smallSubpagePools;
// 按照不同内存使用率组织 PoolChunk
private final PoolChunkList<T> q050; // [50% , 100%)
private final PoolChunkList<T> q025; // [25% , 75%)
private final PoolChunkList<T> q000; // [1% , 50%)
private final PoolChunkList<T> qInit; // [0% , 25%)
private final PoolChunkList<T> q075; // [75% , 100%)
private final PoolChunkList<T> q100; // 100%
protected PoolArena(PooledByteBufAllocator parent, SizeClasses sizeClass) {
// PoolArena 所属的 PooledByteBufAllocator
this.parent = parent;
// Netty 内存规格索引表
this.sizeClass = sizeClass;
// small 内存规格将会在这里分配 —— 类似 kmalloc
// 每一种 small 内存规格都会对应一个 PoolSubpage 链表（类似 slab）
smallSubpagePools = newSubpagePoolArray(sizeClass.nSubpages);
for (int i = 0; i < smallSubpagePools.length; i ++) {
// smallSubpagePools 数组中的每一项是一个带有头结点的 PoolSubpage 结构双向链表
// 双向链表的头结点是 SubpagePoolHead
smallSubpagePools[i] = newSubpagePoolHead(i);
}
// 按照不同内存使用率范围划分 PoolChunkList
q100 = new PoolChunkList<T>(this, null, 100, Integer.MAX_VALUE, sizeClass.chunkSize);// [100 , 2147483647]
q075 = new PoolChunkList<T>(this, q100, 75, 100, sizeClass.chunkSize);
q050 = new PoolChunkList<T>(this, q075, 50, 100, sizeClass.chunkSize);
q025 = new PoolChunkList<T>(this, q050, 25, 75, sizeClass.chunkSize);
q000 = new PoolChunkList<T>(this, q025, 1, 50, sizeClass.chunkSize);
qInit = new PoolChunkList<T>(this, q000, Integer.MIN_VALUE, 25, sizeClass.chunkSize);// [-2147483648 , 25]
// 双向链表组织 PoolChunkList
// 其中比较特殊的是 q000 的前驱节点指向 NULL
// qInit 的前驱节点指向它自己
q100.prevList(q075);
q075.prevList(q050);
q050.prevList(q025);
q025.prevList(q000);
q000.prevList(null);
qInit.prevList(qInit);
}
}

复制代码

另一种清理 PoolThreadCache 缓存的机遇是定时被动清理，定时清理机制默认是关闭的。但我们可以通过 -Dio.netty.allocator.cacheTrimIntervalMillis 参数举行开启，该参数默认为 0 ，单位为毫秒，用于指定定时清理 PoolThreadCache 的时间间隔。

private PoolSubpage<T>[] newSubpagePoolArray(int size) {
return new PoolSubpage[size];
}

复制代码

8.4 PoolThreadCache 的内存回收流程

当 PooledByteBuf 的引用计数为 0 时，Netty 就会将 PooledByteBuf 背后引用的内存块释放回内存池中，而且将 PooledByteBuf 这个实例释放回对象池。

private PoolSubpage<T> newSubpagePoolHead(int index) {
PoolSubpage<T> head = new PoolSubpage<T>(index);
head.prev = head;
head.next = head;
return head;
}

复制代码

如果内存块是 Huge 规格的，那么直接释放回 OS , 如果内存块不是 Huge 规格的，那么就根据内存块 handle 结构中的 isSubpage bit 位判断该内存块是 Small 规格的还是 Normal 规格的。

Small 规格 handle 结构 isSubpage bit 位设置为 1 ，Normal 规格 handle 结构 isSubpage bit 位设置为 0 。

short[][] sizeClasses

复制代码

然后根据内存块的规格释放回对应的 MemoryRegionCache 中。

private static final int LOG2GROUP_IDX = 1;
private static final int LOG2DELTA_IDX = 2;
private static final int NDELTA_IDX = 3;
private static final int PAGESIZE_IDX = 4;
private static final int SUBPAGE_IDX = 5;
private static final int LOG2_DELTA_LOOKUP_IDX = 6;

复制代码

这里缓存添加失败的情况有三种：

对应规格的 MemoryRegionCache 已经满了，对于 Small 规格来说，其对应的 MemoryRegionCache 缓存结构最多可以缓存 256 个内存块，对于 Normal 规格来说，则最多可以缓存 64 个。
PoolThreadCache 并没有提供对应规格尺寸的 MemoryRegionCache 缓存。比如默认情况下，Netty 只会提供 32K 这一种 Normal 规格的缓存，如果释放 40K 的内存块则只能释放回内存池中。
线程对应的本地缓存 PoolThreadCache 已经被释放。比如线程已经退出了，那么其对应的 PoolThreadCache 则会被释放，这时内存块就只能释放回内存池中。

private static int calculateSize(int log2Group, int nDelta, int log2Delta) {
return (1 << log2Group) + (nDelta << log2Delta);
}

复制代码

将内存块释放回对应规格尺寸的 MemoryRegionCache 中。

final class SizeClasses {
// 第一种内存规格的基准 size —— 16B
// 以及第一个内存规格增长间距 —— 16B
static final int LOG2_QUANTUM = 4;
// 每个内存规格组内，规格的个数 —— 4 个
private static final int LOG2_SIZE_CLASS_GROUP = 2;
// size2idxTab 中索引的最大内存规格 —— 4K
private static final int LOG2_MAX_LOOKUP_SIZE = 12;
}

复制代码

8.5 PoolThreadCache 的释放

PoolThreadCache 是线程的本地缓存，内里缓存了内存池中 Small 规格的内存块以及 Normal 规格的内存块。

// 22 - 4 -2 + 1 = 17
int group = log2(chunkSize) - LOG2_QUANTUM - LOG2_SIZE_CLASS_GROUP + 1;

复制代码

当线程终结的时间，其对应的 PoolThreadCache 也随即会被 GC 回收，但这里必要注意的是 GC 回收的只是 PoolThreadCache 这个 Java 实例，其内部缓存的这些内存块 GC 是管不着的，因为 GC 并不知道这里还有一个内存池的存在。
同样的道理雷同于 JDK 中的 DirectByteBuffer，GC 只负责回收 DirectByteBuffer 这个 Java 实例，其背后引用的 Native Memory ，GC 是管不着的，所以我们必要使用额外的机制来保证这些 Native Memory 被及时回收。
对于 JDK 中的 DirectByteBuffer，JDK 使用了 Cleaner 机制来回收背后的 Native Memory ，而对于 PoolThreadCache 来说，Netty 这里则是用了 Finalizer 机制会释放。

对 Cleaner 以及 Finalizer 背后的实现细节感爱好的读者朋友可以回看下笔者之前的文章《以 ZGC 为例，谈一谈 JVM 是如何实现 Reference 语义的》。

PoolThreadCache 中有一个 freeOnFinalize 字段：

SizeClasses(int pageSize, int pageShifts, int chunkSize, int directMemoryCacheAlignment) {
// 一共分为 17 个内存规格组
int group = log2(chunkSize) - LOG2_QUANTUM - LOG2_SIZE_CLASS_GROUP + 1;
// 创建内存规格表 sizeClasses
// 每个内存规格组内有 4 个规格，一共 68 个内存规格，一维数组长度为 68
// 二维数组的长度为 7
// 保存的内存规格信息为：index, log2Group, log2Delta, nDelta, isMultiPageSize, isSubPage, log2DeltaLookup
short[][] sizeClasses = new short[group << LOG2_SIZE_CLASS_GROUP][7];
int normalMaxSize = -1;
// 内存规格 index , 初始为 0
int nSizes = 0;
// 内存规格 size
int size = 0;
// 第一组内存规格的基准 size 为 16B
int log2Group = LOG2_QUANTUM;
// 第一组内存规格之间的间隔为 16B
int log2Delta = LOG2_QUANTUM;
// 每个内存规格组内限定为 4 个规格
int ndeltaLimit = 1 << LOG2_SIZE_CLASS_GROUP;
// 初始化第一个内存规格组 [16B , 64B]，nDelta 从 0 开始
for (int nDelta = 0; nDelta < ndeltaLimit; nDelta++, nSizes++) {
// 初始化对应内存规格的 7 个信息
short[] sizeClass = newSizeClass(nSizes, log2Group, log2Delta, nDelta, pageShifts);
// nSizes 为该内存规格的 index
sizeClasses[nSizes] = sizeClass;
// 通过 sizeClass 计算该内存规格的 size ,然后将 size 向上对齐至 directMemoryCacheAlignment 的最小整数倍
size = sizeOf(sizeClass, directMemoryCacheAlignment);
}
// 每个内存规格组内的最后一个规格，往往是下一个内存规格组的基准 size
// 比如第一个内存规格组内最后一个规格 64B ，它是第二个内存规格组的基准 size
// 4 + 2 = 6，第二个内存规格组的基准 size 为 64B
log2Group += LOG2_SIZE_CLASS_GROUP;
// 初始化剩下的 16 个内存规格组
// 后一个内存规格组的 log2Group，log2Delta 比前一个内存规格组的 log2Group ，log2Delta 多 1
for (; size < chunkSize; log2Group++, log2Delta++) {
// 每个内存规格组内的 nDelta 从 1 到 4 ，最大内存规格不能超过 chunkSize（4M）
for (int nDelta = 1; nDelta <= ndeltaLimit && size < chunkSize; nDelta++, nSizes++) {
// 初始化对应内存规格的 7 个信息
short[] sizeClass = newSizeClass(nSizes, log2Group, log2Delta, nDelta, pageShifts);
// nSizes 为该内存规格的 index
sizeClasses[nSizes] = sizeClass;
size = normalMaxSize = sizeOf(sizeClass, directMemoryCacheAlignment);
}
}
// 最大内存规格不能超过 chunkSize（4M）
// 超过 4M 就是 Huge 内存规格，直接分配不进行池化管理
assert chunkSize == normalMaxSize;
...... 省略 ......
}

复制代码

当 useFinalizer 为 true 的时间，Netty 会创建一个 FreeOnFinalize 实例：

short[][] sizeClasses = new short[group << LOG2_SIZE_CLASS_GROUP][7];

复制代码

FreeOnFinalize 对象再一次循环引用了 PoolThreadCache ， FreeOnFinalize 重写了 finalize() 方法，当 FreeOnFinalize 对象创建的时间，JVM 会为其创建一个 Finalizer 对象（FinalReference 类型），Finalizer 引用了 FreeOnFinalize ，但这种引用关系是一种 FinalReference 类型。

// 4 + 2 = 6
log2Group += LOG2_SIZE_CLASS_GROUP;

复制代码

与 PoolThreadCache 相关的对象引用关系如下图所示：

当线程终结的时间，那么 PoolThreadCache 与 FreeOnFinalize 对象将会被 GC 回收，但由于 FreeOnFinalize 被一个 FinalReference（Finalizer）引用，所以 JVM 会将 FreeOnFinalize 对象再次复活，由于 FreeOnFinalize 对象也引用了 PoolThreadCache，所以 PoolThreadCache 也会被复活。
随后 JDK 中的 2 号线程 —— finalizer 会实验 FreeOnFinalize 对象的 finalize() 方法，释放 PoolThreadCache。

private static short[] newSizeClass(int index, int log2Group, int log2Delta, int nDelta, int pageShifts) {
// 判断规格尺寸是否是 Page 的整数倍
short isMultiPageSize;
if (log2Delta >= pageShifts) {
// 尺寸按照 Page 的倍数递增了，那么一定是 Page 的整数倍
isMultiPageSize = yes;
} else {
int pageSize = 1 << pageShifts;
// size = 1 << log2Group + nDelta * (1 << log2Delta)
int size = calculateSize(log2Group, nDelta, log2Delta);
// 是否能被 pagesize(8k) 整除
isMultiPageSize = size == size / pageSize * pageSize? yes : no;
}
// 规格尺寸小于 32K ，那么就属于 Small 规格，对应的内存块会被 PoolSubpage 管理
short isSubpage = log2Size < pageShifts + LOG2_SIZE_CLASS_GROUP? yes : no;
// 如果内存规格 size 小于等于 MAX_LOOKUP_SIZE（4K），那么 log2DeltaLookup 为 log2Delta
// 如果内存规格 size 大于 MAX_LOOKUP_SIZE（4K），则为 0
// Netty 只会为 4K 以下的内存规格建立 size2idxTab 索引
int log2DeltaLookup = log2Size < LOG2_MAX_LOOKUP_SIZE ||
log2Size == LOG2_MAX_LOOKUP_SIZE && remove == no
? log2Delta : no;
// 初始化内存规格信息
return new short[] {
(short) index, (short) log2Group, (short) log2Delta,
(short) nDelta, isMultiPageSize, isSubpage, (short) log2DeltaLookup
};
}

复制代码

但如果有人不想使用 Finalizer 来释放的话，则可以通过将 -Dio.netty.allocator.disableCacheFinalizersForFastThreadLocalThreads 设置为 true , 那么 useFinalizer 就会变为 false 。
如许一来当线程终结的时间，它的本地缓存 PoolThreadCache 将不会由 Finalizer 来清理。这种情况下，我们就必要特别注意，肯定要通过 FastThreadLocal.removeAll() 或者 PoolThreadLocalCache.remove(PoolThreadCache) 来手动举行清理。否则就会造成内存泄露。

// Small 规格中最大的规格尺寸对应的 index (38)
int smallMaxSizeIdx = 0;
// size2idxTab 中最大的 lookup size （4K）
int lookupMaxSize = 0;
// Page 级别内存规格的个数（32）
int nPSizes = 0;
// Small 内存规格的个数（39）
int nSubpages = 0;
// 遍历内存规格表 sizeClasses，统计 nPSizes ， nSubpages，smallMaxSizeIdx，lookupMaxSize
for (int idx = 0; idx < nSizes; idx++) {
short[] sz = sizeClasses[idx];
// 只要 size 可以被 pagesize 整除，那么就属于 MultiPageSize
if (sz[PAGESIZE_IDX] == yes) {
nPSizes++;
}
// 只要 size 小于 32K 则为 Subpage 的规格
if (sz[SUBPAGE_IDX] == yes) {
nSubpages++;
// small 内存规格中的最大尺寸 28K ，对应的 sizeIndex = 38
smallMaxSizeIdx = idx;
}
// 内存规格小于等于 4K 的都属于 lookup size
if (sz[LOG2_DELTA_LOOKUP_IDX] != no) {
// 4K
lookupMaxSize = sizeOf(sz, directMemoryCacheAlignment);
}
}
// 38
this.smallMaxSizeIdx = smallMaxSizeIdx;
// 4086(4K)
this.lookupMaxSize = lookupMaxSize;
// 32
this.nPSizes = nPSizes;
// 39
this.nSubpages = nSubpages;
// 68
this.nSizes = nSizes;
// 8192(8K)
this.pageSize = pageSize;
// 13
this.pageShifts = pageShifts;
// 4M
this.chunkSize = chunkSize;
// 0
this.directMemoryCacheAlignment = directMemoryCacheAlignment;

复制代码

下面是 PoolThreadCache 的释放流程：

// sizeIndex 与 size 之间的映射
this.sizeIdx2sizeTab = newIdx2SizeTab(sizeClasses, nSizes, directMemoryCacheAlignment);
// 根据 sizeClass 生成 page 级的内存规格表
// pageIndex 到对应的 size 之间的映射
this.pageIdx2sizeTab = newPageIdx2sizeTab(sizeClasses, nSizes, nPSizes, directMemoryCacheAlignment);
// 4k 之内，给定一个 size 转换为 sizeIndex
this.size2idxTab = newSize2idxTab(lookupMaxSize, sizeClasses);

复制代码

9. 内存池相关的 Metrics

为了更好的监控内存池的运行状态，Netty 为内存池中的每个组件都筹划了一个对应的 Metrics 类，用于封装与该组件相关的 Metrics。

此中内存池 PooledByteBufAllocator 提供的 Metrics 如下：

private static int[] newIdx2SizeTab(short[][] sizeClasses, int nSizes, int directMemoryCacheAlignment) {
// 68 种内存规格，映射条目也是 68
int[] sizeIdx2sizeTab = new int[nSizes];
// 遍历内存规格表，建立 index 与规格 size 之间的映射
for (int i = 0; i < nSizes; i++) {
short[] sizeClass = sizeClasses[i];
// size = 1 << log2Group + nDelta * (1 << log2Delta)
sizeIdx2sizeTab[i] = sizeOf(sizeClass, directMemoryCacheAlignment);
}
return sizeIdx2sizeTab;
}

复制代码

PoolArena 提供的 Metrics 如下：

private static int[] newPageIdx2sizeTab(short[][] sizeClasses, int nSizes, int nPSizes,
int directMemoryCacheAlignment) {
// page 级的内存规格，个数为 32
int[] pageIdx2sizeTab = new int[nPSizes];
int pageIdx = 0;
// 遍历内存规格表，建立 pageIdx 与对应 Page 级内存规格 size 之间的映射
for (int i = 0; i < nSizes; i++) {
short[] sizeClass = sizeClasses[i];
if (sizeClass[PAGESIZE_IDX] == yes) {
pageIdx2sizeTab[pageIdx++] = sizeOf(sizeClass, directMemoryCacheAlignment);
}
}
return pageIdx2sizeTab;
}

复制代码

PoolSubpage 提供的 Metrics 如下：

int[] size2idxTab

复制代码

PoolChunkList 提供的 Metrics 如下：

// 基础内存规格
static final int LOG2_QUANTUM = 4;

复制代码

PoolChunk 提供的 Metrics 如下：

lookupSize - 1 >> LOG2_QUANTUM

复制代码

总结

到现在为止，关于 Netty 内存池的整个筹划与实现笔者就为各人剖析完了，从整个内存池的筹划过程来看，我们见到了许多 OS 内核的影子，Netty 也是参考了许多 OS 内存管理方面的筹划，如果对 OS 内存管理这块内容感爱好的读者朋友可以扩展看一下笔者之前写的相关文章：

好了，本日的内容就到这里，我们下篇文章见~~~

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

欢迎光临 IT评测·应用市场-qidao123.com (https://dis.qidao123.com/)