ByteBuf内存分配和释放由具体实现负责,抽象类型只定义的内存分配和释放的时机。
内存分配分两个阶段: 第一阶段,初始化时分配内存。第二阶段: 内存不够用时分配新的内存。ByteBuf抽象层没有定义第一阶段的行为,但定义了第二阶段的方法:
public abstract ByteBuf capacity(int newCapacity)
这个方法负责分配一个长度为newCapacity的新内存。
内存释放的抽象实现在AbstractReferenceCountedByteBuf中实现,这个类实现引用计数,当调用release方法把引用计数变成0时,会调用
protected abstract void deallocate()
执行真正的内存释放操作。
内存相关的属性
ByteBuf定义了两个内存相关的属性:
capacity: 当前的当前的容量,也就是当前使用的内存大小。使用capacity()方法获得。
maxCapacity: 最大容量,也就是可以使用的最大内存大小。使用newCapacity()方法获得。
内存分配时机
AbstractByteBuf定义了内存分配的时机。当writeXX方法被调用的时候,如果如果发现可写空间不足,就调用capacity分配新的内存。下面以writeInt为例详细分析这个过程。
1 @Override
2 public ByteBuf writeInt(int value) {
3 ensureWritable0(4);
4 _setInt(writerIndex, value);
5 writerIndex += 4;
6 return this;
7 }
8
9
10 final void ensureWritable0(int minWritableBytes) {
11 ensureAccessible();
12 if (minWritableBytes <= writableBytes()) {
13 return;
14 }
15
16 if (minWritableBytes > maxCapacity - writerIndex) {
17 throw new IndexOutOfBoundsException(String.format(
18 "writerIndex(%d) + minWritableBytes(%d) exceeds maxCapacity(%d): %s",
19 writerIndex, minWritableBytes, maxCapacity, this));
20 }
21
22 // Normalize the current capacity to the power of 2.
23 int newCapacity = calculateNewCapacity(writerIndex + minWritableBytes);
24
25 // Adjust to the new capacity.
26 capacity(newCapacity);
27 }
28
3行: ensureWritable确保当前能写入4Byte的数据。
11行: 确保当前ByteBuf是可以访问的。防止多线程环境下,ByteBuf内存被释放后读写数据。
12,13行: 如果内存够用,就此作罢。
16行: 如果需要内存大于maxCapacity抛出异常。
23, 26行行: 计算新内存的大小, 调用capacity(int)分配新内存。
重新分配内存之前的一个重要步骤的计算新内存的大小。这个工作由calculateNewCapacity方法完成,它的代码如下:
1 private int calculateNewCapacity(int minNewCapacity) {
2 final int maxCapacity = this.maxCapacity;
3 final int threshold = 1048576 * 4; // 4 MiB page
4
5 if (minNewCapacity == threshold) {
6 return threshold;
7 }
8
9 // If over threshold, do not double but just increase by threshold.
10 if (minNewCapacity > threshold) {
11 int newCapacity = minNewCapacity / threshold * threshold;
12 if (newCapacity > maxCapacity - threshold) {
13 newCapacity = maxCapacity;
14 } else {
15 newCapacity += threshold;
16 }
17 return newCapacity;
18 }
19
20 // Not over threshold. Double up to 4 MiB, starting from 64.
21 int newCapacity = 64;
22 while (newCapacity < minNewCapacity) {
23 newCapacity <<= 1;
24 }
25
26 return Math.min(newCapacity, maxCapacity);
27 }
1行:接受一个最小的新内存参数minNewCapacity。
3行: 定义一个4MB的阈值常量threshold。
5,6行: 如果minNewCapacity==threshold,那么新内存大小就是threshold。
10-17行: 如果minNewCapacity>threshold, 新内存大小是min(maxCapacity, threshold * n)且threshold * n >= minNewCapacity。
21-26行: 如果minNewCapacity<threshold, 新内存大小是min(maxCapacity, 64 * 2^n)且64 * 2^n >= minNewCapacity。
内存分配和释放的具体实现
本章涉及到的内存分配和释放的具体实现只涉及到unpooled类型的ByteBuf,即:
UnpooledHeapByteBuf
UnpooledDirectByteBuf
UnpooledUnsafeHeapByteBuf
UnpooledUnsafeDirectByteBuf
这几个具体实现中涉及到的内存分配和释放代码比较简洁,更容易明白ByteBuf内存管理的原理。相比之下,pooled类型的ByteBuf内存分配和释放的代码要复杂很多,会在后面的章节独立分析。
UnpooledHeapByteBuf和UnpooledUnsafeHeapByteBuf实现
UnpooledHeapByteBuf中,内存分配的实现代码主要集中在capacity(int)和allocateArray()方法中。capacity分配新内存的步骤是
- 调用allocateArray分配一块新内存。
- 把旧内存中的实际复制到新内存中。
- 使用新内存替换旧内存(24行)。
- 释放掉旧内存(25行)。
代码如下:
1 @Override
2 public ByteBuf capacity(int newCapacity) {
3 checkNewCapacity(newCapacity);
4
5 int oldCapacity = array.length;
6 byte[] oldArray = array;
7 if (newCapacity > oldCapacity) {
8 byte[] newArray = allocateArray(newCapacity);
9 System.arraycopy(oldArray, 0, newArray, 0, oldArray.length);
10 setArray(newArray);
11 freeArray(oldArray);
12 } else if (newCapacity < oldCapacity) {
13 byte[] newArray = allocateArray(newCapacity);
14 int readerIndex = readerIndex();
15 if (readerIndex < newCapacity) {
16 int writerIndex = writerIndex();
17 if (writerIndex > newCapacity) {
18 writerIndex(writerIndex = newCapacity);
19 }
20 System.arraycopy(oldArray, readerIndex, newArray, readerIndex, writerIndex - readerIndex);
21 } else {
22 setIndex(newCapacity, newCapacity);
23 }
24 setArray(newArray);
25 freeArray(oldArray);
26 }
27 return this;
28 }
29
capacity中复制旧内存数据到新内存中的时候分两种情况(newCapacity,oldCapacity分别是新旧内存的大小):
- newCapacity>oldCapacity,这种情况比较简单,直接复制旧好了(第8行)。不影响readerIndex和writerIndex。
- newCapacity<oldCapacity, 这种情况比较复杂。capacity尽量把可读的数据复制新内存中。
- 如果readerIndex<newCapacity且writerIndex<newCapacity。可读数据会全部转移到新内存中。readerIndex和writerIndex保持不变。
- 如果readerIndex<newCapacity且writeIndex>newCapacity。可端数据会部分转移的新内存中,会丢失部分可读数据。readerIndex不变,writerIndex变成newCapacity。
- 如果readerIndex>newCapacity,数据全部丢失,readerIndex和writerIndex都会变成newCapacity。
allocateArray方法负责分配一块新内存,它的实现是new byte[]。freeArray方法负责释放内存,这个方法是个空方法。
UnpooledUnsafeHeapByteBuf是UnloopedHeadpByteBuf的直接子类,在内存管理上的差别是allocateArray的实现,UnpooledUnsafeHeapByteBuf的实现是:
1 @Override
2 byte[] allocateArray(int initialCapacity) {
3 return PlatformDependent.allocateUninitializedArray(initialCapacity);
4 }
UnpooledDirectByteBuf和UnpooledUnsafeDirectByteBuf实现
UnpooledDirectByteBuf类使用DirectByteBuffer作为内存,使用了DirectByteBuffer的能力来实现ByteBuf接口。allocateDirect和freeDirect方法负责分配和释放DirectByteBuffer。capacity(int)方法和UnloopedHeapByteBuf类似,使用allocateDirect创建一个新的DirectByteBuffer, 把旧内存数据复制到新内存中,然后使用新内存替换旧内存,最后调用freeDirect方法释放掉旧的DirectByteBuffer。
1 protected ByteBuffer allocateDirect(int initialCapacity) {
2 return ByteBuffer.allocateDirect(initialCapacity);
3 }
4
5 protected void freeDirect(ByteBuffer buffer) {
6 PlatformDependent.freeDirectBuffer(buffer);
7 }
8
9 @Override
10 public ByteBuf capacity(int newCapacity) {
11 checkNewCapacity(newCapacity);
12
13 int readerIndex = readerIndex();
14 int writerIndex = writerIndex();
15
16 int oldCapacity = capacity;
17 if (newCapacity > oldCapacity) {
18 ByteBuffer oldBuffer = buffer;
19 ByteBuffer newBuffer = allocateDirect(newCapacity);
20 oldBuffer.position(0).limit(oldBuffer.capacity());
21 newBuffer.position(0).limit(oldBuffer.capacity());
22 newBuffer.put(oldBuffer);
23 newBuffer.clear();
24 setByteBuffer(newBuffer);
25 } else if (newCapacity < oldCapacity) {
26 ByteBuffer oldBuffer = buffer;
27 ByteBuffer newBuffer = allocateDirect(newCapacity);
28 if (readerIndex < newCapacity) {
29 if (writerIndex > newCapacity) {
30 writerIndex(writerIndex = newCapacity);
31 }
32 oldBuffer.position(readerIndex).limit(writerIndex);
33 newBuffer.position(readerIndex).limit(writerIndex);
34 newBuffer.put(oldBuffer);
35 newBuffer.clear();
36 } else {
37 setIndex(newCapacity, newCapacity);
38 }
39 setByteBuffer(newBuffer);
40 }
41 return this;
42 }
对比UnloopedHeapByteBuf的capacity(int)方法,发现这两个实现非常类似,也分两组情况处理:
- 18-24行,newCapacity > oldCapacity的情况。
- 26-39行, newCapacity < oldCapacity的情况。
两种情况对readerIndex和writerIndex的影响也一样,不同的是数据复制时的具体实现。
UnpooledUnsafeDirectByteBuf和UnpooledDirectByteBuf同属于AbstractReferenceCountedByteBuf的派生类,它们之间没有继承关系。但内存分配和释放实现是一样的,不同的地方是内存I/O。UnpooledUnsafeDirectByteBuf使用UnsafeByteBufUtil类之间读写DirectByteBuffer的内存,没有使用DirectByteBuffer的I/O能力。