您好,登錄后才能下訂單哦!
這篇文章主要介紹“Netty分布式ByteBuf使用subPage級(jí)別內(nèi)存分配的方法”的相關(guān)知識(shí),小編通過(guò)實(shí)際案例向大家展示操作過(guò)程,操作方法簡(jiǎn)單快捷,實(shí)用性強(qiáng),希望這篇“Netty分布式ByteBuf使用subPage級(jí)別內(nèi)存分配的方法”文章能幫助大家解決問(wèn)題。
簡(jiǎn)單起見(jiàn), 我們這里僅僅以16字節(jié)為例, 講解其分配邏輯
在分析其邏輯前, 首先看PoolArean的一個(gè)屬性:
private final PoolSubpage<T>[] tinySubpagePools;
這個(gè)屬性是一個(gè)PoolSubpage的數(shù)組, 有點(diǎn)類(lèi)似于一個(gè)subpage的緩存, 我們創(chuàng)建一個(gè)subpage之后, 會(huì)將創(chuàng)建的subpage與該屬性其中每個(gè)關(guān)聯(lián), 下次在分配的時(shí)候可以直接通過(guò)該屬性的元素去找關(guān)聯(lián)的subpage
tinySubpagePools = newSubpagePoolArray(numTinySubpagePools);
這里為numTinySubpagePools為32
跟到newSubpagePoolArray(numTinySubpagePools)方法里:
private PoolSubpage<T>[] newSubpagePoolArray(int size) { return new PoolSubpage[size]; }
這里直接創(chuàng)建了一個(gè)PoolSubpage數(shù)組, 長(zhǎng)度為32
for (int i = 0; i < tinySubpagePools.length; i ++) { tinySubpagePools[i] = newSubpagePoolHead(pageSize); }
我們跟到newSubpagePoolHead中:
private PoolSubpage<T> newSubpagePoolHead(int pageSize) { PoolSubpage<T> head = new PoolSubpage<T>(pageSize); head.prev = head; head.next = head; return head; }
這里創(chuàng)建了一個(gè)PoolSubpage對(duì)象head
head.prev = head; head.next = head;
這種寫(xiě)法我們知道Subpage其實(shí)也是個(gè)雙向鏈表, 這里的將head的上一個(gè)節(jié)點(diǎn)和下一個(gè)節(jié)點(diǎn)都設(shè)置為自身, 有關(guān)PoolSubpage的關(guān)聯(lián)關(guān)系, 我們稍后會(huì)看到
這樣通過(guò)循環(huán)創(chuàng)建PoolSubpage, 總共會(huì)創(chuàng)建出32個(gè)subpage, 其中每個(gè)subpage實(shí)際代表一塊內(nèi)存大小:
5-8-1
這里就有點(diǎn)類(lèi)之前小節(jié)的緩存數(shù)組tinySubPageDirectCaches的結(jié)構(gòu)
了解了tinySubpagePools屬性, 我們看PoolArean的allocate方法, 也就是緩沖區(qū)的入口方法:
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; } //通過(guò)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); } }
之前我們最這個(gè)方法剖析過(guò)在page級(jí)別相關(guān)內(nèi)存分配邏輯, 這一小節(jié)看subpage級(jí)別分配的相關(guān)邏輯
假設(shè)我們分配16字節(jié)的緩沖區(qū), isTinyOrSmall(normCapacity)就會(huì)返回true, 進(jìn)入if塊
同樣if (tiny)這里會(huì)返回true, 繼續(xù)跟到if (tiny)中:
首先會(huì)在緩存中分配緩沖區(qū), 如果分配不到, 就開(kāi)辟一塊內(nèi)存進(jìn)行內(nèi)存分配
首先看這一步:
tableIdx = tinyIdx(normCapacity);
我們跟進(jìn)去:
static int tinyIdx(int normCapacity) { return normCapacity >>> 4; }
這里將normCapacity除以16, 其實(shí)也就是1
我們回到PoolArena的allocate方法繼續(xù)看:
table = tinySubpagePools
這里將tinySubpagePools賦值到局部變量table中, 繼續(xù)往下看
final PoolSubpage<T> head = table[tableIdx]
這步時(shí)通過(guò)下標(biāo)拿到一個(gè)PoolSubpage, 因?yàn)槲覀円?6字節(jié)為例, 所以我們拿到下標(biāo)為1的PoolSubpage, 對(duì)應(yīng)的內(nèi)存大小也就是16B
再看 final PoolSubpage<T> s = head.next 這一步, 跟我們剛才了解的的tinySubpagePools屬性, 默認(rèn)情況下head.next也是自身, 所以if (s != head)會(huì)返回false, 我們繼續(xù)往下看:
下面, 會(huì)走到allocateNormal(buf, reqCapacity, normCapacity)這個(gè)方法:
private synchronized void allocateNormal(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) { //首先在原來(lái)的chunk上進(jìn)行內(nèi)存分配(1) if (q050.allocate(buf, reqCapacity, normCapacity) || q025.allocate(buf, reqCapacity, normCapacity) || q000.allocate(buf, reqCapacity, normCapacity) || qInit.allocate(buf, reqCapacity, normCapacity) || q075.allocate(buf, reqCapacity, normCapacity)) { ++allocationsNormal; return; } //創(chuàng)建chunk進(jìn)行內(nèi)存分配(2) PoolChunk<T> c = newChunk(pageSize, maxOrder, pageShifts, chunkSize); long handle = c.allocate(normCapacity); ++allocationsNormal; assert handle > 0; //初始化byteBuf(3) c.initBuf(buf, handle, reqCapacity); qInit.add(c); }
這里的邏輯我們之前的小節(jié)已經(jīng)剖析過(guò), 首先在原來(lái)的chunk中分配, 如果分配不成功, 則會(huì)創(chuàng)建chunk進(jìn)行分配
我們看這一步
long handle = c.allocate(normCapacity)
long allocate(int normCapacity) { if ((normCapacity & subpageOverflowMask) != 0) { return allocateRun(normCapacity); } else { return allocateSubpage(normCapacity); } }
上一小節(jié)我們分析page級(jí)別分配的時(shí)候, 剖析的是allocateRun(normCapacity)方法
因?yàn)檫@里我們是以16字節(jié)舉例, 所以這次我們剖析allocateSubpage(normCapacity)方法, 也就是在subpage級(jí)別進(jìn)行內(nèi)存分配
private long allocateSubpage(int normCapacity) { PoolSubpage<T> head = arena.findSubpagePoolHead(normCapacity); synchronized (head) { int d = maxOrder; //表示在第11層分配節(jié)點(diǎn) int id = allocateNode(d); if (id < 0) { return id; } //獲取初始化的subpage final PoolSubpage<T>[] subpages = this.subpages; final int pageSize = this.pageSize; freeBytes -= pageSize; //表示第幾個(gè)subpageIdx int subpageIdx = subpageIdx(id); PoolSubpage<T> subpage = subpages[subpageIdx]; if (subpage == null) { //如果subpage為空 subpage = new PoolSubpage<T>(head, this, id, runOffset(id), pageSize, normCapacity); //則將當(dāng)前的下標(biāo)賦值為subpage subpages[subpageIdx] = subpage; } else { subpage.init(head, normCapacity); } //取出一個(gè)子page return subpage.allocate(); } }
首先, 通過(guò) PoolSubpage<T> head = arena.findSubpagePoolHead(normCapacity) 這種方式找到head節(jié)點(diǎn), 實(shí)際上這里head, 就是我們剛才分析的tinySubpagePools屬性的第一個(gè)節(jié)點(diǎn), 也就是對(duì)應(yīng)16B的那個(gè)節(jié)點(diǎn)
int d = maxOrder 是將11賦值給d, 也就是在內(nèi)存樹(shù)的第11層取節(jié)點(diǎn), 這部分上一小節(jié)剖析過(guò)了, 可以回顧圖5-8-5部分
int id = allocateNode(d) 這里獲取的是上一小節(jié)我們分析過(guò)的, 字節(jié)數(shù)組memoryMap的下標(biāo), 這里指向一個(gè)page, 如果第一次分配, 指向的是0-8k的那個(gè)page, 上一小節(jié)對(duì)此進(jìn)行詳細(xì)的剖析這里不再贅述
final PoolSubpage<T>[] subpages = this.subpages 這一步, 是拿到PoolChunk中成員變量subpages的值, 也是個(gè)PoolSubpage的數(shù)組, 在PoolChunk進(jìn)行初始化的時(shí)候, 也會(huì)初始化該數(shù)組, 長(zhǎng)度為2048
也就是說(shuō)每個(gè)chunk都維護(hù)著一個(gè)subpage的列表, 如果每一個(gè)page級(jí)別的內(nèi)存都需要被切分成子page, 則會(huì)將這個(gè)這個(gè)page放入該列表中, 專(zhuān)門(mén)用于分配子page, 所以這個(gè)列表中的subpage, 其實(shí)就是一個(gè)用于切分的page
5-8-2
int subpageIdx = subpageIdx(id) 這一步是通過(guò)id拿到這個(gè)PoolSubpage數(shù)組的下標(biāo), 如果id對(duì)應(yīng)的page是0-8k的節(jié)點(diǎn), 這里拿到的下標(biāo)就是0
在 if (subpage == null) 中, 因?yàn)槟J(rèn)subpages只是創(chuàng)建一個(gè)數(shù)組, 并沒(méi)有往數(shù)組中賦值, 所以第一次走到這里會(huì)返回true, 跟到if塊中:
subpage = new PoolSubpage<T>(head, this, id, runOffset(id), pageSize, normCapacity);
這里通過(guò)new PoolSubpage創(chuàng)建一個(gè)新的subpage之后, 通過(guò) subpages[subpageIdx] = subpage 這種方式將新創(chuàng)建的subpage根據(jù)下標(biāo)賦值到subpages中的元素中
在new PoolSubpage的構(gòu)造方法中, 傳入head, 就是我們剛才提到過(guò)的tinySubpagePools屬性中的節(jié)點(diǎn), 如果我們分配的16字節(jié)的緩沖區(qū), 則這里對(duì)應(yīng)的就是第一個(gè)節(jié)點(diǎn)
PoolSubpage(PoolSubpage<T> head, PoolChunk<T> chunk, int memoryMapIdx, int runOffset, int pageSize, int elemSize) { this.chunk = chunk; this.memoryMapIdx = memoryMapIdx; this.runOffset = runOffset; this.pageSize = pageSize; bitmap = new long[pageSize >>> 10]; init(head, elemSize); }
這里重點(diǎn)關(guān)注屬性bitmap, 這是一個(gè)long類(lèi)型的數(shù)組, 初始大小為8, 這里只是初始化的大小, 真正的大小要根據(jù)將page切分多少塊而確定
這里將屬性進(jìn)行了賦值, 我們跟到init方法中:
void init(PoolSubpage<T> head, int elemSize) { doNotDestroy = true; this.elemSize = elemSize; if (elemSize != 0) { maxNumElems = numAvail = pageSize / elemSize; nextAvail = 0; bitmapLength = maxNumElems >>> 6; if ((maxNumElems & 63) != 0) { bitmapLength ++; } for (int i = 0; i < bitmapLength; i ++) { //bitmap標(biāo)識(shí)哪個(gè)子page被分配 //0標(biāo)識(shí)未分配, 1表示已分配 bitmap [i] = 0; } } //加到arena里面 addToPool(head); }
this.elemSize = elemSize 表示保存當(dāng)前分配的緩沖區(qū)大小, 這里我們以16字節(jié)舉例, 所以這里是16
maxNumElems = numAvail = pageSize / elemSize
這里初始化了兩個(gè)屬性maxNumElems, numAvail, 值都為pageSize / elemSize, 表示一個(gè)page大小除以分配的緩沖區(qū)大小, 也就是表示當(dāng)前page被劃分了多少分
numAvail則表示剩余可用的塊數(shù), 由于第一次分配都是可用的, 所以 numAvail=maxNumElems
bitmapLength表示bitmap的實(shí)際大小, 剛才我們分析過(guò), bitmap初始化的大小為8, 但實(shí)際上并不一定需要8個(gè)元素, 元素個(gè)數(shù)要根據(jù)page切分的子塊而定, 這里的大小是所切分的子塊數(shù)除以64
再往下看, if ((maxNumElems & 63) != 0) 判斷maxNumElems也就是當(dāng)前配置所切分的子塊是不是64的倍數(shù), 如果不是, 則bitmapLength加1,
最后通過(guò)循環(huán), 將其分配的大小中的元素賦值為0
這里詳細(xì)介紹一下有關(guān)bitmap, 這里是個(gè)long類(lèi)型的數(shù)組, long數(shù)組中的每一個(gè)值, 也就是long類(lèi)型的數(shù)字, 其中的每一個(gè)比特位, 都標(biāo)記著page中每一個(gè)子塊的內(nèi)存是否已分配, 如果比特位是1, 表示該子塊已分配, 如果比特位是0, 表示該子塊未分配, 標(biāo)記順序是其二進(jìn)制數(shù)從低位到高位進(jìn)行排列
這里, 我們應(yīng)該知道為什么bitmap大小要設(shè)置為子塊數(shù)量除以, 64, 因?yàn)閘ong類(lèi)型的數(shù)字是64位, 每一個(gè)元素能記錄64個(gè)子塊的數(shù)量, 這樣就可以通過(guò)子page個(gè)數(shù)除以64的方式?jīng)Q定bitmap中元素的數(shù)量
如果子塊不能整除64, 則通過(guò)元素?cái)?shù)量+1方式, 除以64之后剩余的子塊通過(guò)long中比特位由低到高進(jìn)行排列記錄
這里的邏輯結(jié)構(gòu)如下所示:
5-8-3
private void addToPool(PoolSubpage<T> head) { assert prev == null && next == null; prev = head; next = head.next; next.prev = this; head.next = this; }
這里的head我們剛才講過(guò), 是Arena中數(shù)組tinySubpagePools中的元素, 通過(guò)以上邏輯, 就會(huì)將新創(chuàng)建的Subpage通過(guò)雙向鏈表的方式關(guān)聯(lián)到tinySubpagePools中的元素, 我們以16字節(jié)為例, 關(guān)聯(lián)關(guān)系如圖所示:
5-8-4
這樣, 下次如果還需要分配16字節(jié)的內(nèi)存, 就可以通過(guò)tinySubpagePools找到其元素關(guān)聯(lián)的subpage進(jìn)行分配了
我們?cè)倩氐絇oolChunk的allocateSubpage方法中:
private long allocateSubpage(int normCapacity) { PoolSubpage<T> head = arena.findSubpagePoolHead(normCapacity); synchronized (head) { int d = maxOrder; //表示在第11層分配節(jié)點(diǎn) int id = allocateNode(d); if (id < 0) { return id; } //獲取初始化的subpage final PoolSubpage<T>[] subpages = this.subpages; final int pageSize = this.pageSize; freeBytes -= pageSize; //表示第幾個(gè)subpageIdx int subpageIdx = subpageIdx(id); PoolSubpage<T> subpage = subpages[subpageIdx]; if (subpage == null) { //如果subpage為空 subpage = new PoolSubpage<T>(head, this, id, runOffset(id), pageSize, normCapacity); //則將當(dāng)前的下標(biāo)賦值為subpage subpages[subpageIdx] = subpage; } else { subpage.init(head, normCapacity); } //取出一個(gè)子page return subpage.allocate(); } }
創(chuàng)建完了一個(gè)subpage, 我們就可以通過(guò)subpage.allocate()方法進(jìn)行內(nèi)存分配了
long allocate() { if (elemSize == 0) { return toHandle(0); } if (numAvail == 0 || !doNotDestroy) { return -1; } //取一個(gè)bitmap中可用的id(絕對(duì)id) final int bitmapIdx = getNextAvail(); //除以64(bitmap的相對(duì)下標(biāo)) int q = bitmapIdx >>> 6; //除以64取余, 其實(shí)就是當(dāng)前絕對(duì)id的偏移量 int r = bitmapIdx & 63; assert (bitmap[q] >>> r & 1) == 0; //當(dāng)前位標(biāo)記為1 bitmap[q] |= 1L << r; //如果可用的子page為0 //可用的子page-1 if (-- numAvail == 0) { //則移除相關(guān)子page removeFromPool(); } //bitmapIdx轉(zhuǎn)換成handler return toHandle(bitmapIdx); }
這里的邏輯看起來(lái)比較復(fù)雜, 這里帶著大家一點(diǎn)點(diǎn)剖析:
首先看:
final int bitmapIdx = getNextAvail();
其中bitmapIdx表示從bitmap中找到一個(gè)可用的bit位的下標(biāo), 注意, 這里是bit的下標(biāo), 并不是數(shù)組的下標(biāo), 我們之前分析過(guò), 因?yàn)槊恳槐忍匚淮硪粋€(gè)子塊的內(nèi)存分配情況, 通過(guò)這個(gè)下標(biāo)就可以知道那個(gè)比特位是未分配狀態(tài)
我們跟進(jìn)這個(gè)方法:
private int getNextAvail() { //nextAvail=0 int nextAvail = this.nextAvail; if (nextAvail >= 0) { //一個(gè)子page被釋放之后, 會(huì)記錄當(dāng)前子page的bitmapIdx的位置, 下次分配可以直接通過(guò)bitmapIdx拿到一個(gè)子page this.nextAvail = -1; return nextAvail; } return findNextAvail(); }
這里nextAvail, 表示下一個(gè)可用的bitmapIdx, 在釋放的時(shí)候的會(huì)被標(biāo)記, 標(biāo)記被釋放的子塊對(duì)應(yīng)bitmapIdx的下標(biāo), 如果<0則代表沒(méi)有被釋放的子塊, 則通過(guò)findNextAvail方法進(jìn)行查找
private int findNextAvail() { //當(dāng)前l(fā)ong數(shù)組 final long[] bitmap = this.bitmap; //獲取其長(zhǎng)度 final int bitmapLength = this.bitmapLength; for (int i = 0; i < bitmapLength; i ++) { //第i個(gè) long bits = bitmap[i]; //!=-1 說(shuō)明64位沒(méi)有全部占滿 if (~bits != 0) { //找下一個(gè)節(jié)點(diǎn) return findNextAvail0(i, bits); } } return -1; }
這里會(huì)遍歷bitmap中的每一個(gè)元素, 如果當(dāng)前元素中所有的比特位并沒(méi)有全部標(biāo)記被使用, 則通過(guò)findNextAvail0(i, bits)方法挨個(gè)往后找標(biāo)記未使用的比特位
再繼續(xù)跟findNextAvail0:
private int findNextAvail0(int i, long bits) { //多少份 final int maxNumElems = this.maxNumElems; //乘以64, 代表當(dāng)前l(fā)ong的第一個(gè)下標(biāo) final int baseVal = i << 6; //循環(huán)64次(指代當(dāng)前的下標(biāo)) for (int j = 0; j < 64; j ++) { //第一位為0(如果是2的倍數(shù), 則第一位就是0) if ((bits & 1) == 0) { //這里相當(dāng)于加, 將i*64之后加上j, 獲取絕對(duì)下標(biāo) int val = baseVal | j; //小于塊數(shù)(不能越界) if (val < maxNumElems) { return val; } else { break; } } //當(dāng)前下標(biāo)不為0 //右移一位 bits >>>= 1; } return -1; }
這里從當(dāng)前元素的第一個(gè)比特位開(kāi)始找, 直到找到一個(gè)標(biāo)記為0的比特位, 并返回當(dāng)前比特位的下標(biāo), 大概流程如下圖所示:
5-8-5
long allocate() { if (elemSize == 0) { return toHandle(0); } if (numAvail == 0 || !doNotDestroy) { return -1; } //取一個(gè)bitmap中可用的id(絕對(duì)id) final int bitmapIdx = getNextAvail(); //除以64(bitmap的相對(duì)下標(biāo)) int q = bitmapIdx >>> 6; //除以64取余, 其實(shí)就是當(dāng)前絕對(duì)id的偏移量 int r = bitmapIdx & 63; assert (bitmap[q] >>> r & 1) == 0; //當(dāng)前位標(biāo)記為1 bitmap[q] |= 1L << r; //如果可用的子page為0 //可用的子page-1 if (-- numAvail == 0) { //則移除相關(guān)子page removeFromPool(); } //bitmapIdx轉(zhuǎn)換成handler return toHandle(bitmapIdx); }
找到可用的bitmapIdx之后, 通過(guò) int q = bitmapIdx >>> 6 獲取bitmap中bitmapIdx所屬元素的數(shù)組下標(biāo)
int r = bitmapIdx & 63 表示獲取bitmapIdx的位置是從當(dāng)前元素最低位開(kāi)始的第幾個(gè)比特位
bitmap[q] |= 1L << r 是將bitmap的位置設(shè)置為不可用, 也就是比特位設(shè)置為1, 表示已占用
然后將可用子配置的數(shù)量numAvail減一
如果沒(méi)有可用子page的數(shù)量, 則會(huì)將PoolArena中的數(shù)組tinySubpagePools所關(guān)聯(lián)的subpage進(jìn)行移除, 移除之后參考圖5-8-1
最后通過(guò)toHandle(bitmapIdx)獲取當(dāng)前子塊的handle, 上一小節(jié)我們知道handle指向的是當(dāng)前chunk中的唯一的一塊內(nèi)存, 我們跟進(jìn)toHandle(bitmapIdx)中:
private long toHandle(int bitmapIdx) { return 0x4000000000000000L | (long) bitmapIdx << 32 | memoryMapIdx; }
(long) bitmapIdx << 32 是將bitmapIdx右移32位, 而32位正好是一個(gè)int的長(zhǎng)度, 這樣, 通過(guò) (long) bitmapIdx << 32 | memoryMapIdx 計(jì)算, 就可以將memoryMapIdx, 也就是page所屬的下標(biāo)的二進(jìn)制數(shù)保存在 (long) bitmapIdx << 32 的低32位中
0x4000000000000000L是一個(gè)最高位是1并且所有低位都是0的二進(jìn)制數(shù), 這樣通過(guò)按位或的方式可以將 (long) bitmapIdx << 32 | memoryMapIdx 計(jì)算出來(lái)的結(jié)果保存在0x4000000000000000L的所有低位中, 這樣, 返回對(duì)的數(shù)字就可以指向chunk中唯一的一塊內(nèi)存
我們回到PoolArena的allocateNormal方法中:
private synchronized void allocateNormal(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) { //首先在原來(lái)的chunk上進(jìn)行內(nèi)存分配(1) if (q050.allocate(buf, reqCapacity, normCapacity) || q025.allocate(buf, reqCapacity, normCapacity) || q000.allocate(buf, reqCapacity, normCapacity) || qInit.allocate(buf, reqCapacity, normCapacity) || q075.allocate(buf, reqCapacity, normCapacity)) { ++allocationsNormal; return; } //創(chuàng)建chunk進(jìn)行內(nèi)存分配(2) PoolChunk<T> c = newChunk(pageSize, maxOrder, pageShifts, chunkSize); long handle = c.allocate(normCapacity); ++allocationsNormal; assert handle > 0; //初始化byteBuf(3) c.initBuf(buf, handle, reqCapacity); qInit.add(c); }
我們分析完了long handle = c.allocate(normCapacity)這步, 這里返回的handle就指向chunk中的某個(gè)page中的某個(gè)子塊所對(duì)應(yīng)的連續(xù)內(nèi)存
最后, 通過(guò)iniBuf初始化之后, 將創(chuàng)建的chunk加到ChunkList里面
void initBuf(PooledByteBuf<T> buf, long handle, int reqCapacity) { int memoryMapIdx = memoryMapIdx(handle); //bitmapIdx是后面分配subpage時(shí)候使用到的 int bitmapIdx = bitmapIdx(handle); if (bitmapIdx == 0) { byte val = value(memoryMapIdx); assert val == unusable : String.valueOf(val); //runOffset(memoryMapIdx):偏移量 //runLength(memoryMapIdx):當(dāng)前節(jié)點(diǎn)的長(zhǎng)度 buf.init(this, handle, runOffset(memoryMapIdx), reqCapacity, runLength(memoryMapIdx), arena.parent.threadCache()); } else { initBufWithSubpage(buf, handle, bitmapIdx, reqCapacity); } }
這部分在之前的小節(jié)我們剖析過(guò), 相信大家不會(huì)陌生, 這里有區(qū)別的是 if (bitmapIdx == 0) 的判斷, 這里的bitmapIdx不會(huì)是0, 這樣, 就會(huì)走到initBufWithSubpage(buf, handle, bitmapIdx, reqCapacity)方法中
跟到initBufWithSubpage方法:
private void initBufWithSubpage(PooledByteBuf<T> buf, long handle, int bitmapIdx, int reqCapacity) { assert bitmapIdx != 0; int memoryMapIdx = memoryMapIdx(handle); PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)]; assert subpage.doNotDestroy; assert reqCapacity <= subpage.elemSize; buf.init( this, handle, runOffset(memoryMapIdx) + (bitmapIdx & 0x3FFFFFFF) * subpage.elemSize, reqCapacity, subpage.elemSize, arena.parent.threadCache()); }
首先拿到memoryMapIdx, 這里會(huì)將我們之前計(jì)算handle傳入, 跟進(jìn)去:
private static int memoryMapIdx(long handle) { return (int) handle; }
這里將其強(qiáng)制轉(zhuǎn)化為int類(lèi)型, 也就是去掉高32位, 這樣就得到memoryMapIdx
我們注意在buf調(diào)用init方法中的一個(gè)參數(shù): runOffset(memoryMapIdx) + (bitmapIdx & 0x3FFFFFFF) * subpage.elemSize
這里的偏移量就是, 原來(lái)page的偏移量+子塊的偏移量
bitmapIdx & 0x3FFFFFFF 代表當(dāng)前分配的子page是屬于第幾個(gè)子page
(bitmapIdx & 0x3FFFFFFF) * subpage.elemSize 表示在當(dāng)前page的偏移量
這樣, 分配的ByteBuf在內(nèi)存讀寫(xiě)的時(shí)候, 就會(huì)根據(jù)偏移量進(jìn)行讀寫(xiě)
最后我們跟到init方法中
void init(PoolChunk<T> chunk, long handle, int offset, int length, int maxLength, PoolThreadCache cache) { //初始化 assert handle >= 0; assert chunk != null; //在哪一塊內(nèi)存上進(jìn)行分配的 this.chunk = chunk; //這一塊內(nèi)存上的哪一塊連續(xù)內(nèi)存 this.handle = handle; memory = chunk.memory; //偏移量 this.offset = offset; this.length = length; this.maxLength = maxLength; tmpNioBuf = null; this.cache = cache; }
關(guān)于“Netty分布式ByteBuf使用subPage級(jí)別內(nèi)存分配的方法”的內(nèi)容就介紹到這里了,感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí),可以關(guān)注億速云行業(yè)資訊頻道,小編每天都會(huì)為大家更新不同的知識(shí)點(diǎn)。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。