多年之前，從C記憶體的手動管理上升到java的自動GC，是歷史的巨大進步。然而多年之後，netty的記憶體實現又曲線的回到了手動管理模式，正印證了馬克思哲學觀：社會總是在螺旋式前進的，沒有永遠的最好。的確，就記憶體管理而言，GC給程式設計師帶來的價值是不言而喻的，不僅大大的降低了程式設計師的負擔，而且也極大的減少了記憶體管理帶來的Crash困擾，不過也有很多情況，可能手動的記憶體管理更為合適。

接下去準備幾個篇幅對Netty的記憶體管理進行深入分析。

PoolChunk

為了能夠簡單的操作記憶體，必須保證每次分配到的記憶體時連續的。Netty中底層的記憶體分配和回收管理主要由PoolChunk實現，其內部維護一棵平衡二叉樹memoryMap，所有子節點管理的記憶體也屬於其父節點。

memoryMap

poolChunk預設由2048個page組成，一個page預設大小為8k，圖中節點的值為在陣列memoryMap的下標。
1、如果需要分配大小8k的記憶體，則只需要在第11層，找到第一個可用節點即可。
2、如果需要分配大小16k的記憶體，則只需要在第10層，找到第一個可用節點即可。
3、如果節點1024存在一個已經被分配的子節點2048，則該節點不能被分配，如需要分配大小16k的記憶體，這個時候節點2048已被分配，節點2049未被分配，就不能直接分配節點1024，因為該節點目前只剩下8k記憶體。

poolChunk內部會保證每次分配記憶體大小為8K*(2^n)，為了分配一個大小為chunkSize/(2^k)的節點，需要在深度為k的層從左開始匹配節點，那麼如何快速的分配到指定記憶體？

memoryMap初始化：

memoryMap = new byte[maxSubpageAllocs << 1]; depthMap = new byte[memoryMap.length]; int memoryMapIndex = 1; for (int d = 0; d <= maxOrder; ++ d) { // move down the tree one level at a time int depth = 1 << d; for (int p = 0; p < depth; ++ p) { // in each level traverse left to right and set value to the depth of subtree memoryMap[memoryMapIndex] = (byte) d; depthMap[memoryMapIndex] = (byte) d; memoryMapIndex ++; } }

123456789101112

memoryMap=newbyte[maxSubpageAllocs<<1];depthMap=newbyte[memoryMap.length];intmemoryMapIndex=1;for(intd=0;d<=maxOrder;++d){// move down the tree one level at a timeintdepth=1<<d;for(intp=0;p<depth;++p){// in each level traverse left to right and set value to the depth of subtreememoryMap[memoryMapIndex]=(byte)d;depthMap[memoryMapIndex]=(byte)d;memoryMapIndex++;}}

memoryMap陣列中每個位置儲存的是該節點所在的層數，有什麼作用？對於節點512，其層數是9，則：
1、如果memoryMap[512] = 9，則表示其本身到下面所有的子節點都可以被分配；
2、如果memoryMap[512] = 10， 則表示節點512下有子節點已經分配過，則該節點不能直接被分配，而其子節點中的第10層還存在未分配的節點;
3、如果memoryMap[512] = 12 (即總層數 + 1）, 可分配的深度已經大於總層數, 則表示該節點下的所有子節點都已經被分配。

下面看看如何向PoolChunk申請一段記憶體：

long allocate(int normCapacity) { if ((normCapacity & subpageOverflowMask) != 0) { // >= pageSize return allocateRun(normCapacity); } else { return allocateSubpage(normCapacity); } }

1234567

longallocate(intnormCapacity){if((normCapacity&subpageOverflowMask)!=0){// >= pageSizereturnallocateRun(normCapacity);}else{returnallocateSubpage(normCapacity);}}

1、當需要分配的記憶體大於pageSize時，使用allocateRun實現記憶體分配。
2、否則使用方法allocateSubpage分配記憶體，在allocateSubpage實現中，會把一個page分割成多段，進行記憶體分配。

這裡先看看allocateRun是如何實現的：

private long allocateRun(int normCapacity) { int d = maxOrder - (log2(normCapacity) - pageShifts); int id = allocateNode(d); if (id < 0) { return id; } freeBytes -= runLength(id); return id; }

123456789

privatelongallocateRun(intnormCapacity){intd=maxOrder-(log2(normCapacity)-pageShifts);intid=allocateNode(d);if(id<0){returnid;}freeBytes-=runLength(id);returnid;}

1、normCapacity是處理過的值，如申請大小為1000的記憶體，實際申請的記憶體大小為1024。
2、d = maxOrder – (log2(normCapacity) – pageShifts) 可以確定需要在二叉樹的d層開始節點匹配。
其中pageShifts預設值為13，為何是13？因為只有當申請記憶體大小大於2^13（8192）時才會使用方法allocateRun分配記憶體。
3、方法allocateNode實現在二叉樹中進行節點匹配，具體實現如下：

private int allocateNode(int d) { int id = 1; int initial = - (1 << d); //value(id)=memoryMap[id] byte val = value(id); if (val > d) { // unusable return -1; } while (val < d || (id & initial) == 0) { // id & initial == 1 << d for all ids at depth d, for < d it is 0 id <<= 1; val = value(id); if (val > d) { id ^= 1; val = value(id); } } byte value = value(id); assert value == d && (id & initial) == 1 << d : String.format("val = %d, id & initial = %d, d = %d", value, id & initial, d); setValue(id, unusable); // mark as unusable updateParentsAlloc(id); return id; }

1234567891011121314151617181920212223

privateintallocateNode(intd){intid=1;intinitial=-(1<<d);//value(id)=memoryMap[id] byteval=value(id);if(val>d){// unusablereturn-1;}while(val<d||(id&initial)==0){// id & initial == 1 << d for all ids at depth d, for < d it is 0id<<=1;val=value(id);if(val>d){id^=1;val=value(id);}}bytevalue=value(id);assertvalue==d&&(id&initial)==1<<d:String.format("val = %d, id & initial = %d, d = %d",value,id&initial,d);setValue(id,unusable);// mark as unusableupdateParentsAlloc(id);returnid;}

1. 從根節點開始遍歷，如果當前節點的val<d，則通過id <<=1匹配下一層；

2、如果val > d，則表示存在子節點被分配的情況，而且剩餘節點的記憶體大小不夠，此時需要在兄弟節點上繼續查詢；

3、分配成功的節點需要標記為不可用，防止被再次分配，在memoryMap對應位置更新為12；

4、分配節點完成後，其父節點的狀態也需要更新，並可能引起更上一層父節點的更新，實現如下：

private void updateParentsAlloc(int id) { while (id > 1) { int parentId = id >>> 1; byte val1 = value(id); byte val2 = value(id ^ 1); byte val = val1 < val2 ? val1 : val2; setValue(parentId, val); id = parentId; } }

12345678910

privatevoidupdateParentsAlloc(intid){while(id>1){intparentId=id>>>1;byteval1=value(id);byteval2=value(id^1);byteval=val1<val2?val1:val2;setValue(parentId,val);id=parentId;}}

比如節點2048被分配出去，更新過程如下：

memoryMap節點更新

到目前為止，基於poolChunk的節點分配已經完成。

打賞支援我寫出更多好文章，謝謝！

打賞作者

打賞支援我寫出更多好文章，謝謝！