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

更新時(shí)間：2022年03月28日 17:32:48 作者：向南是個(gè)萬(wàn)人迷

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

ByteBuf回收

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

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

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

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

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) 中判斷當(dāng)前byteBuf是否沒(méi)有被引用了, 如果沒(méi)有被引用, 則通過(guò)deallocate()方法進(jìn)行釋放

因?yàn)槲覀兪且訮ooledUnsafeDirectByteBuf為例, 所以這里會(huì)調(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表示當(dāng)前的ByteBuf不再指向任何一塊內(nèi)存

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

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

recycle()是將對(duì)象放入的對(duì)象回收站, 循環(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 {
        //那種級(jí)別的Size
        SizeClass sizeClass = sizeClass(normCapacity);
        //加到緩存里
        if (cache != null && cache.add(this, chunk, handle, normCapacity, sizeClass)) {
            return;
        }
        //將緩存對(duì)象標(biāo)記為未使用
        freeChunk(chunk, handle, sizeClass);
    }
}

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

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

cache.add(this, chunk, handle, normCapacity, sizeClass)是將當(dāng)前當(dāng)前ByteBuf進(jìn)行緩存

我們之前講過(guò), 再分配ByteBuf時(shí)首先在緩存上分配, 而這步, 就是將其緩存的過(guò)程, 跟進(jìn)去:

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

首先根據(jù)根據(jù)類型拿到相關(guān)類型緩存節(jié)點(diǎn), 這里會(huì)根據(jù)不同的內(nèi)存規(guī)格去找不同的對(duì)象, 我們簡(jiǎn)單回顧一下, 每個(gè)緩存對(duì)象都包含一個(gè)queue, queue中每個(gè)節(jié)點(diǎn)是entry, 每一個(gè)entry中包含一個(gè)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類型, 這里就會(huì)走到cacheForTiny(area, normCapacity)方法中, 跟進(jìn)去:

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

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

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

我們這里看到, 是直接通過(guò)下標(biāo)拿的緩存對(duì)象

回到add方法中

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

這里的cache對(duì)象調(diào)用了一個(gè)add方法, 這個(gè)方法就是將chunk和handle封裝成一個(gè)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;
}

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

然后通過(guò)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 {
        //那種級(jí)別的Size
        SizeClass sizeClass = sizeClass(normCapacity);
        //加到緩存里
        if (cache != null && cache.add(this, chunk, handle, normCapacity, sizeClass)) {
            return;
        } 
        freeChunk(chunk, handle, sizeClass);
    }
}

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

跟進(jìn)freeChunk簡(jiǎn)單分析下:

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);
    }
}

我們?cè)俑絝ree方法中:

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

chunk.free(handle)的意思是通過(guò)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)這里判斷是當(dāng)前緩沖區(qū)分配的級(jí)別是Page還是Subpage, 如果是Subpage, 則會(huì)找到相關(guān)的Subpage將其位圖標(biāo)記為0

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

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

回到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();
    }
}

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

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

您可能感興趣的文章: