protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
    PoolThreadCache cache = threadCache.get();
    PoolArena<ByteBuffer> directArena = cache.directArena;
    ByteBuf buf;
    if (directArena != null) { 
        buf = directArena.allocate(cache, initialCapacity, maxCapacity);
    } else {
        if (PlatformDependent.hasUnsafe()) {
            buf = UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);
        } else {
            buf = new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
        }
    }
    return toLeakAwareBuffer(buf);
}

獲取了directArena對(duì)象之后, 通過allocate方法分配一個(gè)ByteBuf, 這里allocate方法是PoolArena類中的方法

跟到allocate方法中:

PooledByteBuf<T> allocate(PoolThreadCache cache, int reqCapacity, int maxCapacity) { 
    PooledByteBuf<T> buf = newByteBuf(maxCapacity); 
    allocate(cache, buf, reqCapacity);
    return buf;
}

首先通過newByteBuf獲得一個(gè)ByteBuf對(duì)象

再通過allocate方法進(jìn)行分配, 這里要注意, 這里進(jìn)行分配的時(shí)候是線程私有的directArena進(jìn)行分配

我們跟到newByteBuf方法中

因?yàn)槭莇irectArena調(diào)用的newByteBuf, 所以這里會(huì)進(jìn)入DirectArena類的newByteBuf中:

protected PooledByteBuf<ByteBuffer> newByteBuf(int maxCapacity) { 
    if (HAS_UNSAFE) { 
        return PooledUnsafeDirectByteBuf.newInstance(maxCapacity);
    } else {
        return PooledDirectByteBuf.newInstance(maxCapacity);
    }
}

因?yàn)槟J(rèn)通常是有unsafe對(duì)象的, 所以這里會(huì)走到這一步中PooledUnsafeDirectByteBuf.newInstance(maxCapacity)

通過靜態(tài)方法newInstance創(chuàng)建一個(gè)PooledUnsafeDirectByteBuf對(duì)象

跟到newInstance方法中:

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

這里通過RECYCLER.get()這種方式拿到一個(gè)ByteBuf對(duì)象, RECYCLER其實(shí)是一個(gè)對(duì)象回收站, 這部分內(nèi)容會(huì)在后面的內(nèi)容中詳細(xì)剖析, 這里我們只需要知道, 這種方式能從回收站中拿到一個(gè)對(duì)象, 如果回收站里沒有相關(guān)對(duì)象, 則創(chuàng)建一個(gè)新

因?yàn)檫@里有可能是從回收站中拿出的一個(gè)對(duì)象, 所以通過reuse進(jìn)行復(fù)用

跟到reuse方法中

final void reuse(int maxCapacity) {
    maxCapacity(maxCapacity);
    setRefCnt(1);
    setIndex0(0, 0);
    discardMarks();
}

這里設(shè)置了的最大可擴(kuò)容內(nèi)存, 對(duì)象的引用數(shù)量, 讀寫指針位置都重置為0, 以及讀寫指針的位置標(biāo)記也都重置為0

我們回到PoolArena的allocate方法中:

PooledByteBuf<T> allocate(PoolThreadCache cache, int reqCapacity, int maxCapacity) { 
    PooledByteBuf<T> buf = newByteBuf(maxCapacity); 
    allocate(cache, buf, reqCapacity);
    return buf;
}

拿到了ByteBuf對(duì)象, 就可以通過allocate(cache, buf, reqCapacity)方法進(jìn)行內(nèi)存分配了

跟到allocate方法中

