Netty分布式ByteBuf使用的回收邏輯是什么

這篇文章主要介紹“Netty分布式ByteBuf使用的回收邏輯是什么”，在日常操作中，相信很多人在Netty分布式ByteBuf使用的回收邏輯是什么問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”Netty分布式ByteBuf使用的回收邏輯是什么”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

ByteBuf回收

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

PooledUnsafeDirectByteBuf中內存釋放的入口方法是其父類AbstractReferenceCountedByteBuf中的release方法:

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

這里調用了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為例, 所以這里會調用其父類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不再指向任何一塊內存

memory = null這里將memory也設置為null

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

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

我們首先分析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節點
    MemoryRegionCache<?> cache = cache(area, normCapacity, sizeClass);
    if (cache == null) {
        return false;
    }
    //將chunk, 和handle封裝成實體加到queue里面
    return cache.add(chunk, handle);
}

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

我們跟到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();
    }
}

假設我們是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);
}

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

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節點
    MemoryRegionCache<?> cache = cache(area, normCapacity, sizeClass);
    if (cache == null) {
        return false;
    }
    //將chunk, 和handle封裝成實體加到queue里面
    return cache.add(chunk, handle);
}

這里的cache對象調用了一個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, 如果不成功, 就會調用freeChunk(chunk, handle, sizeClass)方法, 這個方法的意義是, 將原先給ByteBuf分配的內存區段標記為未使用

跟進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釋放一段連續的內存

再跟到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)這 里判斷是當前緩沖區分配的級別是Page還是Subpage, 如果是Subpage, 則會找到相關的Subpage將其位圖標記為0

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

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

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

到此，關于“Netty分布式ByteBuf使用的回收邏輯是什么”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續學習更多相關知識，請繼續關注億速云網站，小編會繼續努力為大家帶來更多實用的文章！