文章

6. ByteBuf原理

发表于
作者 jamespud
47 分钟阅读
6. ByteBuf原理

ByteBuf定义

Buffer的存在是为了提高IO 的效率，当Buffer达到一定长度后再2进行数据传输操作，而不是每次写操作完后立即开始数据传输。

首先看看ByteBuf注释中的示意图

1
2
3
4
5
6

+-------------------+------------------+------------------+
| discardable bytes |  readable bytes  |  writable bytes  |
|                   |     (CONTENT)    |                  |
+-------------------+------------------+------------------+
|                   |                  |                  |
0      <=      readerIndex   <=   writerIndex    <=    capacity

ByteBuf提供了两个指针来支持顺序读取和写入操作: readerIndex和writerIndex。

同时，ByteBuf也定义了大量对ByteBuf进行操作的抽象方法以供子类实现。

AbstracByteBuf作为其主要的子类，实现了大部分方法。首先看看定义的变量和构造方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14

// 读指针
int readerIndex;
// 写指针
int writerIndex;
// 保存读指针
private int markedReaderIndex;
// 保存写指针
private int markedWriterIndex;
// 最大分配容量
private int maxCapacity;
protected AbstractByteBuf(int maxCapacity) {
    checkPositiveOrZero(maxCapacity, "maxCapacity");
    this.maxCapacity = maxCapacity;
}

没有重写的方法则交给子类实现。再来看写操作的方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

public ByteBuf writeBytes(ByteBuf src) {
    writeBytes(src, src.readableBytes());
    return this;
}
public int readableBytes() {
    return writerIndex - readerIndex;
}
public ByteBuf writeBytes(ByteBuf src, int length) {
    // 参数验证
    if (checkBounds) {
        checkReadableBounds(src, length);
    }
    writeBytes(src, src.readerIndex(), length);
    src.readerIndex(src.readerIndex() + length);
    return this;
}
public ByteBuf writeBytes(ByteBuf src, int srcIndex, int length) {
    ensureWritable(length);
    setBytes(writerIndex, src, srcIndex, length);
    writerIndex += length;
    return this;
}

首先得到可读字节数，也就是写指针和读指针之间的长度。再调用自身的writeBytes方法，在保证写的长度小于可读长度的情况下再执行读操作。此处又掉了自身的writeBytes重载方法，首先对验证传入的length大于零，再对长度和容量的长度继续宁验证。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

public ByteBuf ensureWritable(int minWritableBytes) {
    ensureWritable0(checkPositiveOrZero(minWritableBytes, "minWritableBytes"));
    return this;
}

final void ensureWritable0(int minWritableBytes) {
    // 获取当前写入索引
    final int writerIndex = writerIndex();
    // 计算目标容量，即当前写入索引加上需要的最小可写字节数
    final int targetCapacity = writerIndex + minWritableBytes;

    // 使用非短路的 & 运算符来减少分支 - 这是一个热路径，目标容量应该很少溢出
    // 如果目标容量大于等于0且小于等于容量，则无需扩容
    if (targetCapacity >= 0 & targetCapacity <= capacity()) {
        // 确保缓冲区可访问
        // 引用计数为0时，抛出IllegalReferenceCountException异常
        ensureAccessible();
        return;
    }

    // 检查目标容量是否超出边界
    // 如果目标容量小于0或大于最大容量，则抛出索引越界异常
    if (checkBounds && (targetCapacity < 0 || targetCapacity > maxCapacity)) {
        ensureAccessible();
        // 抛出索引越界异常
        throw new IndexOutOfBoundsException(String.format(
            "writerIndex(%d) + minWritableBytes(%d) exceeds maxCapacity(%d): %s",
            writerIndex, minWritableBytes, maxCapacity, this));
    }

    // 获得可以写入的字节数
    final int fastWritable = maxFastWritableBytes();
    // 计算新的容量，如果fastWritable大于等于所需的最小可写字节数，则使用当前写入索引加上快速可写字节数
    // 否则，使用分配器计算新的容量
    int newCapacity = fastWritable >= minWritableBytes ? writerIndex + fastWritable
        : alloc().calculateNewCapacity(targetCapacity, maxCapacity);

    // 调整到新的容量
    capacity(newCapacity);
}

做了两个参数校验：

  • ByteBuf写入的长度如果要小于可写字节数，直接进行写操作，无需扩容
  • minWritableBytes+ writerIndex> maxCapacity 也就是需要写入的长度+写指针必须要小于最大分配的内存, 否则报错

如果需要写入的长度超过了可写字节数, 并且需要写入的长度+写指针不超过最大内存, 则就开始了ByteBuf的自动扩容。其中alloc()返回的是当前bytebuf返回的缓冲区分配器对象。接着看扩容的方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

public int calculateNewCapacity(int minNewCapacity, int maxCapacity) {
    // 检查最小新容量是否为正数
    checkPositiveOrZero(minNewCapacity, "minNewCapacity");

    // 如果最小新容量大于最大容量，抛出异常
    if (minNewCapacity > maxCapacity) {
        throw new IllegalArgumentException(String.format(
            "minNewCapacity: %d (expected: not greater than maxCapacity(%d)",
            minNewCapacity, maxCapacity));
    }

    // 定义阈值，表示4 MiB的页面大小
    final int threshold = CALCULATE_THRESHOLD; // 4 MiB page

    // 如果最小新容量等于阈值，直接返回阈值
    if (minNewCapacity == threshold) {
        return threshold;
    }

    // 如果最小新容量超过阈值，则不进行翻倍，而是增加阈值
    if (minNewCapacity > threshold) {
        int newCapacity = minNewCapacity / threshold * threshold; // 计算新的容量
        // 如果新的容量超过最大容量减去阈值，则将新的容量设为最大容量
        if (newCapacity > maxCapacity - threshold) {
            newCapacity = maxCapacity;
        } else {
            newCapacity += threshold; // 否则增加阈值
        }
        return newCapacity; // 返回计算后的新容量
    }

    // 如果最小新容量在64到阈值之间，找到下一个正的2的幂
    final int newCapacity = MathUtil.findNextPositivePowerOfTwo(Math.max(minNewCapacity, 64));
    // 返回新的容量，确保不超过最大容量
    return Math.min(newCapacity, maxCapacity);
}

也就是如果小于阈值(4mb)， 采用倍增的方式，如果大于阈值(4mb)，采用平移4mb的方式。

再回到writeBytes(ByteBuf src, int srcIndex, int length)方法：

1
2
3
4
5
6

public ByteBuf writeBytes(byte[] src, int srcIndex, int length) {
    ensureWritable(length);
    setBytes(writerIndex, src, srcIndex, length);
    writerIndex += length;
    return this;
}

接下来调用setBytes从当前byteBuf的writerIndex节点开始写入, 将源缓冲区src的读指针位置, 写lenght个字节。AbstractByteBuf并没有提供具体实现，最后交给不同的子类实现。写完后移动写指针。

ByteBuf分类

Pooled和Unpooled

  • pooled是从一块内存里去取一段连续内存封装成byteBuf。
  • Unpooled是分配的时候直接调用系统api进行实现。

Direct和Heap

  • 基于直接内存的ByteBuf, 具体标志是类名中包含单词Direct的ByteBuf。
  • 基于堆内存的ByteBuf, 具体标志是类名中包含单词heap的ByteBuf。

Safe和Unsafe

  • Unsafe类型的byteBuf, 则可以直接拿到byteBuf在jvm中的具体内存, 可以通过调用jdk的Usafe对象进行读写
  • 非Unsafe不能拿到jvm的具体内存

在实际debug中能够发现实际上使用的时PooledUnsafeDirectByteBuf (windows 11)，以下以Pooled来介绍HeapBuf和DirectBuf

首先看PooledHeapByteBuf

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

class PooledHeapByteBuf extends PooledByteBuf<byte[]> {

    private static final ObjectPool<PooledHeapByteBuf> RECYCLER = ObjectPool.newPool(
            new ObjectCreator<PooledHeapByteBuf>() {
        @Override
        public PooledHeapByteBuf newObject(Handle<PooledHeapByteBuf> handle) {
            return new PooledHeapByteBuf(handle, 0);
        }
    });

    static PooledHeapByteBuf newInstance(int maxCapacity) {
        PooledHeapByteBuf buf = RECYCLER.get();
        buf.reuse(maxCapacity);
        return buf;
    }

    PooledHeapByteBuf(Handle<? extends PooledHeapByteBuf> recyclerHandle, int maxCapacity) {
        super(recyclerHandle, maxCapacity);
    }
	// 省略代码
}

首先注意到存在一个类型为ObjectPool的类变量RECYCLER，根据注释发现这是一个对象池，用于复用Buf，当对象池中存在可以复用的对象即复用该对象，否则重新创建对象，初始化时也定义了创建对象的方法。并且还存在一个static方法newInstance用于获取对象，即从对象池中获取对象，重新初始化后返回，这也解释了为什么构造器的修饰符为defaul。

再来看构造器的执行流程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

PooledHeapByteBuf(Handle<? extends PooledHeapByteBuf> recyclerHandle, int maxCapacity) {
    super(recyclerHandle, maxCapacity);
}
protected PooledByteBuf(Handle<? extends PooledByteBuf<T>> recyclerHandle, int maxCapacity) {
    super(maxCapacity);
    this.recyclerHandle = (EnhancedHandle<PooledByteBuf<T>>) recyclerHandle;
}
protected AbstractReferenceCountedByteBuf(int maxCapacity) {
    super(maxCapacity);
    updater.setInitialValue(this);
}
protected AbstractByteBuf(int maxCapacity) {
    checkPositiveOrZero(maxCapacity, "maxCapacity");
    this.maxCapacity = maxCapacity;
}

首先调用PooledByteBuf的构造器来初始化recyclerHandle：当引用数为0是释放资源。

AbstractReferenceCountedByteBuf构造器中初始化了updater，也就是Buf的引用计数器。

最后到AbstractByteBuf中初始化最大容量。

创建完成后使用setBytes方法初始化PooledByteBuf的成员变量T memory。根据PooledHeapByteBuf得知Buf最后的数据结构为byte[]。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

abstract class PooledByteBuf<T> extends AbstractReferenceCountedByteBuf {

    private final EnhancedHandle<PooledByteBuf<T>> recyclerHandle;

    protected PoolChunk<T> chunk;
    protected long handle;
    protected T memory;
    protected int offset;
    protected int length;
    int maxLength;
    PoolThreadCache cache;
    ByteBuffer tmpNioBuf;
    private ByteBufAllocator allocator;
}

class PooledHeapByteBuf extends PooledByteBuf<byte[]>{
	// 省略代码
}

1
2
3
4
5
6
7
8
9
10
11

public final ByteBuf setBytes(int index, ByteBuf src, int srcIndex, int length) {
    checkSrcIndex(index, length, srcIndex, src.capacity());
    if (src.hasMemoryAddress()) {
        PlatformDependent.copyMemory(src.memoryAddress() + srcIndex, memory, idx(index), length);
    } else if (src.hasArray()) {
        setBytes(index, src.array(), src.arrayOffset() + srcIndex, length);
    } else {
        src.getBytes(srcIndex, memory, idx(index), length);
    }
    return this;
}

接着看PooledDirectByteBuf

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

final class PooledDirectByteBuf extends PooledByteBuf<ByteBuffer> {

    private static final ObjectPool<PooledDirectByteBuf> RECYCLER = ObjectPool.newPool(
            new ObjectCreator<PooledDirectByteBuf>() {
        @Override
        public PooledDirectByteBuf newObject(Handle<PooledDirectByteBuf> handle) {
            return new PooledDirectByteBuf(handle, 0);
        }
    });

    static PooledDirectByteBuf newInstance(int maxCapacity) {
        PooledDirectByteBuf buf = RECYCLER.get();
        buf.reuse(maxCapacity);
        return buf;
    }

    private PooledDirectByteBuf(Handle<PooledDirectByteBuf> recyclerHandle, int maxCapacity) {
        super(recyclerHandle, maxCapacity);
    }
 	// 省略代码   
}

与PooledHeapByteBuf一样，通过newInstance创建对象。但二者的访问修饰符有所区别，PooledHeapByteBuf不为private的原因为PooledUnsafeHeapByteBuf继承了PooledHeapByteBuf。

目前为止没有看到Buf数据的初始化，接着往下走发现存在init方法来初始化成员变量。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

void init(PoolChunk<ByteBuffer> chunk, ByteBuffer nioBuffer,
          long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
    super.init(chunk, nioBuffer, handle, offset, length, maxLength, cache);
    initMemoryAddress();
}
PooledByteBuf#init
    void init(PoolChunk<T> chunk, ByteBuffer nioBuffer,
              long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
    init0(chunk, nioBuffer, handle, offset, length, maxLength, cache);
}