private void allocate(PoolThreadCache cache, PooledByteBuf<T> buf, final int reqCapacity) {
    //規(guī)格化
    final int normCapacity = normalizeCapacity(reqCapacity);
    if (isTinyOrSmall(normCapacity)) { 
        int tableIdx;
        PoolSubpage<T>[] table;
        //判斷是不是tinty
        boolean tiny = isTiny(normCapacity);
        if (tiny) { // < 512
            //緩存分配
            if (cache.allocateTiny(this, buf, reqCapacity, normCapacity)) {
                return;
            }
            //通過tinyIdx拿到tableIdx
            tableIdx = tinyIdx(normCapacity);
            //subpage的數(shù)組
            table = tinySubpagePools;
        } else {
            if (cache.allocateSmall(this, buf, reqCapacity, normCapacity)) {
                return;
            }
            tableIdx = smallIdx(normCapacity);
            table = smallSubpagePools;
        }
        //拿到對(duì)應(yīng)的節(jié)點(diǎn)
        final PoolSubpage<T> head = table[tableIdx];

        synchronized (head) {
            final PoolSubpage<T> s = head.next;
            //默認(rèn)情況下, head的next也是自身
            if (s != head) {
                assert s.doNotDestroy && s.elemSize == normCapacity;
                long handle = s.allocate();
                assert handle >= 0;
                s.chunk.initBufWithSubpage(buf, handle, reqCapacity);

                if (tiny) {
                    allocationsTiny.increment();
                } else {
                    allocationsSmall.increment();
                }
                return;
            }
        }
        allocateNormal(buf, reqCapacity, normCapacity);
        return;
    }
    if (normCapacity <= chunkSize) {
        //首先在緩存上進(jìn)行內(nèi)存分配
        if (cache.allocateNormal(this, buf, reqCapacity, normCapacity)) {
            //分配成功, 返回
            return;
        }
        //分配不成功, 做實(shí)際的內(nèi)存分配
        allocateNormal(buf, reqCapacity, normCapacity);
    } else {
        //大于這個(gè)值, 就不在緩存上分配
        allocateHuge(buf, reqCapacity);
    }
}

這里看起來邏輯比較長(zhǎng), 其實(shí)主要步驟分為兩步

1.首先在緩存上進(jìn)行分配

對(duì)應(yīng)步驟是:

cache.allocateTiny(this, buf, reqCapacity, normCapacity)

cache.allocateSmall(this, buf, reqCapacity, normCapacity)

cache.allocateNormal(this, buf, reqCapacity, normCapacity)

2.如果在緩存上分配不成功, 則實(shí)際分配一塊內(nèi)存

對(duì)應(yīng)步驟是

allocateNormal(buf, reqCapacity, normCapacity)

在這里對(duì)幾種類型的內(nèi)存進(jìn)行介紹:

之前的小節(jié)我們介紹過, 緩沖區(qū)內(nèi)存類型分為tiny, small, 和normal, 其實(shí)還有種不常見的類型叫做huge, 那么這幾種類型的內(nèi)存有什么區(qū)別呢, 實(shí)際上這幾種類型是按照緩沖區(qū)初始化空間的范圍進(jìn)行區(qū)分的, 具體區(qū)分如下:

tiny類型對(duì)應(yīng)的緩沖區(qū)范圍為0-512B

small類型對(duì)應(yīng)的緩沖區(qū)范圍為512B-8K

normal類型對(duì)應(yīng)的緩沖區(qū)范圍為8K-16MB

huge類型對(duì)應(yīng)緩沖區(qū)范圍為大于16MB

簡(jiǎn)單介紹下有關(guān)范圍的含義:

16MB對(duì)應(yīng)一個(gè)chunk, netty是以chunk為單位向操作系統(tǒng)申請(qǐng)內(nèi)存的

8k對(duì)應(yīng)一個(gè)page, page是將chunk切分后的結(jié)果, 一個(gè)chunk對(duì)應(yīng)2048個(gè)page

8k以下對(duì)應(yīng)一個(gè)subpage, subpage是page的切分, 一個(gè)page可以切分多個(gè)subpage, 具體切分幾個(gè)需要根據(jù)subpage的大小而定, 比如只要分配1k的緩沖區(qū), 則會(huì)將page切分成8個(gè)subpage

以上就是directArena內(nèi)存分配的大概流程和相關(guān)概念，更多關(guān)于Netty分布式ByteBuf directArena分配緩沖區(qū)的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: