Netty分布式ByteBuf使用的回收邏輯剖析

更新時間：2022年03月28日 17:32:48 作者：向南是個萬人迷

這篇文章主要介紹了Netty分布式ByteBuf使用的回收邏輯剖析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪

ByteBuf回收

之前的章節(jié)我們提到過, 堆外內(nèi)存是不受jvm垃圾回收機制控制的, 所以我們分配一塊堆外內(nèi)存進行ByteBuf操作時, 使用完畢要對對象進行回收, 這一小節(jié), 就以PooledUnsafeDirectByteBuf為例講解有關(guān)內(nèi)存分配的相關(guān)邏輯

PooledUnsafeDirectByteBuf中內(nèi)存釋放的入口方法是其父類AbstractReferenceCountedByteBuf中的release方法:

@Override
 public boolean release() {
     return release0(1);
 }

這里調(diào)用了release0, 跟進去

private boolean release0(int decrement) {
    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;
        }
    }
}

if (refCnt == decrement) 中判斷當前byteBuf是否沒有被引用了, 如果沒有被引用, 則通過deallocate()方法進行釋放

因為我們是以PooledUnsafeDirectByteBuf為例, 所以這里會調(diào)用其父類PooledByteBuf的deallocate方法:

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

this.handle = -1表示當前的ByteBuf不再指向任何一塊內(nèi)存

memory = null這里將memory也設(shè)置為null

chunk.arena.free(chunk, handle, maxLength, cache)這一步是將ByteBuf的內(nèi)存進行釋放

recycle()是將對象放入的對象回收站, 循環(huán)利用

我們首先分析free方法

void free(PoolChunk<T> chunk, long handle, int normCapacity, PoolThreadCache cache) {
    //是否為unpooled
    if (chunk.unpooled) {
        int size = chunk.chunkSize();
        destroyChunk(chunk);
        activeBytesHuge.add(-size);
        deallocationsHuge.increment();
    } else {
        //那種級別的Size
        SizeClass sizeClass = sizeClass(normCapacity);
        //加到緩存里
        if (cache != null && cache.add(this, chunk, handle, normCapacity, sizeClass)) {
            return;
        }
        //將緩存對象標記為未使用
        freeChunk(chunk, handle, sizeClass);
    }
}

首先判斷是不是unpooled, 我們這里是Pooled, 所以會走到else塊中:

sizeClass(normCapacity)計算是哪種級別的size, 我們按照tiny級別進行分析

cache.add(this, chunk, handle, normCapacity, sizeClass)是將當前當前ByteBuf進行緩存

我們之前講過, 再分配ByteBuf時首先在緩存上分配, 而這步, 就是將其緩存的過程, 跟進去:

boolean add(PoolArena<?> area, PoolChunk chunk, long handle, int normCapacity, SizeClass sizeClass) {
    //拿到MemoryRegionCache節(jié)點
    MemoryRegionCache<?> cache = cache(area, normCapacity, sizeClass);
    if (cache == null) {
        return false;
    }
    //將chunk, 和handle封裝成實體加到queue里面
    return cache.add(chunk, handle);
}

首先根據(jù)根據(jù)類型拿到相關(guān)類型緩存節(jié)點, 這里會根據(jù)不同的內(nèi)存規(guī)格去找不同的對象, 我們簡單回顧一下, 每個緩存對象都包含一個queue, queue中每個節(jié)點是entry, 每一個entry中包含一個chunk和handle, 可以指向唯一的連續(xù)的內(nèi)存

我們跟到cache中

private MemoryRegionCache<?> cache(PoolArena<?> area, int normCapacity, SizeClass sizeClass) {
    switch (sizeClass) {
    case Normal:
        return cacheForNormal(area, normCapacity);
    case Small:
        return cacheForSmall(area, normCapacity);
    case Tiny:
        return cacheForTiny(area, normCapacity);
    default:
        throw new Error();
    }
}

假設(shè)我們是tiny類型, 這里就會走到cacheForTiny(area, normCapacity)方法中, 跟進去:

private MemoryRegionCache<?> cacheForTiny(PoolArena<?> area, int normCapacity) { 
    int idx = PoolArena.tinyIdx(normCapacity);
    if (area.isDirect()) {
        return cache(tinySubPageDirectCaches, idx);
    }
    return cache(tinySubPageHeapCaches, idx);
}

這個方法我們之前剖析過, 就是根據(jù)大小找到第幾個緩存中的第幾個緩存, 拿到下標之后, 通過cache去超相對應(yīng)的緩存對象:

