7.netty內存管理-ByteBuf

<!-- TOC -->

• ByteBuf
  • ByteBuf是什麼
  • ByteBuf重要API
    • read、write、set、skipBytes
    • mark和reset
    • duplicate、slice、copy
    • retain、release
    • ByteBuf擴容
  • ByteBuf種類
• ByteBufAllocate
  • UnPooledByteBufAllocate
    • newHeapBuffer
    • newDirectBuffer

<!-- /TOC -->

ByteBuf

ByteBuf是什麼

爲了平衡數據傳輸時CPU與各類IO設備速度的差別性，計算機設計者引入了緩衝區這一重要抽象。jdkNIO庫提供了java.nio.Buffer接口，而且提供了7種默認實現，常見的實現類爲ByteBuffer。不過netty並無直接使用nio的ByteBuffer，這主要是因爲jdk的Buffer有如下幾個缺點：

1. 當調用allocate方法分配內存時，Buffer的長度就固定了，不能動態擴展和收縮，當寫入數據大於緩衝區的capacity時會發生數組越界錯誤
2. Buffer只有一個位置標誌位屬性position，讀寫切換時，必須先調用flip或rewind方法。不只如此，由於flip的切換
3. Buffer只提供了存取、翻轉、釋放、標誌、比較、批量移動等緩衝區的基本操做，想使用高級的功能(好比池化),就得本身手動進行封裝及維護，使用很是不方便。 也所以，netty實現了本身的緩衝區——ByteBuf,連名字都如此類似。那麼ByteBuf是如何規避ByteBuffer的缺點的？ 第一點顯然是很好解決的,因爲ByteBuf底層也是數組，那麼它就能夠像ArrayList同樣，在寫入操做時進行容量檢查，當容量不足時進行擴容。 第二點，ByteBuf經過2個索引readerIndex，writerIndex將數組分爲3部分，以下圖所示

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

初始化時，readerIndex和writerIndex都是0，隨着數據的寫入writerIndex會增長，此時readable byte部分增長，writable bytes減小。當讀取時，discardable bytes增長，readable bytes減小。因爲讀操做只修改readerIndex，寫操做只修改writerIndex，讓ByteBuf的使用更加容易理解，避免了因爲遺漏flip致使的功能異常。 此外，當調用discardReadBytes方法時，能夠把discardable bytes這部分的內存釋放。整體想法是經過將readerIndex移動到0，writerIndex移動到writerIndex-readerIndex下標，具體移動下標的方式依據ByteBuf實現類有所不一樣。這個方法能夠顯著提升緩衝區的空間複用率，避免無限度的擴容，但會發生字節數組的內存複製，屬於以時間換空間的作法。

ByteBuf重要API

read、write、set、skipBytes

前3個系列的方法及最後一個skipBytes都屬於改變指針的方法。舉例來講，readByte會移動readerIndex1個下標位，而int是4個byte的大小，因此readInt會移動readerIndex4個下標位，相應的，writeByte會移動writerIndex1個下標位，writeInt會移動writerIndex4個下標位。set系列方法比較特殊，它的參數爲index和value，意即將value寫入指定的index位置，但這個操做不會改變readerIndex和writerIndex。skipBytes比較簡單粗暴，直接將readerIndex移動指定長度。

mark和reset

markReaderIndex和markWriterIndex能夠將對應的指針作一個標記，當須要從新操做這部分數據時，再使用resetReaderIndex或resetWriterIndex，將對應指針復位到mark的位置。

duplicate、slice、copy

這3種方法均可以複製一份字節數組，不一樣之處在於duplicate和slice兩個方法返回的新ByteBuf和原有的老ByteBuf之間的內容會互相影響，而copy則不會。duplicate和slice的區別在於前者複製整個ByteBuf的字節數組，然後者默認僅複製可讀部分，但能夠經過slice(index, length)分割指定的區間。

retain、release

這是ByteBuf接口繼承自ReferenceCounted接口的方法，用於引用計數，以便在不使用對象時及時釋放。實現思路是當須要使用一個對象時，計數加1；再也不使用時，計數減1。考慮到多線程場景，通常也多采用AtomicInteger實現。netty卻另闢蹊徑，選擇了volatile + AtomicIntegerFieldUpdater這樣一種更節省內存的方式。

ByteBuf擴容

在ByteBuf寫入數據時會檢查可寫入的容量，若容量不足會進行擴容。

final void ensureWritable0(int minWritableBytes) {
    if (minWritableBytes <= writableBytes()) {
        return;
    }
    int minNewCapacity = writerIndex + minWritableBytes;
    int newCapacity = alloc().calculateNewCapacity(minNewCapacity, maxCapacity);
    int fastCapacity = writerIndex + maxFastWritableBytes();
    if (newCapacity > fastCapacity && minNewCapacity <= fastCapacity) {
        newCapacity = fastCapacity;
    }
    capacity(newCapacity);
}

忽略一些檢驗性質的代碼後，能夠看到擴容時先嚐試將現有寫索引加上須要寫入的容量大小做爲最小新容量，並調用ByteBufAllocate的calculateNewCapacity方法進行計算。跟入這個方法：

public int calculateNewCapacity(int minNewCapacity, int maxCapacity) {
    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;
    }
    int newCapacity = 64;
    while (newCapacity < minNewCapacity) {
        newCapacity <<= 1;
    }
    return Math.min(newCapacity, maxCapacity);
}

能夠看到這個方法的目的則是計算比可寫容量稍大的2的冪次方。minNewCapacity由上一個方法傳入，而maxCapacity則爲Integer.MAX_VALUE。具體步驟是首先判斷新容量minNewCapacity是否超過了計算限制CALCULATE_THRESHOLD，默認爲4M，若是沒有超過4MB，那麼從64B開始不斷以2的冪次方形式擴容，直到newCapacity超過minNewCapacity。而若一開始新容量就超過了4M，則調整新容量到4M的倍數+1。好比newCapacity爲6M,由於6/4 = 1，因此調整爲(1+1)*4M=8M。

在計算完容量以後會調用capacity方法。這是一個抽象方法，這裏以UnpooledHeapByteBuf爲例。

public ByteBuf capacity(int newCapacity) {
    checkNewCapacity(newCapacity);
    byte[] oldArray = array;
    int oldCapacity = oldArray.length;
    if (newCapacity == oldCapacity) {
        return this;
    }
    int bytesToCopy;
    if (newCapacity > oldCapacity) {
        bytesToCopy = oldCapacity;
    } else {
        trimIndicesToCapacity(newCapacity);
        bytesToCopy = newCapacity;
    }
    byte[] newArray = allocateArray(newCapacity);
    System.arraycopy(oldArray, 0, newArray, 0, bytesToCopy);
    setArray(newArray);
    freeArray(oldArray);
    return this;
}

首先檢查newCapacity是否大於0且小於最大容量。以後準備好老數組要複製的長度。trimIndicesToCapacity(newCapacity)是縮容時調用的，它將readerIndex和newCapacity的較小值設置爲新的readerIndex，將newCapacity設置爲新的writerIndex。 以後便分配一個新數組，並開始複製舊數組的元素。複製成功後，將新數組保存爲成員變量，將老數組釋放掉。

ByteBuf種類

出於性能和空間的多方考慮，netty從3個維度定義了各類不一樣的ByteBuf實現類，主要是池化、堆內堆外、能否使用Unsafe類這3個維度，從而演化出8種不一樣的ByteBuf，它們分別是PooledUnsafeHeapBytebuf、PooledHeapByteBuf、PooledUnsafeDirectByteBuf、PooledDirectBytebuf、UnpooledUnsafeHeapByteBuf、UnpooledHeapByteBuf、UnpooledUnsafeDirectByteBuf、UnpooledDirectByteBuf。 ByteBuf接口之下有一個抽象類AbstractByteBuf，實現了接口定義的read、write、set相關的方法，但在實現時只作了檢查，而具體邏輯則定義一系列以_開頭的proteced方法，留待子類實現。

ByteBufAllocate

不一樣於通常形式的建立對象，ByteBuf須要經過內存分配器ByteBufAllocate分配，對應於不一樣的ByteBuf也會有不一樣的BtteBufferAllocate。netty將之抽象爲ByteBufAllocate接口。咱們看一下有哪些方法:

1. buffer()、buffer(initialCapacity)、buffer(initialCapacity、maxCapacity)，分配ByteBuf的方法，具體分配的Buffer是堆內仍是堆外則由實現類決定。2個重載方法分別以給定初始容量、最大容量的方式分配內存
2. ioBuffer()、ioBuffer(initialCapacity)、ioBuffer(initialCapacity、maxCapacity)更傾向於分配堆外內存的方法，由於堆外內存更適合用於IO操做。重載方法同上
3. heapBuffer()、heapBuffer(initialCapacity)、heapBuffer(initialCapacity、maxCapacity)分配堆內內存的方法。
4. directBuffer()、directBuffer(initialCapacity)、directBuffer(initialCapacity、maxCapacity)分配堆外內存的方法。
5. compositeBuffer()。能夠將多個ByteBuf合併爲一個ByteBuf，多個ByteBuf能夠部分是堆內內存，部分是堆外內存。 ByteBufAllocate接口定義了heap和direct這一個維度，其餘維度則交由子類來定義。

UnPooledByteBufAllocate

ByteBufAllocate有一個直接實現類AbstractByteBufAllocate，它實現了大部分方法，只留下2個抽象方法newHeapBuffer和newDirectBuffer交由子類實現。AbstractByteBufAllocate有2個子類PooledByteBufAllocate和UnpooledByteBufAllocate，在這裏定義了pooled池化維度的分配方式。 看看UnpooledByteBufAllocate如何實現2個抽象方法：

newHeapBuffer

protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
    return PlatformDependent.hasUnsafe() ?
            new UnpooledUnsafeHeapByteBuf(this, initialCapacity, maxCapacity) :
            new UnpooledHeapByteBuf(this, initialCapacity, maxCapacity);
}

能夠看到實現類根據PlatformDependent.hasUnsafe()方法自動斷定是否使用unsafe維度，這個方法經過在靜態代碼塊中嘗試初始化sun.misc.Unsafe來判斷Unsafe類是否在當前平臺可用，在juc中，這個類使用頗多，做爲與高併發打交道的netty，出現這個類不使人意外。UnpooledUnsafeHeapByteBuf與UnpooledHeapByteBuf並非平級關係，事實上前者繼承了後者，在構造方法上也直接調用UnpooledHeapByteBuf的構造方法。構造方法比較簡單，初始化byte數組、初始容量、最大容量，將讀寫指針的設置爲0，並將子類傳入的this指針保存到alloc變量中。 兩種Bytebuf的區別在於unsafe會嘗試經過反射的方式建立byte數組，並將數組的地址保存起來，以後再獲取數據時也會調用Unsafe的getByte方法，經過數組在內存中的地址+偏移量的形式直接獲取，而普通的SafeByteBuf則是保存byte數組，經過數組索引即array[index]訪問。

// UnsafeHeapByteBuf初始化數組
protected byte[] allocateArray(int initialCapacity) {
    return PlatformDependent.allocateUninitializedArray(initialCapacity);
}
// HeapByteBuf初始化數組
protected byte[] allocateArray(int initialCapacity) {
    return new byte[initialCapacity];
}
// UnsafeHeapByteBuf經過UnsafeByteBufUtil獲取字節
static byte getByte(byte[] data, int index) {
    return UNSAFE.getByte(data, BYTE_ARRAY_BASE_OFFSET + index);
}
// HeapByteBuf獲取字節
static byte getByte(byte[] memory, int index) {
    return memory[index];
}

newDirectBuffer

protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
    return PlatformDependent.hasUnsafe() ?
            new UnpooledUnsafeDirectByteBuf(this, initialCapacity, maxCapacity) :
                    new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
}

DirectByteBuf構造方法大體與heap的相似，只是保存數據的容器由字節數組變爲了jdk的ByteBuffer。相應的，分配與釋放內存的方法也變成調用jdk的ByteBuffer方法。而UnsafeByteBuf更是直接用long類型記錄內存地址。

// DirectByteBuf獲取字節
protected byte _getByte(int index) {
    return buffer.get(index);
}
// UnsafeDirectByteBuf獲取字節
protected byte _getByte(int index) {
    return UnsafeByteBufUtil.getByte(addr(index));
}
// 獲取內存地址
final long addr(int index) {
    return memoryAddress + index;
}
// UnsafeByteBufUtil獲取字節
static byte getByte(long address) {
    return UNSAFE.getByte(address);
}

因爲PooledByteBufAllocate內容較爲龐大，放入下一節講述。 未完待續···

原文出處：https://www.cnblogs.com/spiritsx/p/12158853.html