netty源碼解析(4.0)-23 ByteBuf內存管理:分配和釋放

　　ByteBuf內存分配和釋放由具體實現負責，抽象類型只定義的內存分配和釋放的時機。

 

　　內存分配分兩個階段: 第一階段，初始化時分配內存。第二階段: 內存不夠用時分配新的內存。ByteBuf抽象層沒有定義第一階段的行為，但定義了第二階段的方法：

　　public abstract ByteBuf capacity(int newCapacity)

　　這個方法負責分配一個長度為newCapacity的新內存。

 

　　內存釋放的抽象實現在AbstractReferenceCountedByteBuf中實現，這個類實現引用計數，當調用release方法把引用計數變成0時，會調用

　　protected abstract void deallocate()

　　執行真正的內存釋放操作。

 

內存相關的屬性

　　ByteBuf定義了兩個內存相關的屬性:

　　capacity: 當前的當前的容量，也就是當前使用的內存大小。使用capacity()方法獲得。

　　maxCapacity: 最大容量，也就是可以使用的最大內存大小。使用newCapacity()方法獲得。

 

內存分配時機

　　AbstractByteBuf定義了內存分配的時機。當writeXX方法被調用的時候，如果如果發現可寫空間不足，就調用capacity分配新的內存。下面以writeInt為例詳細分析這個過程。

    @Override
    public ByteBuf writeInt(int value) {
        ensureWritable0(4);
        _setInt(writerIndex, value);
        writerIndex += 4;
        return this;
    }    


    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.
        int newCapacity = calculateNewCapacity(writerIndex + minWritableBytes);

        // Adjust to the new capacity.
        capacity(newCapacity);
    }
        

　　3行: ensureWritable確保當前能寫入4Byte的數據。

　　11行: 確保當前ByteBuf是可以訪問的。防止多線程環境下，ByteBuf內存被釋放后讀寫數據。

　　12，13行: 如果內存夠用，就此作罷。

　　16行:  如果需要內存大於maxCapacity拋出異常。

　　23, 26行行: 計算新內存的大小, 調用capacity(int)分配新內存。

　　

　　重新分配內存之前的一個重要步驟的計算新內存的大小。這個工作由calculateNewCapacity方法完成，它的代碼如下:

    private int calculateNewCapacity(int minNewCapacity) {
        final int maxCapacity = this.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) {
            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);
    }

　　1行：接受一個最小的新內存參數minNewCapacity。

　　3行： 定義一個4MB的閾值常量threshold。

　　5，6行: 如果minNewCapacity==threshold，那么新內存大小就是threshold。

　　10-17行: 如果minNewCapacity>threshold, 新內存大小是min(maxCapacity, threshold * n)且threshold * n >= minNewCapacity。

　　21-26行: 如果minNewCapacity<threshold, 新內存大小是min(maxCapacity, 64 * 2ⁿ)且64 * 2ⁿ >= minNewCapacity。

 

內存分配和釋放的具體實現

　　本章涉及到的內存分配和釋放的具體實現只涉及到unpooled類型的ByteBuf，即:

　　UnpooledHeapByteBuf

　　UnpooledDirectByteBuf

　　UnpooledUnsafeHeapByteBuf

　　UnpooledUnsafeDirectByteBuf

　　這幾個具體實現中涉及到的內存分配和釋放代碼比較簡潔，更容易明白ByteBuf內存管理的原理。相比之下，pooled類型的ByteBuf內存分配和釋放的代碼要復雜很多，會在后面的章節獨立分析。

　　

　　UnpooledHeapByteBuf和UnpooledUnsafeHeapByteBuf實現

　　 UnpooledHeapByteBuf中，內存分配的實現代碼主要集中在capacity(int)和allocateArray()方法中。capacity分配新內存的步驟是

• 調用allocateArray分配一塊新內存。
• 把舊內存中的實際復制到新內存中。
• 使用新內存替換舊內存(24行)。
• 釋放掉舊內存(25行)。

　　 代碼如下:

    @Override
    public ByteBuf capacity(int newCapacity) {
        checkNewCapacity(newCapacity);

        int oldCapacity = array.length;
        byte[] oldArray = array;
        if (newCapacity > oldCapacity) {
            byte[] newArray = allocateArray(newCapacity);
            System.arraycopy(oldArray, 0, newArray, 0, oldArray.length);
            setArray(newArray);
            freeArray(oldArray);
        } else if (newCapacity < oldCapacity) {
            byte[] newArray = allocateArray(newCapacity);
            int readerIndex = readerIndex();
            if (readerIndex < newCapacity) {
                int writerIndex = writerIndex();
                if (writerIndex > newCapacity) {
                    writerIndex(writerIndex = newCapacity);
                }
                System.arraycopy(oldArray, readerIndex, newArray, readerIndex, writerIndex - readerIndex);
            } else {
                setIndex(newCapacity, newCapacity);
            }
            setArray(newArray);
            freeArray(oldArray);
        }
        return this;
    }
    

　　capacity中復制舊內存數據到新內存中的時候分兩種情況(newCapacity，oldCapacity分別是新舊內存的大小):

• newCapacity>oldCapacity，這種情況比較簡單，直接復制就好了(第8行)。不影響readerIndex和writerIndex。
• newCapacity<oldCapacity,  這種情況比較復雜。capacity盡量把可讀的數據復制新內存中。
  • 如果readerIndex<newCapacity且writerIndex<newCapacity。可讀數據會全部轉移到新內存中。readerIndex和writerIndex保持不變。
  • 如果readerIndex<newCapacity且writeIndex>newCapacity。可端數據會部分轉移的新內存中，會丟失部分可讀數據。readerIndex不變，writerIndex變成newCapacity。
  • 如果readerIndex>newCapacity，數據全部丟失，readerIndex和writerIndex都會變成newCapacity。

 　　allocateArray方法負責分配一塊新內存，它的實現是new byte[]。freeArray方法負責釋放內存，這個方法是個空方法。

　　UnpooledUnsafeHeapByteBuf是UnloopedHeadpByteBuf的直接子類，在內存管理上的差別是allocateArray的實現，UnpooledUnsafeHeapByteBuf的實現是:　　

    @Override
    byte[] allocateArray(int initialCapacity) {
        return PlatformDependent.allocateUninitializedArray(initialCapacity);
    }

 

　　UnpooledDirectByteBuf和UnpooledUnsafeDirectByteBuf實現

　　UnpooledDirectByteBuf類使用DirectByteBuffer作為內存，使用了DirectByteBuffer的能力來實現ByteBuf接口。allocateDirect和freeDirect方法負責分配和釋放DirectByteBuffer。capacity(int)方法和UnloopedHeapByteBuf類似，使用allocateDirect創建一個新的DirectByteBuffer, 把舊內存數據復制到新內存中，然后使用新內存替換舊內存，最后調用freeDirect方法釋放掉舊的DirectByteBuffer。

    protected ByteBuffer allocateDirect(int initialCapacity) {
        return ByteBuffer.allocateDirect(initialCapacity);
    }

    protected void freeDirect(ByteBuffer buffer) {
        PlatformDependent.freeDirectBuffer(buffer);
    }

　　@Override
    public ByteBuf capacity(int newCapacity) {
        checkNewCapacity(newCapacity);

        int readerIndex = readerIndex();
        int writerIndex = writerIndex();

        int oldCapacity = capacity;
        if (newCapacity > oldCapacity) {
            ByteBuffer oldBuffer = buffer;
            ByteBuffer newBuffer = allocateDirect(newCapacity);
            oldBuffer.position(0).limit(oldBuffer.capacity());
            newBuffer.position(0).limit(oldBuffer.capacity());
            newBuffer.put(oldBuffer);
            newBuffer.clear();
            setByteBuffer(newBuffer);
        } else if (newCapacity < oldCapacity) {
            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;
    }

　　對比UnloopedHeapByteBuf的capacity(int)方法，發現這兩個實現非常類似，也分兩種情況處理:

• 18-24行，newCapacity > oldCapacity的情況。
• 26-39行, newCapacity < oldCapacity的情況。

　　兩種情況對readerIndex和writerIndex的影響也一樣，不同的是數據復制時的具體實現。　　

　　UnpooledUnsafeDirectByteBuf和UnpooledDirectByteBuf同屬於AbstractReferenceCountedByteBuf的派生類，它們之間沒有繼承關系。但內存分配和釋放實現是一樣的，不同的地方是內存I/O。UnpooledUnsafeDirectByteBuf使用UnsafeByteBufUtil類之間讀寫DirectByteBuffer的內存，沒有使用DirectByteBuffer的I/O能力。