netty源碼解析(4.0)-24 ByteBuf基於內存池的內存管理

　io.netty.buffer.PooledByteBuf<T>使用內存池中的一塊內存作為自己的數據內存，這個塊內存是PoolChunk<T>的一部分。PooledByteBuf<T>是一個抽象類型，它有4個派生類:

• PooledHeapByteBuf, PooledUnsafeHeapByteBuf 使用堆內存的PooledByteBuffer<byte[]>。
• PooledDirectByteBuf, PooledUnsafeDirectByteBuf 使用直接內存的PooledByteBuf<ByteBuffer>。

初始化

　　PooledByteBuf的初始化過程分為兩個步驟：創建實例；初始化內存。這兩個步驟的代碼如下：

    protected PooledByteBuf(Recycler.Handle recyclerHandle, int maxCapacity) {
        super(maxCapacity);
        this.recyclerHandle = recyclerHandle;
    }

    void init(PoolChunk<T> chunk, long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
        init0(chunk, handle, offset, length, maxLength, cache);
    }

    private void init0(PoolChunk<T> chunk, long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
        assert handle >= 0;
        assert chunk != null;

        this.chunk = chunk;
        memory = chunk.memory;
        allocator = chunk.arena.parent;
        this.cache = cache;
        this.handle = handle;
        this.offset = offset;
        this.length = length;
        this.maxLength = maxLength;
        tmpNioBuf = null;
    }

　　創建實例時調用的構造方法只是為maxCapacity和recyclerHandler屬性賦值，構造方法是protected，不打算暴露到外面。派生類都提供了newInstance方法創建實例，以PooledHeapByteBuf為例，它的newInstance方法實現如下：

    private static final Recycler<PooledHeapByteBuf> RECYCLER = new Recycler<PooledHeapByteBuf>() {
        @Override
        protected PooledHeapByteBuf newObject(Handle handle) {
            return new PooledHeapByteBuf(handle, 0);
        }
    };

    static PooledHeapByteBuf newInstance(int maxCapacity) {
        PooledHeapByteBuf buf = RECYCLER.get();
        buf.reuse(maxCapacity);
        return buf;
    }

　　這里的newInstance使用RECYCLER創建實例對象。Recycler<T>是一個輕量級的，支持循環使用的對象池。當對象池中沒有可用對象時，會在第4行使用構造方法創建一個新的對象。

 

　　init調用init0初始化數據內存，init0方法為幾個內存相關的關鍵屬性賦值:

• chunk:  PoolChunk<T>對象，這個PooledByteBuf使用的內存就是它的一部分。
• memory: 內存對象。更准確地說，PooledByteBuf使用的內存是它的一部分。
• allocator: 創建這個PooledByteBuf的PooledByteBufAllocator對象。
• cache:  線程專用的內存緩存。分配內存時會優先從這個緩存中尋找合適的內存塊。
• handle:  內存在chunk中node的句柄。chunk使用handle可以計算出它對應內存的起始位置offset。
• offset:  分配內存的起始位置。
• length: 分配內存的長度，也是這個PooledByteBuf的capacity。
• maxLength: 這塊內存node的最大長度。當調用capacity(int newCapacity)方法增加capacity時，只要newCapacity不大於這個值，就不用從新分配內存。

　　內存初始化完成之后，這個PooledByteBuf可使用的內存范圍是memory內存中[offset, offset+length)。idx方法可以把ByteBuf的索引轉換成memory的索引:

    protected final int idx(int index) {
        return offset + index;
    }

 

重新分配內存

　　和前面講過的ByteBuf實現一樣，PooledByteBuf也需要使用capacity(int newCapacity)改變內存大小，也會涉及到把數據從舊內存中復制到新內存的問題。也就是說，要解決的問題是一樣的，只是具體實現的差異。

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

        // If the request capacity does not require reallocation, just update the length of the memory.
        if (chunk.unpooled) {
            if (newCapacity == length) {
                return this;
            }
        } else {
            if (newCapacity > length) {
                if (newCapacity <= maxLength) {
                    length = newCapacity;
                    return this;
                }
            } else if (newCapacity < length) {
                if (newCapacity > maxLength >>> 1) {
                    if (maxLength <= 512) {
                        if (newCapacity > maxLength - 16) {
                            length = newCapacity;
                            setIndex(Math.min(readerIndex(), newCapacity), Math.min(writerIndex(), newCapacity));
                            return this;
                        }
                    } else { // > 512 (i.e. >= 1024)
                        length = newCapacity;
                        setIndex(Math.min(readerIndex(), newCapacity), Math.min(writerIndex(), newCapacity));
                        return this;
                    }
                }
            } else {
                return this;
            }
        }

        // Reallocation required.
        chunk.arena.reallocate(this, newCapacity, true);
        return this;
    }

　　這個方法處理兩大類情況: 不重新分配內存；重新分配內存並復制ByteBuf中的數據和狀態。

　　不重新分配內存: 

　　8行： chunk不需要回收到內存池中，且newCapacity沒有變化。