private static <T>  MemoryRegionCache<T> cache(MemoryRegionCache<T>[] cache, int idx) {
    if (cache == null || idx > cache.length - 1) {
        return null;
    }
    return cache[idx];
}

我們這里看到, 是直接通過下標拿的緩存對象

回到add方法中

boolean add(PoolArena<?> area, PoolChunk chunk, long handle, int normCapacity, SizeClass sizeClass) {
    //拿到MemoryRegionCache節(jié)點
    MemoryRegionCache<?> cache = cache(area, normCapacity, sizeClass);
    if (cache == null) {
        return false;
    }
    //將chunk, 和handle封裝成實體加到queue里面
    return cache.add(chunk, handle);
}

這里的cache對象調(diào)用了一個add方法, 這個方法就是將chunk和handle封裝成一個entry加到queue里面

我們跟到add方法中:

public final boolean add(PoolChunk<T> chunk, long handle) {
    Entry<T> entry = newEntry(chunk, handle); 
    boolean queued = queue.offer(entry);
    if (!queued) {
        entry.recycle();
    }
    return queued;
}

我們之前介紹過, 從在緩存中分配的時候從queue彈出一個entry, 會放到一個對象池里面, 而這里Entry<T> entry = newEntry(chunk, handle)就是從對象池里去取一個entry對象, 然后將chunk和handle進行賦值

然后通過queue.offer(entry)加到queue中

我們回到free方法中

void free(PoolChunk<T> chunk, long handle, int normCapacity, PoolThreadCache cache) {
    //是否為unpooled
    if (chunk.unpooled) {
        int size = chunk.chunkSize();
        destroyChunk(chunk);
        activeBytesHuge.add(-size);
        deallocationsHuge.increment();
    } else {
        //那種級別的Size
        SizeClass sizeClass = sizeClass(normCapacity);
        //加到緩存里
        if (cache != null && cache.add(this, chunk, handle, normCapacity, sizeClass)) {
            return;
        } 
        freeChunk(chunk, handle, sizeClass);
    }
}

這里加到緩存之后, 如果成功, 就會return, 如果不成功, 就會調(diào)用freeChunk(chunk, handle, sizeClass)方法, 這個方法的意義是, 將原先給ByteBuf分配的內(nèi)存區(qū)段標記為未使用

跟進freeChunk簡單分析下:

void freeChunk(PoolChunk<T> chunk, long handle, SizeClass sizeClass) {
    final boolean destroyChunk;
    synchronized (this) {
        switch (sizeClass) {
        case Normal:
            ++deallocationsNormal;
            break;
        case Small:
            ++deallocationsSmall;
            break;
        case Tiny:
            ++deallocationsTiny;
            break;
        default:
            throw new Error();
        }
        destroyChunk = !chunk.parent.free(chunk, handle);
    }
    if (destroyChunk) {
        destroyChunk(chunk);
    }
}

我們再跟到free方法中:

boolean free(PoolChunk<T> chunk, long handle) {
    chunk.free(handle);
    if (chunk.usage() < minUsage) {
        remove(chunk);
        return move0(chunk);
    }
    return true;
}

chunk.free(handle)的意思是通過chunk釋放一段連續(xù)的內(nèi)存

再跟到free方法中:

void free(long handle) {
    int memoryMapIdx = memoryMapIdx(handle);
    int bitmapIdx = bitmapIdx(handle);

    if (bitmapIdx != 0) { 
        PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];
        assert subpage != null && subpage.doNotDestroy;
        PoolSubpage<T> head = arena.findSubpagePoolHead(subpage.elemSize);
        synchronized (head) {
            if (subpage.free(head, bitmapIdx & 0x3FFFFFFF)) {
                return;
            }
        }
    }
    freeBytes += runLength(memoryMapIdx);
    setValue(memoryMapIdx, depth(memoryMapIdx));
    updateParentsFree(memoryMapIdx);
}

if (bitmapIdx != 0)這里判斷是當前緩沖區(qū)分配的級別是Page還是Subpage, 如果是Subpage, 則會找到相關(guān)的Subpage將其位圖標記為0

如果不是subpage, 這里通過分配內(nèi)存的反向標記, 將該內(nèi)存標記為未使用

這段邏輯可以讀者自行分析, 如果之前分配相關(guān)的知識掌握扎實的話, 這里的邏輯也不是很難

回到PooledByteBuf的deallocate方法中:

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

最后, 通過recycle()將釋放的ByteBuf放入對象回收站, 有關(guān)對象回收站的知識, 會在以后的章節(jié)進行剖析

以上就是內(nèi)存回收的大概邏輯，更多關(guān)于Netty分布式ByteBuf使用回收的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: