上一節(jié)查看了ByteBufAllocator，并了解了其抽象實現(xiàn)，和一些根據(jù)不同的內(nèi)存類型進行內(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來分配
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)存，其實是分配了一個byte數(shù)組，并保存在UnpooledHeapByteBuf#array成員變量中。該內(nèi)存的初始值容量和最大可擴展容量可以指定。
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) {
//初始化了一個新的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)存分配，其實例化過成和InstrumentedUnpooledUnsafeHeapByteBuf一樣。在這里看不出safe和unsafe的區(qū)別，經(jīng)過之前的代碼，可以重獲取的時候getByte方法進入查看.
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獲取的時候。直接通過下標(biāo)獲取byte數(shù)組中的byte
@Override
public byte getByte(int index) {
ensureAccessible();
return _getByte(index);
}
@Override
protected byte _getByte(int index) {
//該array為初始化的時候，實例化的byte[]
return HeapByteBufUtil.getByte(array, index);
}
static byte getByte(byte[] memory, int index) {
//直接拿到一個數(shù)組
return memory[index];
}
- 查看
UnpooledUnsafeHeapByteBuf#getByte()方法，獲取byte字節(jié)的時候，調(diào)用的是jdk的UNSAFE對象。
@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) {
//通過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對象
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)存分配實際上是維護了一個底層jdk的一個DirectByteBuffer。分配內(nèi)存的時候就創(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()，就比較簡單了，直接使用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ū)別在于它計算了一個內(nèi)存首地址并且保存起來，在計算內(nèi)存首地址的時候是通過UNSAFE對象去獲取的。保存內(nèi)存首地址的好處是可以在獲取的時候直接通過計算下標(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;
//計算內(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)容獲取方式是通過內(nèi)存首地址 + 偏移量獲取的。
@Override
protected byte _getByte(int index) {
//通過計算地址獲取
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);
}
