【源起Netty 正傳】升級版卡車——ByteBuf

時間 2019-12-07

原文原文鏈接

卡車

卡車指的是java原生類ByteBuffer，這兄弟在NIO界大名鼎鼎，與Channel、Selector的鐵三角組合構築了NIO的核心。之因此稱它爲卡車，只因《編程思想》中有段比喻：java

咱們能夠把它想象成一個煤礦，通道（Channel）是一個包含煤層（數據）的礦藏，而緩衝器（ByteBuffer）則是派送到礦藏中的卡車。卡車滿載煤炭而歸，咱們再從卡車上得到煤炭。也就是說，咱們並無直接和通道交互；咱們只是和緩衝器交互，並把緩衝器派送到通道。

那麼升級版卡車，天然指的就是ByteBuf。數據庫

結構和功能

Netty之因此再次打造了升級版的緩衝器，顯然是不滿ByteBuffer中的某些弊端。編程

ByteBuffer長度固定
使用者常常須要調用flip()、rewind()方法調整position的位置，不方便
API功能有限

ByteBuffer中有三個重要的位置屬性：position、limit、capacity，一個寫操做以後大概是這樣的緩存

如若想進行讀操做，那麼flip()的調用是少不了的，從圖中不難看出，目前position到limit啥也沒有。
調用flip()以後則不同了（咱們不同~）：網絡

而ByteBuf的人設則不相同，它的兩個位置屬性readIndex、writeIndex，分別和讀操做、寫操做相對應。「寫」不操做readIndex，「讀」不操做writeIndex，二者不會相互干擾。這裏盜幾張圖說明下好了：框架

初始狀態

寫入N個字節

讀取M個（M<N）字節

釋放已讀緩存discardReadBytes

重點在於ByteBuf的read和write相關方法，已經封裝好了對readIndex、writeIndex位置索引的操做，不須要使用者繁瑣的flip()。且write()方法中，ByteBuf設計了自動擴容，這一點後續章節會進行詳細說明。ide

功能方面，主要關注兩點：函數

Derived buffers，相似於數據庫視圖。ByteBuf提供了多個接口用於建立某ByteBuf的視圖或複製ByteBuf：性能
- duplicate：返回當前ByteBuf的複製對象，緩衝區內容共享（修改複製的ByteBuf，原來的ByteBuf內容也隨之改變），索引獨立維護。
- copy：內容和索引都獨立。
- slice：返回當前ByteBuf的可讀子緩衝區，內容共享，索引獨立。
轉換成ByteBuffer
nio的SocketChanel進行網絡操做，仍是操做的java原生的ByteBuffer，因此ByteBuf轉換成ByteBuffer的需求仍是有市場的。this
- ByteBuffer nioBuffer()：當前ByteBuf的可讀緩衝區轉換成ByteBuffer，緩衝區內容共享，索引獨立。須要指出的是，返回後的ByteBuffer沒法感知原ByteBuf的動態擴展操做。

ByteBuf星系

稱之爲「星系」，是由於ByteBuf一脈涉及到的類實在太多了，但多而不亂，歸功於類關係結構的設計。

類關係結構

依然盜圖：

從內存分配角度，ByteBuf可分爲兩類

堆內存HeapByteBuf字節緩衝區
直接內存DirectByteBuf字節緩衝區

從內存回收角度，ByteBuf也可分爲兩類：

普通緩衝區UnpooledByteBuf
池化緩衝區PooledByteBuf

縱觀該關繼承節構，給我留下的印象就是每層各司其職：讀操做以及其它的一些公共功能由父類實現，差別化功能由子類實現。

下面聊下筆者感興趣的幾個點……

AbstractByteBuf的寫操做簇

AbstractByteBuf的寫操做有不少，這裏以writeBytes(byte[] src, int srcIndex, int length)方法爲例

@Override
public ByteBuf writeBytes(byte[] src, int srcIndex, int length) {
    ensureWritable(length);    //1、確保可寫，對邊界進行驗證
    setBytes(writerIndex, src, srcIndex, length);    //2、寫入操做，不一樣類型的子類實現方式不一樣
    writerIndex += length;
    return this;
}

註釋部分分別展開看下。

註釋1、確保可寫，對邊界進行驗證

跟調用棧ensureWritable -> ensureWritable0，觀察ensureWritable0方法

final void ensureWritable0(int minWritableBytes) {
    ensureAccessible();    //確保對象可用
    if (minWritableBytes <= writableBytes()) {
        return;
    }

    if (minWritableBytes > maxCapacity - writerIndex) {
        throw new IndexOutOfBoundsException(String.format(
                "writerIndex(%d) + minWritableBytes(%d) exceeds maxCapacity(%d): %s",
                writerIndex, minWritableBytes, maxCapacity, this));
    }

    // Normalize the current capacity to the power of 2.
    // 3、計算擴容量
    int newCapacity = alloc().calculateNewCapacity(writerIndex + minWritableBytes, maxCapacity);

    // Adjust to the new capacity.
    capacity(newCapacity);    //4、內存分配
}

比較

先對要寫入的字節數minWritableBytes進行判斷：若是minWritableBytes < capacity - writeIndex，那麼很好，不須要擴容；若是minWritableBytes > maxCapacity - writerIndex，也就是要寫入字節數超過了容許的最大字節數，直接拋出越界異常IndexOutOfBoundsException。

眼尖的朋友可能發現了，兩次用來判斷的上界並不相同——capacity / maxCapacity。maxCapacity是AbstractByteBuf的屬性，而capacity設定在其子類中。簡單看下UnpooledDirectByteBuf的構造函數:

public UnpooledDirectByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
    super(maxCapacity);    //爲AbstractByteBuf的maxCapacity屬性賦值
    
    /**
     *    ……
     *    省略無關部分
     */
     
    setByteBuffer(ByteBuffer.allocateDirect(initialCapacity));    //capacity賦值
}

也就是說，ByteBuf的結構，可當作這樣：

擴容計算

@Override
public int calculateNewCapacity(int minNewCapacity, int maxCapacity) {
    if (minNewCapacity < 0) {
        throw new IllegalArgumentException("minNewCapacity: " + minNewCapacity + " (expected: 0+)");
    }
    if (minNewCapacity > maxCapacity) {
        throw new IllegalArgumentException(String.format(
                "minNewCapacity: %d (expected: not greater than maxCapacity(%d)",
                minNewCapacity, maxCapacity));
    }

    /** 
     *  設置閥值爲4MB
     *  1.若是擴展的容量大於閥值，對擴張後的內存和最大內存進行比較：大於最大長度使用最大長度，不然步進4M
     *  2.若是須要擴展的容量小於閥值，以64進行計數倍增:64->128->256；爲防止倍增過猛，最後與最大值再次進行比較
     */
    final int threshold = CALCULATE_THRESHOLD; // 4 MiB page

    if (minNewCapacity == threshold) {
        return threshold;
    }

    // If over threshold, do not double but just increase by threshold.
    if (minNewCapacity > threshold) {
        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;
    while (newCapacity < minNewCapacity) {
        newCapacity <<= 1;
    }

    return Math.min(newCapacity, maxCapacity);
}

具體的擴容策略，已拍入註釋中，儘可查看！

註釋2、寫入操做，不一樣類型的子類實現方式不一樣

對比下UnpooledDirectByteBuf和UnpooledHeapByteBuf的實現

UnpooledDirectByteBuf

@Override
public ByteBuf setBytes(int index, byte[] src, int srcIndex, int length) {
    checkSrcIndex(index, length, srcIndex, src.length);
    ByteBuffer tmpBuf = internalNioBuffer();    //分配
    tmpBuf.clear().position(index).limit(index + length);
    tmpBuf.put(src, srcIndex, length);
    return this;
}

UnpooledHeapByteBuf

@Override
public ByteBuf setBytes(int index, byte[] src, int srcIndex, int length) {
    checkSrcIndex(index, length, srcIndex, src.length);
    System.arraycopy(src, srcIndex, array, index, length); //分配
    return this;
}

篇幅有限，不展開說了，結論就是：
UnpooledDirectByteBuf的底層實現爲ByteBuffer.allocateDirect，分配時複製體經過buffer.duplicate()獲取複製體；而UnpooledHeapByteBuf的底層實現爲byte[]，分配時經過System.arraycopy方法拷貝副本。

AbstractReferenceCountedByteBuf

AbstractReferenceCountedByteBuf的名字就挺有意思——「引用計數」，一副JVM垃圾回收的即視感。而事實上，也差很少一個意思。

看下類屬性：

private static final AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> refCntUpdater =
          AtomicIntegerFieldUpdater.newUpdater(AbstractReferenceCountedByteBuf.class, "refCnt");

private volatile int refCnt;

以原子方式更新屬性的AtomicIntegerFieldUpdater起了關鍵做用，將會對volatile修飾的refCnt進行更新，見retain方法（下面展現的是retain的關鍵部分retain0）：

private ByteBuf retain0(final int increment) {
    int oldRef = refCntUpdater.getAndAdd(this, increment);
    if (oldRef <= 0 || oldRef + increment < oldRef) {
        // Ensure we don't resurrect (which means the refCnt was 0) and also that we encountered an overflow.
        refCntUpdater.getAndAdd(this, -increment);
        throw new IllegalReferenceCountException(oldRef, increment);
    }
    return this;
}

源碼閱讀頗有意思的一點就是能看到些本身不熟悉的類，好比AtomicIntegerFieldUpdater我之前就沒接觸過！

內存池

內存池可有效的提高效率，道理和線程池、數據庫鏈接池相通，即省去了重複建立銷燬的過程。

到目前爲止，看到的都是ByteBuf中的各Unpooled實現，而池化版的ByteBuf沒怎麼提過。爲什麼如此？由於池化的實現較複雜，以我如今的功力尚不能徹底掌握透徹。

先聊下內存池的設計思路，漲漲姿式：
爲了集中集中管理內存的分配和釋放，同事提升分配和釋放內存時候的性能，不少框架和應用都會經過預先申請一大塊內存，而後經過提供相應的分配和釋放接口來使用內存。這樣一來，堆內存的管理就被集中到幾個類或函數中，因爲再也不頻繁使用系統調用來申請和釋放內存，應用或系統的性能也會大大提升。 ——節選自《Netty權威指南》

Netty的ByteBuf內存池也是按照這個思路搞的。首先，看下官方註釋：

/**
 * Notation: The following terms are important to understand the code
 * > page  - a page is the smallest unit of memory chunk that can be allocated
 * > chunk - a chunk is a collection of pages
 * > in this code chunkSize = 2^{maxOrder} * pageSize
 */

這裏面有兩個重要的概念page（頁）和chunk（塊），chunk管理多個page組成二叉樹結構，大概就是這個樣子：

選擇二叉樹是有緣由的:

/**
 * To search for the first offset in chunk that has at least requested size available we construct a
 * complete balanced binary tree and store it in an array (just like heaps) - memoryMap
 */

爲了在chunk中找到至少可用的size的偏移量offset。
繼線性結構後，人們又發明了樹形結構的意義在於「提高查詢效率」，也一樣是這裏選擇二叉樹的緣由。

小於一個page的內存，直接在PoolSubpage中分配完成。

某塊內存是否分配，將經過狀態位進行標識。