上一節(jié)查看了ByteBufAllocator，并了解了其抽象實(shí)現(xiàn)，和一些根據(jù)不同的內(nèi)存類型進(jìn)行內(nèi)存分配的思路。
本節(jié)研究UnpooledByteBufAllocator，包括heap和direct的內(nèi)存分配闷畸，以及Unsafe和非unsafe的區(qū)別。
關(guān)于heap內(nèi)存的分配
- 入口
@Override
protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
//判斷是有unsafe來(lái)分配
return PlatformDependent.hasUnsafe() ?
new InstrumentedUnpooledUnsafeHeapByteBuf(this, initialCapacity, maxCapacity) :
new InstrumentedUnpooledHeapByteBuf(this, initialCapacity, maxCapacity);
}
- 查看
new InstrumentedUnpooledUnsafeHeapByteBuf(this, initialCapacity, maxCapacity) :
發(fā)現(xiàn)分配Unpooled吞滞、Unsafe佑菩、Heap內(nèi)存，其實(shí)是分配了一個(gè)byte數(shù)組裁赠，并保存在UnpooledHeapByteBuf#array成員變量中殿漠。該內(nèi)存的初始值容量和最大可擴(kuò)展容量可以指定。
InstrumentedUnpooledUnsafeHeapByteBuf(UnpooledByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
super(alloc, initialCapacity, maxCapacity);
}
UnpooledUnsafeHeapByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
super(alloc, initialCapacity, maxCapacity);
}
public UnpooledHeapByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
super(maxCapacity);
checkNotNull(alloc, "alloc");
if (initialCapacity > maxCapacity) {
throw new IllegalArgumentException(String.format(
"initialCapacity(%d) > maxCapacity(%d)", initialCapacity, maxCapacity));
}
this.alloc = alloc;
//設(shè)置array
setArray(allocateArray(initialCapacity));
//設(shè)置readerIndex和writerIndex指針初始值為0
setIndex(0, 0);
}
protected byte[] allocateArray(int initialCapacity) {
//初始化了一個(gè)新的byte數(shù)組
return new byte[initialCapacity];
}
private void setArray(byte[] initialArray) {
//保存數(shù)組
array = initialArray;
tmpNioBuf = null;
}
@Override
public ByteBuf setIndex(int readerIndex, int writerIndex) {
if (checkBounds) {
checkIndexBounds(readerIndex, writerIndex, capacity());
}
//設(shè)置
setIndex0(readerIndex, writerIndex);
return this;
}
final void setIndex0(int readerIndex, int writerIndex) {
//設(shè)置讀寫指針
this.readerIndex = readerIndex;
this.writerIndex = writerIndex;
}
- 查看
new InstrumentedUnpooledHeapByteBuf(this, initialCapacity, maxCapacity);內(nèi)存分配佩捞，其實(shí)例化過(guò)成和InstrumentedUnpooledUnsafeHeapByteBuf一樣绞幌。在這里看不出safe和unsafe的區(qū)別，經(jīng)過(guò)之前的代碼一忱，可以重獲取的時(shí)候getByte方法進(jìn)入查看.
InstrumentedUnpooledHeapByteBuf(UnpooledByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
super(alloc, initialCapacity, maxCapacity);
}
public UnpooledHeapByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
super(maxCapacity);
checkNotNull(alloc, "alloc");
if (initialCapacity > maxCapacity) {
throw new IllegalArgumentException(String.format(
"initialCapacity(%d) > maxCapacity(%d)", initialCapacity, maxCapacity));
}
this.alloc = alloc;
//設(shè)置array
setArray(allocateArray(initialCapacity));
//設(shè)置readerIndex和writerIndex指針初始值為0
setIndex(0, 0);
}
- 查看
UnpooledHeapByteBuf#getByte()方法莲蜘，堆內(nèi)存類型的ByteBuf獲取的時(shí)候。直接通過(guò)下標(biāo)獲取byte數(shù)組中的byte
@Override
public byte getByte(int index) {
ensureAccessible();
return _getByte(index);
}
@Override
protected byte _getByte(int index) {
//該array為初始化的時(shí)候帘营，實(shí)例化的byte[]
return HeapByteBufUtil.getByte(array, index);
}
static byte getByte(byte[] memory, int index) {
//直接拿到一個(gè)數(shù)組
return memory[index];
}
- 查看
UnpooledUnsafeHeapByteBuf#getByte()方法票渠，獲取byte字節(jié)的時(shí)候，調(diào)用的是jdk的UNSAFE對(duì)象芬迄。
@Override
public byte getByte(int index) {
checkIndex(index);
return _getByte(index);
}
@Override
protected byte _getByte(int index) {
return UnsafeByteBufUtil.getByte(array, index);
}
static byte getByte(byte[] array, int index) {
return PlatformDependent.getByte(array, index);
}
public static byte getByte(byte[] data, int index) {
return PlatformDependent0.getByte(data, index);
}
static byte getByte(byte[] data, int index) {
//通過(guò)UNSAFE去獲取
return UNSAFE.getByte(data, BYTE_ARRAY_BASE_OFFSET + index);
}
關(guān)于direct內(nèi)存的分配
- 入口
UnpooledByteBufAllocator#newDirectBuffer()
@Override
protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
final ByteBuf buf;
//判斷是否有unsafe對(duì)象
if (PlatformDependent.hasUnsafe()) {
buf = noCleaner ? new InstrumentedUnpooledUnsafeNoCleanerDirectByteBuf(this, initialCapacity, maxCapacity) :
new InstrumentedUnpooledUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);
} else {
buf = new InstrumentedUnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
}
return disableLeakDetector ? buf : toLeakAwareBuffer(buf);
}
- 跟蹤
buf = new InstrumentedUnpooledDirectByteBuf(this, initialCapacity, maxCapacity);可以發(fā)現(xiàn)问顷，Unpooled、Direct類型得內(nèi)存分配實(shí)際上是維護(hù)了一個(gè)底層jdk的一個(gè)DirectByteBuffer禀梳。分配內(nèi)存的時(shí)候就創(chuàng)建它杜窄，并將他保存到buffer成員變量。
InstrumentedUnpooledDirectByteBuf(
UnpooledByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
super(alloc, initialCapacity, maxCapacity);
}
public UnpooledDirectByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
super(maxCapacity);
if (alloc == null) {
throw new NullPointerException("alloc");
}
//檢查合法性
checkPositiveOrZero(initialCapacity, "initialCapacity");
checkPositiveOrZero(maxCapacity, "maxCapacity");
if (initialCapacity > maxCapacity) {
throw new IllegalArgumentException(String.format(
"initialCapacity(%d) > maxCapacity(%d)", initialCapacity, maxCapacity));
}
this.alloc = alloc;
//獲取jdkDirectBuffer并保存到成員變量
setByteBuffer(allocateDirect(initialCapacity));
}
private void setByteBuffer(ByteBuffer buffer) {
ByteBuffer oldBuffer = this.buffer;
//釋放舊的buffer
if (oldBuffer != null) {
if (doNotFree) {
doNotFree = false;
} else {
freeDirect(oldBuffer);
}
}
//保存新buffer
this.buffer = buffer;
tmpNioBuf = null;
capacity = buffer.remaining();
}
protected ByteBuffer allocateDirect(int initialCapacity) {
//分配
return ByteBuffer.allocateDirect(initialCapacity);
}
public static ByteBuffer allocateDirect(int capacity) {
return new DirectByteBuffer(capacity);
}
跟蹤iUnpooledHeapByteBuf#_getByte()算途，就比較簡(jiǎn)單了塞耕，直接使用jdk的api獲取。
@Override
protected byte _getByte(int index) {
//使用buffer
return buffer.get(index);
}
- 跟蹤
new InstrumentedUnpooledUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);可以發(fā)現(xiàn)Unpooled郊艘、Unsafe荷科、Direct的內(nèi)存分配唯咬，和非Unsafe的區(qū)別在于它計(jì)算了一個(gè)內(nèi)存首地址并且保存起來(lái)纱注，在計(jì)算內(nèi)存首地址的時(shí)候是通過(guò)UNSAFE對(duì)象去獲取的。保存內(nèi)存首地址的好處是可以在獲取的時(shí)候直接通過(guò)計(jì)算下標(biāo)直接獲取胆胰。
InstrumentedUnpooledUnsafeDirectByteBuf(
UnpooledByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
super(alloc, initialCapacity, maxCapacity);
}
public UnpooledUnsafeDirectByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
super(maxCapacity);
if (alloc == null) {
throw new NullPointerException("alloc");
}
checkPositiveOrZero(initialCapacity, "initialCapacity");
checkPositiveOrZero(maxCapacity, "maxCapacity");
if (initialCapacity > maxCapacity) {
throw new IllegalArgumentException(String.format(
"initialCapacity(%d) > maxCapacity(%d)", initialCapacity, maxCapacity));
}
this.alloc = alloc;
//分配jdk底層DirectByteBuffer狞贱，設(shè)置buffer
setByteBuffer(allocateDirect(initialCapacity), false);
}
final void setByteBuffer(ByteBuffer buffer, boolean tryFree) {
if (tryFree) {
ByteBuffer oldBuffer = this.buffer;
if (oldBuffer != null) {
if (doNotFree) {
doNotFree = false;
} else {
freeDirect(oldBuffer);
}
}
}
this.buffer = buffer;
//計(jì)算內(nèi)存地址
memoryAddress = PlatformDependent.directBufferAddress(buffer);
tmpNioBuf = null;
capacity = buffer.remaining();
}
public static long directBufferAddress(ByteBuffer buffer) {
return PlatformDependent0.directBufferAddress(buffer);
}
static long directBufferAddress(ByteBuffer buffer) {
return getLong(buffer, ADDRESS_FIELD_OFFSET);
}
private static long getLong(Object object, long fieldOffset) {
//調(diào)用UNSAFE獲取內(nèi)存地址
return UNSAFE.getLong(object, fieldOffset);
}
跟蹤UnpooledUnsafeDirectByteBuf#_getByte()，可以知道UNSAFE的直接內(nèi)存內(nèi)容獲取方式是通過(guò)內(nèi)存首地址 + 偏移量獲取的蜀涨。
@Override
protected byte _getByte(int index) {
//通過(guò)計(jì)算地址獲取
return UnsafeByteBufUtil.getByte(addr(index));
}
long addr(int index) {
//直接從 內(nèi)存首地址 + 偏移量 獲取
return memoryAddress + index;
}
static byte getByte(long address) {
//address正是memoryAddress + index
return UNSAFE.getByte(address);
}