private void init0(PoolChunk<T> chunk, ByteBuffer nioBuffer,
                   long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
    assert handle >= 0;
    assert chunk != null;
    assert !PoolChunk.isSubpage(handle) ||
        chunk.arena.sizeClass.size2SizeIdx(maxLength) <= chunk.arena.sizeClass.smallMaxSizeIdx:
    "Allocated small sub-page handle for a buffer size that isn't \"small.\"";

    chunk.incrementPinnedMemory(maxLength);
    this.chunk = chunk;
    memory = chunk.memory;
    tmpNioBuf = nioBuffer;
    allocator = chunk.arena.parent;
    this.cache = cache;
    this.handle = handle;
    this.offset = offset;
    this.length = length;
    this.maxLength = maxLength;
}

初始化后调用再初始化Buf的内存地址

1
2
3

private void initMemoryAddress() {
    memoryAddress = PlatformDependent.directBufferAddress(memory) + offset;
}

ByteBufAllocator

在以上两个类的初始化中都传入了ByteBufAllocator，顾名思义就是分配缓冲区的工具，抽象类AbstractByteBufAllocator实现了ByteBufAllocator接口, 并且实现了其大部分功能。先来看看分配buffer的方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

public ByteBuf buffer() {
    if (directByDefault) {
        return directBuffer();
    }
    return heapBuffer();
}

/**
* Instance use heap buffers by default
*/
protected AbstractByteBufAllocator() {
    this(false);
}

protected AbstractByteBufAllocator(boolean preferDirect) {
    directByDefault = preferDirect && PlatformDependent.hasUnsafe();
    emptyBuf = new EmptyByteBuf(this);
}

根据成员变量directByDefault来判断创建那种Buf。从构造方法可以发现，默认为创建HeapBuf，需要创建的DirectBuf时还需要系统支持Unsafe。

先来看directBuf的创建：

1
2
3
4
5
6
7
8
9
10
11
12
13

public ByteBuf directBuffer() {
    return directBuffer(DEFAULT_INITIAL_CAPACITY, DEFAULT_MAX_CAPACITY);
}

public ByteBuf directBuffer(int initialCapacity, int maxCapacity) {
    if (initialCapacity == 0 && maxCapacity == 0) {
        return emptyBuf;
    }
    validate(initialCapacity, maxCapacity);
    return newDirectBuffer(initialCapacity, maxCapacity);
}

protected abstract ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity);

调用了重载的directBuffer方法，传入了默认初始容量和默认最大容量，分别为256和Integer. MAX_VALUE，具体的创建交给子类实现。

heapBuf与directBuf类似，不再赘述。

newDirectBuffer和newHeapBuffer两个抽象方法中, 在其子类PooledByteBufAllocator和UnpooledByteBufAllocator中都有实现

以PooledByteBufAllocator为例，先来看newHeapBuffer：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

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);
}

1

PoolThreadCache cache = threadCache.get();

首先获取当前线程的缓存。这缓存对象为PooledByteBufAllocator的内部类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

private final class PoolThreadLocalCache extends FastThreadLocal<PoolThreadCache> {
    private final boolean useCacheForAllThreads;

    PoolThreadLocalCache(boolean useCacheForAllThreads) {
        this.useCacheForAllThreads = useCacheForAllThreads;
    }

    @Override
    protected synchronized PoolThreadCache initialValue() {
        final PoolArena<byte[]> heapArena = leastUsedArena(heapArenas);
        final PoolArena<ByteBuffer> directArena = leastUsedArena(directArenas);

        final Thread current = Thread.currentThread();
        final EventExecutor executor = ThreadExecutorMap.currentExecutor();

        if (useCacheForAllThreads ||
            // If the current thread is a FastThreadLocalThread we will always use the cache
            current instanceof FastThreadLocalThread ||
            // The Thread is used by an EventExecutor, let's use the cache as the chances are good that we
            // will allocate a lot!
            executor != null) {
            final PoolThreadCache cache = new PoolThreadCache(
                heapArena, directArena, smallCacheSize, normalCacheSize,
                DEFAULT_MAX_CACHED_BUFFER_CAPACITY, DEFAULT_CACHE_TRIM_INTERVAL, useCacheFinalizers(current));

            if (DEFAULT_CACHE_TRIM_INTERVAL_MILLIS > 0) {
                if (executor != null) {
                    executor.scheduleAtFixedRate(trimTask, DEFAULT_CACHE_TRIM_INTERVAL_MILLIS,
                                                 DEFAULT_CACHE_TRIM_INTERVAL_MILLIS, TimeUnit.MILLISECONDS);
                }
            }
            return cache;
        }
        // No caching so just use 0 as sizes.
        return new PoolThreadCache(heapArena, directArena, 0, 0, 0, 0, false);
    }
}

该类继承了一个与ThreadLocal类似的缓存FastThreadLocal。

在重写的initialValue方法中初始化了heapArena和directArena两个属性之后, 通过new PoolThreadCache()这种方式创建了PoolThreadCache对象。

PoolThreadLocalCache是一个FastThreadLocal, 而PoolThreadCache才是线程局部缓存, 这两个类名非常非常像, 千万别搞混了

其中heapArena和directArena是分别是用来分配堆和堆外内存用的两个对象, 以directArena为例, 通过leastUsedArena(directArenas)这种方式获得, directArenas是一个directArena类型的数组, leastUsedArena(directArenas)这个方法是用来获取数组中一个使用最少的directArena对象

directArenas是PooledByteBufAllocator的成员变量, 是在其构造方法中初始化:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

public PooledByteBufAllocator(boolean preferDirect, int nHeapArena, int nDirectArena, int pageSize, int maxOrder,
                              int smallCacheSize, int normalCacheSize,
                              boolean useCacheForAllThreads, int directMemoryCacheAlignment) {
	// 省略代码
    if (nHeapArena > 0) {
        heapArenas = newArenaArray(nHeapArena);
        List<PoolArenaMetric> metrics = new ArrayList<PoolArenaMetric>(heapArenas.length);
        final SizeClasses sizeClasses = new SizeClasses(pageSize, pageShifts, chunkSize, 0);
        for (int i = 0; i < heapArenas.length; i ++) {
            PoolArena.HeapArena arena = new PoolArena.HeapArena(this, sizeClasses);
            heapArenas[i] = arena;
            metrics.add(arena);
        }
        heapArenaMetrics = Collections.unmodifiableList(metrics);
    } else {
        heapArenas = null;
        heapArenaMetrics = Collections.emptyList();
    }

    if (nDirectArena > 0) {
        directArenas = newArenaArray(nDirectArena);
        List<PoolArenaMetric> metrics = new ArrayList<PoolArenaMetric>(directArenas.length);
        final SizeClasses sizeClasses = new SizeClasses(pageSize, pageShifts, chunkSize,
                directMemoryCacheAlignment);
        for (int i = 0; i < directArenas.length; i ++) {
            PoolArena.DirectArena arena = new PoolArena.DirectArena(this, sizeClasses);
            directArenas[i] = arena;
            metrics.add(arena);
        }
        directArenaMetrics = Collections.unmodifiableList(metrics);
    } else {
        directArenas = null;
        directArenaMetrics = Collections.emptyList();
    }
    metric = new PooledByteBufAllocatorMetric(this);
}

可以的看到heapArenas通过heapArenas = newArenaArray(nHeapArena)初始化。初始值为

1
2
3
4
5
6
7
8
9
10
11
12

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)));

看看newArenaArray

1
2
3

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

这里仅创建了PoolArena数组。创建完成后在循环中为数组赋值。

再回到newDirectBuffer方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

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);
}

在PoolThreadCache构造方法中将directArena和heapArena中保存在成员变量中, 这样就可以直接通过cache.directArena这种方式拿到其成员变量的内容。再根据是否为null开始创建Buf。

到这里大概明白了arena的作用，Allocator创建和线程数量相等的arena, 并以数组的形式存储在PooledByteBufAllocator的成员变量中, 每一个PoolThreadCache创建的时候, 都会在当前线程拿到一个arena, 并保存在自身的成员变量中。

PoolThreadCache除了维护了arena之外, 还维护了一个缓存列表, 我们在重复分配ByteBuf的时候, 并不需要每次都通过arena进行分配, 可以直接从缓存列表中拿一个ByteBuf。

在PooledByteBufAllocator中维护了2个值：

  • smallCacheSize
  • normalCacheSize

代表能保存多少个不同类型的缓存。注意这里抛弃了tinyCache，融入了smallCache。

回到PoolThreadLocalCache的构造方法能够发现这些值通过PoolThreadCache的构造方法传入。

1
2
3
4
5
6
7
8
9
10
11

private static final int DEFAULT_SMALL_CACHE_SIZE;
private static final int DEFAULT_NORMAL_CACHE_SIZE;

static {
    DEFAULT_SMALL_CACHE_SIZE = SystemPropertyUtil.getInt("io.netty.allocator.smallCacheSize", 256);
    DEFAULT_NORMAL_CACHE_SIZE = SystemPropertyUtil.getInt("io.netty.allocator.normalCacheSize", 64);
}

public PooledByteBufAllocator(boolean preferDirect) {
    this(preferDirect, DEFAULT_NUM_HEAP_ARENA, DEFAULT_NUM_DIRECT_ARENA, DEFAULT_PAGE_SIZE, DEFAULT_MAX_ORDER);
}

传入了静态类变量，分别初始化为256，64。在重载的构造方法中初始化缓存大小。

1
2
3
4
5
6
7
8
9

public PooledByteBufAllocator(boolean preferDirect, int nHeapArena, int nDirectArena, int pageSize, int maxOrder,
                              int smallCacheSize, int normalCacheSize,
                              boolean useCacheForAllThreads, int directMemoryCacheAlignment) {
    super(preferDirect);
    threadCache = new PoolThreadLocalCache(useCacheForAllThreads);
    this.smallCacheSize = smallCacheSize;
    this.normalCacheSize = normalCacheSize;
    // 省略代码
}

这也就是PoolThreadLocalCache initialValue所使用的参数。

回到PoolThreadCache的创建：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

PoolThreadCache(PoolArena<byte[]> heapArena, PoolArena<ByteBuffer> directArena,
                int smallCacheSize, int normalCacheSize, int maxCachedBufferCapacity,
                int freeSweepAllocationThreshold, boolean useFinalizer) {
    checkPositiveOrZero(maxCachedBufferCapacity, "maxCachedBufferCapacity");
    this.freeSweepAllocationThreshold = freeSweepAllocationThreshold;
    this.heapArena = heapArena;
    this.directArena = directArena;
    if (directArena != null) {
        smallSubPageDirectCaches = createSubPageCaches(smallCacheSize, directArena.sizeClass.nSubpages);
        normalDirectCaches = createNormalCaches(normalCacheSize, maxCachedBufferCapacity, directArena);
        directArena.numThreadCaches.getAndIncrement();
    } else {
        // No directArea is configured so just null out all caches
        smallSubPageDirectCaches = null;
        normalDirectCaches = null;
    }
    if (heapArena != null) {
        // Create the caches for the heap allocations
        smallSubPageHeapCaches = createSubPageCaches(smallCacheSize, heapArena.sizeClass.nSubpages);
        normalHeapCaches = createNormalCaches(normalCacheSize, maxCachedBufferCapacity, heapArena);
        heapArena.numThreadCaches.getAndIncrement();
    } else {
        // No heapArea is configured so just null out all caches
        smallSubPageHeapCaches = null;
        normalHeapCaches = null;
    }

    // Only check if there are caches in use.
    if ((smallSubPageDirectCaches != null || normalDirectCaches != null
         || smallSubPageHeapCaches != null || normalHeapCaches != null)
        && freeSweepAllocationThreshold < 1) {
        throw new IllegalArgumentException("freeSweepAllocationThreshold: "
                                           + freeSweepAllocationThreshold + " (expected: > 0)");
    }
    freeOnFinalize = useFinalizer ? new FreeOnFinalize(this) : null;
}

现在关注smallCache和normalCache的初始化，也就对应着PoolThreadCache中的成员变量

1
2
3
4
5

// Hold the caches for the different size classes, which are small and normal.
private final MemoryRegionCache<byte[]>[] smallSubPageHeapCaches;
private final MemoryRegionCache<ByteBuffer>[] smallSubPageDirectCaches;
private final MemoryRegionCache<byte[]>[] normalHeapCaches;
private final MemoryRegionCache<ByteBuffer>[] normalDirectCaches;

首先看smallSubPageDirectCaches = createSubPageCaches(smallCacheSize, directArena.sizeClass.nSubpages);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

private static <T> MemoryRegionCache<T>[] createSubPageCaches(
    int cacheSize, int numCaches) {
    if (cacheSize > 0 && numCaches > 0) {
        @SuppressWarnings("unchecked")
        MemoryRegionCache<T>[] cache = new MemoryRegionCache[numCaches];
        for (int i = 0; i < cache.length; i++) {
            // TODO: maybe use cacheSize / cache.length
            cache[i] = new SubPageMemoryRegionCache<T>(cacheSize);
        }
        return cache;
    } else {
        return null;
    }
}

private abstract static class MemoryRegionCache<T> {
    private final int size;
    private final Queue<Entry<T>> queue;
    private final SizeClass sizeClass;
    private int allocations;

    MemoryRegionCache(int size, SizeClass sizeClass) {
        this.size = MathUtil.safeFindNextPositivePowerOfTwo(size);
        queue = PlatformDependent.newFixedMpscQueue(this.size);
        this.sizeClass = sizeClass;
    }
    // 省略代码
}

从代码中可以发现：创建了numCaches个大小为cacheSize的SubPageMemoryRegionCache。这个类继承于MemoryRegionCache，其中使用Queue来保存缓存。

接着看

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

private static <T> MemoryRegionCache<T>[] createNormalCaches(
        int cacheSize, int maxCachedBufferCapacity, PoolArena<T> area) {
    if (cacheSize > 0 && maxCachedBufferCapacity > 0) {
        int max = Math.min(area.sizeClass.chunkSize, maxCachedBufferCapacity);
        // Create as many normal caches as we support based on how many sizeIdx we have and what the upper
        // bound is that we want to cache in general.
        List<MemoryRegionCache<T>> cache = new ArrayList<MemoryRegionCache<T>>() ;
        for (int idx = area.sizeClass.nSubpages; idx < area.sizeClass.nSizes &&
                area.sizeClass.sizeIdx2size(idx) <= max; idx++) {
            cache.add(new NormalMemoryRegionCache<T>(cacheSize));
        }
        return cache.toArray(new MemoryRegionCache[0]);
    } else {
        return null;
    }
}

与smallCache的情形类似，只不过用于保存缓存的不是数组而是Arraylist。

再来看NormalMemoryRegionCache和SubPageMemoryRegionCache的区别

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

/**
 * Cache used for buffers which are backed by TINY or SMALL size.
 */
private static final class SubPageMemoryRegionCache<T> extends MemoryRegionCache<T> {
    SubPageMemoryRegionCache(int size) {
        super(size, SizeClass.Small);
    }

    @Override
    protected void initBuf(
            PoolChunk<T> chunk, ByteBuffer nioBuffer, long handle, PooledByteBuf<T> buf, int reqCapacity,
            PoolThreadCache threadCache) {
        chunk.initBufWithSubpage(buf, nioBuffer, handle, reqCapacity, threadCache);
    }
}

/**
 * Cache used for buffers which are backed by NORMAL size.
 */
private static final class NormalMemoryRegionCache<T> extends MemoryRegionCache<T> {
    NormalMemoryRegionCache(int size) {
        super(size, SizeClass.Normal);
    }

    @Override
    protected void initBuf(
            PoolChunk<T> chunk, ByteBuffer nioBuffer, long handle, PooledByteBuf<T> buf, int reqCapacity,
            PoolThreadCache threadCache) {
        chunk.initBuf(buf, nioBuffer, handle, reqCapacity, threadCache);
    }
}

从类定义上来看二者并无区别，只是在创建时向父类传递的类型(SizeClass)存在区别。

// TODO：io.netty.buffer.PoolChunk#initBuf中二者的区别

directArena分配Buf

回到newDirectBuffer方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

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);
}

当directArena不为空时，通过allocate方法分配了一个PoolArena<ByteBuffer>。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

PooledByteBuf<T> allocate(PoolThreadCache cache, int reqCapacity, int maxCapacity) {
    PooledByteBuf<T> buf = newByteBuf(maxCapacity);
    allocate(cache, buf, reqCapacity);
    return buf;
}

protected PooledByteBuf<ByteBuffer> newByteBuf(int maxCapacity) {
    if (HAS_UNSAFE) {
        return PooledUnsafeDirectByteBuf.newInstance(maxCapacity);
    } else {
        return PooledDirectByteBuf.newInstance(maxCapacity);
    }
}

private void allocate(PoolThreadCache cache, PooledByteBuf<T> buf, final int reqCapacity) {
    final int sizeIdx = sizeClass.size2SizeIdx(reqCapacity);

    if (sizeIdx <= sizeClass.smallMaxSizeIdx) {
        tcacheAllocateSmall(cache, buf, reqCapacity, sizeIdx);
    } else if (sizeIdx < sizeClass.nSizes) {
        tcacheAllocateNormal(cache, buf, reqCapacity, sizeIdx);
    } else {
        int normCapacity = sizeClass.directMemoryCacheAlignment > 0
            ? sizeClass.normalizeSize(reqCapacity) : reqCapacity;
        // Huge allocations are never served via the cache so just call allocateHuge
        allocateHuge(buf, normCapacity);
    }
}

首先通过newByteBuf获得Buf对象，在通过allocate进行分配。newByteBuf中通过newInstance静态方法获得Buf，这里的静态方法也就是前面提到的从对象池中获取对象。

在allocate方法中，首先计算出规范后的缓存容量，再根据容量大小来决定分配那种缓存。如果small和normal均不满足，则申请huge缓存。

以tcacheAllocateSmall为例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

private void tcacheAllocateSmall(PoolThreadCache cache, PooledByteBuf<T> buf, final int reqCapacity,
                                 final int sizeIdx) {

    // 尝试从缓存中分配小型内存块
    if (cache.allocateSmall(this, buf, reqCapacity, sizeIdx)) {
        // 如果成功从缓存中分配内存，则直接返回
        return;
    }

    /*
         * 在头部上进行同步。这是必要的，因为 {@link PoolChunk#allocateSubpage(int)} 和
         * {@link PoolChunk#free(long)} 可能会修改双向链表。
         */
    final PoolSubpage<T> head = smallSubpagePools[sizeIdx]; // 获取对应大小索引的头部
    final boolean needsNormalAllocation; // 标记是否需要正常分配
    head.lock(); // 锁定头部以进行安全操作
    try {
        final PoolSubpage<T> s = head.next; // 获取下一个子页面
        needsNormalAllocation = s == head; // 检查是否需要正常分配
        if (!needsNormalAllocation) {
            // 确保子页面未被销毁且元素大小与索引匹配
            assert s.doNotDestroy && s.elemSize == sizeClass.sizeIdx2size(sizeIdx) : "doNotDestroy=" +
                s.doNotDestroy + ", elemSize=" + s.elemSize + ", sizeIdx=" + sizeIdx;
            long handle = s.allocate(); // 从子页面分配内存
            assert handle >= 0; // 确保分配成功
            s.chunk.initBufWithSubpage(buf, null, handle, reqCapacity, cache); // 初始化缓冲区
        }
    } finally {
        head.unlock(); // 解锁头部
    }

    // 如果需要正常分配，则进行正常分配
    if (needsNormalAllocation) {
        lock(); // 锁定以进行安全操作
        try {
            allocateNormal(buf, reqCapacity, sizeIdx, cache); // 调用正常分配方法
        } finally {
            unlock(); // 解锁
        }
    }

    incSmallAllocation(); // 增加小型分配计数
}

首先从缓存中分配内存

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

boolean allocateSmall(PoolArena<?> area, PooledByteBuf<?> buf, int reqCapacity, int sizeIdx) {
    return allocate(cacheForSmall(area, sizeIdx), buf, reqCapacity);
}

private MemoryRegionCache<?> cacheForSmall(PoolArena<?> area, int sizeIdx) {
    if (area.isDirect()) {
        return cache(smallSubPageDirectCaches, sizeIdx);
    }
    return cache(smallSubPageHeapCaches, sizeIdx);
}

private static <T> MemoryRegionCache<T> cache(MemoryRegionCache<T>[] cache, int sizeIdx) {
    if (cache == null || sizeIdx > cache.length - 1) {
        return null;
    }
    return cache[sizeIdx];
}

通过cacheForSmall方法来获取位于sizeIdx的缓存，如果规范化后得到的sizeIdx在缓存的范围内的话，返回cache[sizeIdx]。

随后进入allocate方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

private boolean allocate(MemoryRegionCache<?> cache, PooledByteBuf buf, int reqCapacity) {
    if (cache == null) {
        // no cache found so just return false here
        return false;
    }
    boolean allocated = cache.allocate(buf, reqCapacity, this);
    if (++ allocations >= freeSweepAllocationThreshold) {
        allocations = 0;
        trim();
    }
    return allocated;
}

public final boolean allocate(PooledByteBuf<T> buf, int reqCapacity, PoolThreadCache threadCache) {
    Entry<T> entry = queue.poll();
    if (entry == null) {
        return false;
    }
    initBuf(entry.chunk, entry.nioBuffer, entry.handle, buf, reqCapacity, threadCache);
    entry.unguardedRecycle();

    // allocations is not thread-safe which is fine as this is only called from the same thread all time.
    ++ allocations;
    return true;
}

如果cache不为空，调用cache的allocate方法继续分配。这里会尝试从io.netty.buffer.PoolThreadCache.MemoryRegionCache#queue来获取已分配的缓存，如果queue不为空，获得entry后开始初始化；为空则返回false等到后续在分配。

回到PoolThreadCache#allocate

1
2
3
4
5
6
7
8
9
10
11
12

private boolean allocate(MemoryRegionCache<?> cache, PooledByteBuf buf, int reqCapacity) {
    if (cache == null) {
        // no cache found so just return false here
        return false;
    }
    boolean allocated = cache.allocate(buf, reqCapacity, this);
    if (++ allocations >= freeSweepAllocationThreshold) {
        allocations = 0;
        trim();
    }
    return allocated;
}

之后再判断已分配的容量是否到了GC的阈值，是的话开始释放资源，在此不做展开。

1
2
3
4
5
6

void trim() {
    trim(smallSubPageDirectCaches);
    trim(normalDirectCaches);
    trim(smallSubPageHeapCaches);
    trim(normalHeapCaches);
}

再回到io.netty.buffer.PoolArena#tcacheAllocateSmall

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

private void tcacheAllocateSmall(PoolThreadCache cache, PooledByteBuf<T> buf, final int reqCapacity,
                                 final int sizeIdx) {

    // 尝试从缓存中分配小型内存块
    if (cache.allocateSmall(this, buf, reqCapacity, sizeIdx)) {
        // 如果成功从缓存中分配内存，则直接返回
        return;
    }

    /*
         * 在头部上进行同步。这是必要的，因为 {@link PoolChunk#allocateSubpage(int)} 和
         * {@link PoolChunk#free(long)} 可能会修改双向链表。
         */
    final PoolSubpage<T> head = smallSubpagePools[sizeIdx]; // 获取对应大小索引的头部
    final boolean needsNormalAllocation; // 标记是否需要正常分配
    head.lock(); // 锁定头部以进行安全操作
    try {
        final PoolSubpage<T> s = head.next; // 获取下一个子页面
        needsNormalAllocation = s == head; // 检查是否需要正常分配
        if (!needsNormalAllocation) {
            // 确保子页面未被销毁且元素大小与索引匹配
            assert s.doNotDestroy && s.elemSize == sizeClass.sizeIdx2size(sizeIdx) : "doNotDestroy=" +
                s.doNotDestroy + ", elemSize=" + s.elemSize + ", sizeIdx=" + sizeIdx;
            long handle = s.allocate(); // 从子页面分配内存
            assert handle >= 0; // 确保分配成功
            s.chunk.initBufWithSubpage(buf, null, handle, reqCapacity, cache); // 初始化缓冲区
        }
    } finally {
        head.unlock(); // 解锁头部
    }

    // 如果需要正常分配，则进行正常分配
    if (needsNormalAllocation) {
        lock(); // 锁定以进行安全操作
        try {
            allocateNormal(buf, reqCapacity, sizeIdx, cache); // 调用正常分配方法
        } finally {
            unlock(); // 解锁
        }
    }

    incSmallAllocation(); // 增加小型分配计数
}

在根据参数判断是否使用subpage缓存分配。此处先谈论normal的方式。

Normal 缓存

在进入allocateNormal方法前，对成员对象ReentrantLock lock进行加锁。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

    private void allocateNormal(PooledByteBuf<T> buf, int reqCapacity, int sizeIdx, PoolThreadCache threadCache) {
        // 确保当前线程持有锁
        assert lock.isHeldByCurrentThread();
        
        // 尝试从不同的内存池中分配内存
        // 如果任一分配成功，则直接返回
        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; // 如果成功分配，退出方法
        }

        // 如果所有池都无法分配内存，则创建一个新的内存块
        PoolChunk<T> c = newChunk(sizeClass.pageSize, sizeClass.nPSizes, sizeClass.pageShifts, sizeClass.chunkSize);
        
        // 尝试在新创建的内存块中分配所需的内存
        boolean success = c.allocate(buf, reqCapacity, sizeIdx, threadCache);
        
        // 确保分配成功
        assert success;
        
        // 将新创建的内存块添加到初始化队列中
        qInit.add(c);
    }

