Netty學習之旅----ByteBuf原始碼解讀之初探UnpooledHeapByteBuf、UnpooledDirectByteBuf
阿新 • • 發佈：2019-01-06
前沿：
在讀原始碼的過程中我發現有如下幾個點，感覺有問題：
1）UnpooledDirectByteBuf的 capacity(int newCapacity)方法，在readerIndex大於或等於newCapacity時，此時不需要將原ByteBuffer中的資料寫入到新的ByteBuffer中，這可以理解，但為什麼要呼叫setIndex(newCapacity,newCapacity)將readerIndex,writerIndex設定為capacity呢？設定之後，該ByteBuf不能讀也不能寫。
2）setByteBuffer 方法，在第一次oldBuffer不為空的時候，此時doNotFree為true(在構造方法中設定)，為什麼第一次oldBuffer不為空的時候，不釋放掉該部分記憶體，不明白。 
目前，先關注UnpooledByteBufAllocator，物件池的ByteBuf在後續章節中重點關注。 結合原始碼，有如下兩個方法引起了我的注意： 1、容量擴容規則（容量增長規則）calculateNewCapacity方法 2、直接記憶體的分配。newDirectBuffer方法 2.3.2.1 calculateNewCapacity
1、首先，我們先看一下ByteBuf的類設計圖，從中更進一步瞭解ByteBuf。
package io.netty.buffer;
import io.netty.util.IllegalReferenceCountException;
import io.netty.util.internal.PlatformDependent;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
/**
* Abstract base class for {@link ByteBuf} implementations that count references.
*/
public abstract class AbstractReferenceCountedByteBuf extends AbstractByteBuf {
private static final AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> refCntUpdater;
static {
AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> updater =
PlatformDependent.newAtomicIntegerFieldUpdater(AbstractReferenceCountedByteBuf.class, "refCnt");
if (updater == null) {
updater = AtomicIntegerFieldUpdater.newUpdater(AbstractReferenceCountedByteBuf.class, "refCnt");
}
refCntUpdater = updater;
}
private volatile int refCnt = 1;
protected AbstractReferenceCountedByteBuf(int maxCapacity) {
super(maxCapacity);
}
@Override
public final int refCnt() {
return refCnt;
}
/**
* An unsafe operation intended for use by a subclass that sets the reference count of the buffer directly
*/
protected final void setRefCnt(int refCnt) {
this.refCnt = refCnt;
}
@Override
public ByteBuf retain() {
for (;;) {
int refCnt = this.refCnt;
if (refCnt == 0) {
throw new IllegalReferenceCountException(0, 1);
}
if (refCnt == Integer.MAX_VALUE) {
throw new IllegalReferenceCountException(Integer.MAX_VALUE, 1);
}
if (refCntUpdater.compareAndSet(this, refCnt, refCnt + 1)) {
break;
}
}
return this;
}
@Override
public ByteBuf retain(int increment) {
if (increment <= 0) {
throw new IllegalArgumentException("increment: " + increment + " (expected: > 0)");
}
for (;;) {
int refCnt = this.refCnt;
if (refCnt == 0) {
throw new IllegalReferenceCountException(0, increment);
}
if (refCnt > Integer.MAX_VALUE - increment) {
throw new IllegalReferenceCountException(refCnt, increment);
}
if (refCntUpdater.compareAndSet(this, refCnt, refCnt + increment)) {
break;
}
}
return this;
}
@Override
public ByteBuf touch() {
return this;
}
@Override
public ByteBuf touch(Object hint) {
return this;
}
@Override
public final boolean release() {
for (;;) {
int refCnt = this.refCnt;
if (refCnt == 0) {
throw new IllegalReferenceCountException(0, -1);
}
if (refCntUpdater.compareAndSet(this, refCnt, refCnt - 1)) {
if (refCnt == 1) {
deallocate();
return true;
}
return false;
}
}
}
@Override
public final boolean release(int decrement) {
if (decrement <= 0) {
throw new IllegalArgumentException("decrement: " + decrement + " (expected: > 0)");
}
for (;;) {
int refCnt = this.refCnt;
if (refCnt < decrement) {
throw new IllegalReferenceCountException(refCnt, -decrement);
}
if (refCntUpdater.compareAndSet(this, refCnt, refCnt - decrement)) {
if (refCnt == decrement) {
deallocate();
return true;
}
return false;
}
}
}
/**
* Called once {@link #refCnt()} is equals 0.
*/
protected abstract void deallocate();
}
public ByteBuf capacity(int newCapacity) {
ensureAccessible(); // @1
if (newCapacity < 0 || newCapacity > maxCapacity()) {
throw new IllegalArgumentException("newCapacity: " + newCapacity);
}
int readerIndex = readerIndex();
int writerIndex = writerIndex();
int oldCapacity = capacity;
if (newCapacity > oldCapacity) { // @2
ByteBuffer oldBuffer = buffer;
ByteBuffer newBuffer = allocateDirect(newCapacity); //@21
oldBuffer.position(0).limit(oldBuffer.capacity()); //@22
newBuffer.position(0).limit(oldBuffer.capacity()); //@23
newBuffer.put(oldBuffer); //@24
newBuffer.clear(); //@25
setByteBuffer(newBuffer); //@26
} else if (newCapacity < oldCapacity) { //@3
ByteBuffer oldBuffer = buffer;
ByteBuffer newBuffer = allocateDirect(newCapacity);
if (readerIndex < newCapacity) {
if (writerIndex > newCapacity) {
writerIndex(writerIndex = newCapacity);
}
oldBuffer.position(readerIndex).limit(writerIndex);
newBuffer.position(readerIndex).limit(writerIndex);
newBuffer.put(oldBuffer);
newBuffer.clear();
} else {
setIndex(newCapacity, newCapacity);
}
setByteBuffer(newBuffer);
}
return this;
}private void setByteBuffer(ByteBuffer buffer) {
ByteBuffer oldBuffer = this.buffer;
if (oldBuffer != null) {
if (doNotFree) {
doNotFree = false;
} else {
freeDirect(oldBuffer);
}
}
this.buffer = buffer;
tmpNioBuf = null;
capacity = buffer.remaining();
}目前，先關注UnpooledByteBufAllocator，物件池的ByteBuf在後續章節中重點關注。 結合原始碼，有如下兩個方法引起了我的注意： 1、容量擴容規則（容量增長規則）calculateNewCapacity方法 2、直接記憶體的分配。newDirectBuffer方法 2.3.2.1 calculateNewCapacity
public int calculateNewCapacity(int minNewCapacity, int maxCapacity) {
if (minNewCapacity < 0) {
throw new IllegalArgumentException("minNewCapacity: " + minNewCapacity + " (expectd: 0+)");
}
if (minNewCapacity > maxCapacity) {
throw new IllegalArgumentException(String.format(
"minNewCapacity: %d (expected: not greater than maxCapacity(%d)",
minNewCapacity, maxCapacity));
}
final int threshold = 1048576 * 4; // 4 MiB page
if (minNewCapacity == threshold) {
return threshold;
}
// If over threshold, do not double but just increase by threshold.
if (minNewCapacity > threshold) { //@1
int newCapacity = minNewCapacity / threshold * threshold;
if (newCapacity > maxCapacity - threshold) {
newCapacity = maxCapacity;
} else {
newCapacity += threshold;
}
return newCapacity;
}
// Not over threshold. Double up to 4 MiB, starting from 64.
int newCapacity = 64;c // @2
while (newCapacity < minNewCapacity) {
newCapacity <<= 1;
}
return Math.min(newCapacity, maxCapacity);
}@Override
protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
ByteBuf buf;
if (PlatformDependent.hasUnsafe()) {
buf = new UnpooledUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);
} else {
buf = new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
}
return toLeakAwareBuffer(buf);
}