　　11-32行: chunk需要回收到內存池中。

　　　　13-14行：內存增大，且newcapacity不大於maxLength。把容量修改成newCapacity即可。

　　　　20-22行: 內存減小,  newCapacity 大於maxLength的一半，maxLength<=512， newCapacity >maxLength - 16。 把容量修改成newCapacity, 調整readerIndex, writerIndex。

　　　　25-27行: 內存減小，newCapacity大於maxLength的一半,  maxLength > 512。把容量修改成newCapacity, 調整readerIndex, writerIndex。

　　　　31行: 內存不變，不做任何操作。

　　需要重新分配內存:

　　36行: 任何不滿足以上情況的都要重新分配內存。這里使用Arena的reallocate方法重新分配內存，並把舊內存釋放調，代碼如下:

    //io.netty.buffer.PoolArena#reallocate, 
    void reallocate(PooledByteBuf<T> buf, int newCapacity, boolean freeOldMemory) {
        if (newCapacity < 0 || newCapacity > buf.maxCapacity()) {
            throw new IllegalArgumentException("newCapacity: " + newCapacity);
        }

        int oldCapacity = buf.length;
        if (oldCapacity == newCapacity) {
            return;
        }

        PoolChunk<T> oldChunk = buf.chunk;
        long oldHandle = buf.handle;
        T oldMemory = buf.memory;
        int oldOffset = buf.offset;
        int oldMaxLength = buf.maxLength;
        int readerIndex = buf.readerIndex();
        int writerIndex = buf.writerIndex();

        allocate(parent.threadCache(), buf, newCapacity);
        if (newCapacity > oldCapacity) {
            memoryCopy(
                    oldMemory, oldOffset,
                    buf.memory, buf.offset, oldCapacity);
        } else if (newCapacity < oldCapacity) {
            if (readerIndex < newCapacity) {
                if (writerIndex > newCapacity) {
                    writerIndex = newCapacity;
                }
                memoryCopy(
                        oldMemory, oldOffset + readerIndex,
                        buf.memory, buf.offset + readerIndex, writerIndex - readerIndex);
            } else {
                readerIndex = writerIndex = newCapacity;
            }
        }

        buf.setIndex(readerIndex, writerIndex);

        if (freeOldMemory) {
            free(oldChunk, oldHandle, oldMaxLength, buf.cache);
        }
    }

　　7-9行: 內存大小沒變化，返回。

　　12-18行: 記錄下舊內存的信息，readerIndex, writerIndex。

　　20行: 為PooledByteBuf分配新內存。

　　21-38行: 把舊內存中數據復制到新內存，並把readerIndex，writerIndex設置在正確。

　　41行: 釋放就內存。

 

釋放內存

　　內存釋放代碼在deallocate中:

    @Override
    protected final void deallocate() {
        if (handle >= 0) {
            final long handle = this.handle;
            this.handle = -1;
            memory = null;
            tmpNioBuf = null;
            chunk.arena.free(chunk, handle, maxLength, cache);
            chunk = null;
            recycle();
        }
    }

　　關鍵是第8行代碼，使用PoolArena的free方法釋放內存。然后是recycle把當前PooledByteBuf對象放到RECYCLER中循環使用。

 

PooledByteBufAllocator創建內存管理模塊

　　在前面分析PooledByteBuf內存初始化，重新分配及釋放時，看到了內存管理的三個核心模塊: PoolArena(chunk.arena),  PoolChunk(chunk),  PoolThreadCache(cache)。PooledByteBuf的內存管理能力都是使用這三模塊實現的，它本身沒有實現內存管理算法。當需要為PooledByteBuf分配一塊內存時，先從一個線程專用的PoolThreadCache中得到一個PoolArena,  使用PoolArena的allocate找到一個滿足要求內存塊PoolChunk,  從這個內存塊中分配一塊連續的內存handle，計算出這塊內存起始位置的偏移量offset, 最后調用PooledByteBuf的init方法初始化內存完成內存分配。 釋放內存調用PoolArena的free方法。在內存分配時，會優先從PoolThreadCache中尋找合適的內存塊。在內存釋放時會把內存塊暫時放在PoolThreadCache中，等使用頻率過低時才會還給PoolChunk。這三個模塊中PoolArena,  PoolThreadCache由PooledByteBufAllocator創建，PoolChunk由PoolArean維護。

　　PooledByteBufAllocator維護了相關的幾個屬性:

　　PoolArena<byte[]>[] heapArenas

　　PoolArena<ByteBuffer>[] directArenas

　　PoolThreadLocalCache threadCache

　　headArenas和directArenas分別維護了多個PoolArena, 他們分別用來分配堆內存和直接內存。 如果使用得當，可以讓每個線程持有一個專用的PoolArena,  避免線程間數據同步的開銷。PoolThreadLocalCache會為每個線程創建一個專用的PoolThreadCache實例，這個實例分別持有一個heapArena和directArena。

　　接下來的的幾個章節會詳細分析PoolArena和PoolThreadCache的實現代碼。