首先尝试从内存块池(PoolChunkList)中分配缓存，失败的话创建新的内存块再进行分配。

先来看内存块池分配缓存的逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

boolean allocate(PooledByteBuf<T> buf, int reqCapacity, int sizeIdx, PoolThreadCache threadCache) {
    int normCapacity = arena.sizeClass.sizeIdx2size(sizeIdx);
    if (normCapacity > maxCapacity) {
        // Either this PoolChunkList is empty or the requested capacity is larger then the capacity which can
        // be handled by the PoolChunks that are contained in this PoolChunkList.
        return false;
    }

    for (PoolChunk<T> cur = head; cur != null; cur = cur.next) {
        if (cur.allocate(buf, reqCapacity, sizeIdx, threadCache)) {
            if (cur.freeBytes <= freeMinThreshold) {
                remove(cur);
                nextList.add(cur);
            }
            return true;
        }
    }
    return false;
}

boolean allocate(PooledByteBuf<T> buf, int reqCapacity, int sizeIdx, PoolThreadCache cache) {
    final long handle; // 用于存储分配的句柄
    // 检查请求的大小索引是否在小型内存块的范围内
    if (sizeIdx <= arena.sizeClass.smallMaxSizeIdx) {
        final PoolSubpage<T> nextSub; // 用于存储下一个子页面
        // 小型内存块的分配
        // 获取 PoolArena 拥有的 PoolSubPage 池的头部并对其进行同步
        // 这是必要的，因为我们可能会将其添加回去，从而改变链表结构
        PoolSubpage<T> head = arena.smallSubpagePools[sizeIdx];
        head.lock(); // 锁定头部以确保线程安全
        try {
            nextSub = head.next; // 获取下一个子页面
            // 检查下一个子页面是否有效且未被销毁
            if (nextSub != head) {
                assert nextSub.doNotDestroy && nextSub.elemSize == arena.sizeClass.sizeIdx2size(sizeIdx) :
                "doNotDestroy=" + nextSub.doNotDestroy + ", elemSize=" + nextSub.elemSize + ", sizeIdx=" +
                    sizeIdx;
                handle = nextSub.allocate(); // 从下一个子页面分配内存
                assert handle >= 0; // 确保分配成功
                assert isSubpage(handle); // 确保句柄是子页面
                nextSub.chunk.initBufWithSubpage(buf, null, handle, reqCapacity, cache); // 初始化缓冲区
                return true; // 返回成功
            }
            // 如果没有可用的子页面，则尝试分配新的子页面
            handle = allocateSubpage(sizeIdx, head);
            if (handle < 0) {
                return false; // 分配失败
            }
            assert isSubpage(handle); // 确保句柄是子页面
        } finally {
            head.unlock(); // 解锁头部
        }
    } else {
        // 正常内存块的分配
        // runSize 必须是 pageSize 的倍数
        int runSize = arena.sizeClass.sizeIdx2size(sizeIdx); // 获取请求大小对应的运行大小
        handle = allocateRun(runSize); // 分配运行
        if (handle < 0) {
            return false; // 分配失败
        }
        assert !isSubpage(handle); // 确保句柄不是子页面
    }

    // 从缓存中获取 NIO 缓冲区（如果存在）
    ByteBuffer nioBuffer = cachedNioBuffers != null ? cachedNioBuffers.pollLast() : null;
    // 初始化缓冲区
    initBuf(buf, nioBuffer, handle, reqCapacity, cache);
    return true; // 返回成功
}

遍历列表中的每个元素，尝试进行分配。由于此处是第一次进入，每个内存块池均为空，具体逻辑先跳过。

// TODO

内存块池分配失败后创建新的内存块。根据是否需要对齐进行内存分配。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

protected PoolChunk<ByteBuffer> newChunk(int pageSize, int maxPageIdx,
                                         int pageShifts, int chunkSize) {
    // 检查直接内存缓存对齐是否为 0
    if (sizeClass.directMemoryCacheAlignment == 0) {
        // 分配直接内存
        ByteBuffer memory = allocateDirect(chunkSize);
        // 返回新的 PoolChunk 实例
        return new PoolChunk<ByteBuffer>(this, memory, memory, pageSize, pageShifts,
                                         chunkSize, maxPageIdx);
    }

    // 如果需要对齐，分配额外的内存
    final ByteBuffer base = allocateDirect(chunkSize + sizeClass.directMemoryCacheAlignment);
    // 对齐直接内存缓冲区
    final ByteBuffer memory = PlatformDependent.alignDirectBuffer(base, sizeClass.directMemoryCacheAlignment);
    // 返回新的 PoolChunk 实例
    return new PoolChunk<ByteBuffer>(this, base, memory, pageSize,
                                     pageShifts, chunkSize, maxPageIdx);
}

private static ByteBuffer allocateDirect(int capacity) {
    return PlatformDependent.useDirectBufferNoCleaner() ? PlatformDependent.allocateDirectNoCleaner(capacity)
        : ByteBuffer.allocateDirect(capacity);
}

分配好内存块之后，在此内存块上分配缓存。分配后开始初始化initBuf(buf, nioBuffer, handle, reqCapacity, cache);，最后将内存块放入内存块池。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

void initBuf(PooledByteBuf<T> buf, ByteBuffer nioBuffer, long handle, int reqCapacity,
             PoolThreadCache threadCache) {
    if (isSubpage(handle)) {
        initBufWithSubpage(buf, nioBuffer, handle, reqCapacity, threadCache);
    } else {
        int maxLength = runSize(pageShifts, handle);
        buf.init(this, nioBuffer, handle, runOffset(handle) << pageShifts,
                 reqCapacity, maxLength, arena.parent.threadCache());
    }
}

void initBufWithSubpage(PooledByteBuf<T> buf, ByteBuffer nioBuffer, long handle, int reqCapacity,
                        PoolThreadCache threadCache) {
    int runOffset = runOffset(handle);
    int bitmapIdx = bitmapIdx(handle);

    PoolSubpage<T> s = subpages[runOffset];
    assert s.isDoNotDestroy();
    assert reqCapacity <= s.elemSize : reqCapacity + "<=" + s.elemSize;

    int offset = (runOffset << pageShifts) + bitmapIdx * s.elemSize;
    buf.init(this, nioBuffer, handle, offset, reqCapacity, s.elemSize, threadCache);
}

void init(PoolChunk<ByteBuffer> chunk, ByteBuffer nioBuffer,
          long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
    super.init(chunk, nioBuffer, handle, offset, length, maxLength, cache);
    initMemoryAddress();
}

void init(PoolChunk<T> chunk, ByteBuffer nioBuffer,
          long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
    init0(chunk, nioBuffer, handle, offset, length, maxLength, cache);
}

private void init0(PoolChunk<T> chunk, ByteBuffer nioBuffer,
                   long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
    assert handle >= 0;
    assert chunk != null;
    assert !PoolChunk.isSubpage(handle) ||
        chunk.arena.sizeClass.size2SizeIdx(maxLength) <= chunk.arena.sizeClass.smallMaxSizeIdx:
    "Allocated small sub-page handle for a buffer size that isn't \"small.\"";

    chunk.incrementPinnedMemory(maxLength);
    this.chunk = chunk;
    memory = chunk.memory;
    tmpNioBuf = nioBuffer;
    allocator = chunk.arena.parent;
    this.cache = cache;
    this.handle = handle;
    this.offset = offset;
    this.length = length;
    this.maxLength = maxLength;
}

最后进入到init0对Buf进行初始化。

附PoolChunk内存模型。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

/-----------------\
| run             |
|                 |
|                 |
|-----------------|
| run             |
|                 |
|-----------------|
| unalloctated    |
| (freed)         |
|                 |
|-----------------|
| subpage         |
|-----------------|
| unallocated     |
| (freed)         |
| ...             |
| ...             |
| ...             |
|                 |
|                 |
|                 |
\-----------------/

Subpage缓存

// TODO：

现在来看看subpage缓存的分配

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

private void tcacheAllocateSmall(PoolThreadCache cache, PooledByteBuf<T> buf, final int reqCapacity,
                                 final int sizeIdx) {

    // 尝试从缓存中分配小型内存块
    if (cache.allocateSmall(this, buf, reqCapacity, sizeIdx)) {
        // 如果成功从缓存中分配内存，则直接返回
        return;
    }

    /*
         * 在头部上进行同步。这是必要的，因为 {@link PoolChunk#allocateSubpage(int)} 和
         * {@link PoolChunk#free(long)} 可能会修改双向链表。
         */
    final PoolSubpage<T> head = smallSubpagePools[sizeIdx]; // 获取对应大小索引的头部
    final boolean needsNormalAllocation; // 标记是否需要正常分配
    head.lock(); // 锁定头部以进行安全操作
    try {
        final PoolSubpage<T> s = head.next; // 获取下一个子页面
        needsNormalAllocation = s == head; // 检查是否需要正常分配
        if (!needsNormalAllocation) {
            // 确保子页面未被销毁且元素大小与索引匹配
            assert s.doNotDestroy && s.elemSize == sizeClass.sizeIdx2size(sizeIdx) : "doNotDestroy=" +
                s.doNotDestroy + ", elemSize=" + s.elemSize + ", sizeIdx=" + sizeIdx;
            long handle = s.allocate(); // 从子页面分配内存
            assert handle >= 0; // 确保分配成功
            s.chunk.initBufWithSubpage(buf, null, handle, reqCapacity, cache); // 初始化缓冲区
        }
    } finally {
        head.unlock(); // 解锁头部
    }

    // 如果需要正常分配，则进行正常分配
    if (needsNormalAllocation) {
        lock(); // 锁定以进行安全操作
        try {
            allocateNormal(buf, reqCapacity, sizeIdx, cache); // 调用正常分配方法
        } finally {
            unlock(); // 解锁
        }
    }

    incSmallAllocation(); // 增加小型分配计数
}

确定使用subpage分配后long handle = s.allocate();开始进行分配。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

long allocate() {
    // 检查是否有可用的元素，或者是否允许销毁
    if (numAvail == 0 || !doNotDestroy) {
        return -1; // 如果没有可用元素或不允许销毁，返回 -1
    }

    // 获取下一个可用的位图索引
    final int bitmapIdx = getNextAvail();
    // 检查位图索引是否有效
    if (bitmapIdx < 0) {
        removeFromPool(); // 如果位图索引无效，移除子页面以防止重复错误
        throw new AssertionError("No next available bitmap index found (bitmapIdx = " + bitmapIdx + "), " +
                                 "even though there are supposed to be (numAvail = " + numAvail + ") " +
                                 "out of (maxNumElems = " + maxNumElems + ") available indexes.");
    }
    // 计算位图索引的行和列
    int q = bitmapIdx >>> 6; // 行索引
    int r = bitmapIdx & 63;   // 列索引
    // 确保该位图位置未被占用
    assert (bitmap[q] >>> r & 1) == 0;
    // 标记该位图位置为已占用
    bitmap[q] |= 1L << r;

    // 减少可用元素的数量，并检查是否为零
    if (-- numAvail == 0) {
        removeFromPool(); // 如果没有可用元素，移除子页面
    }

    // 返回位图索引的句柄
    return toHandle(bitmapIdx);
}

private long toHandle(int bitmapIdx) {
    int pages = runSize >> pageShifts;
    return (long) runOffset << RUN_OFFSET_SHIFT
        | (long) pages << SIZE_SHIFT
        | 1L << IS_USED_SHIFT
        | 1L << IS_SUBPAGE_SHIFT
        | bitmapIdx;
}

分配完成后开始初始化

1
2
3
4
5
6
7
8
9
10
11
12

void initBufWithSubpage(PooledByteBuf<T> buf, ByteBuffer nioBuffer, long handle, int reqCapacity,
                        PoolThreadCache threadCache) {
    int runOffset = runOffset(handle);
    int bitmapIdx = bitmapIdx(handle);

    PoolSubpage<T> s = subpages[runOffset];
    assert s.isDoNotDestroy();
    assert reqCapacity <= s.elemSize : reqCapacity + "<=" + s.elemSize;

    int offset = (runOffset << pageShifts) + bitmapIdx * s.elemSize;
    buf.init(this, nioBuffer, handle, offset, reqCapacity, s.elemSize, threadCache);
}

init方法最后会调用上面提到的init0方法。

从零开始的Netty源码阅读
blog
本文由作者按照 CC BY 4.0 进行授权