點選上方“芋道原始碼”，選擇“置頂公眾號”

技術文章第一時間送達！

原始碼精品專欄

 

• 精盡 Dubbo 原理與原始碼專欄( 已經完成 69+ 篇，預計總共 75+ 篇 )

• 中文詳細註釋的開源專案

• Java 併發原始碼合集

• RocketMQ 原始碼合集

• Sharding-JDBC 原始碼解析合集

• Spring MVC 和 Security 原始碼合集

• MyCAT 原始碼解析合集

摘要: 原創出處 https://www.jianshu.com/p/d91060311437 「佔小狼」歡迎轉載，保留摘要，謝謝！

---

上一節中分析瞭如何在poolChunk中分配一塊大於pageSize的記憶體，但在實際應用中，存在很多分配小記憶體的情況，如果也佔用一個page，明顯很浪費。針對這種情況，Netty提供了PoolSubpage把poolChunk的一個page節點8k記憶體劃分成更小的記憶體段，透過對每個記憶體段的標記與清理標記進行記憶體的分配與釋放。

PoolSubpage

final class PoolSubpage<T> {
    // 當前page在chunk中的id
    private final int memoryMapIdx;
    // 當前page在chunk.memory的偏移量
    private final int runOffset;
    // page大小
    private final int pageSize;
    //透過對每一個二進位制位的標記來修改一段記憶體的佔用狀態
    private final long[] bitmap;

    PoolSubpage prev;
    PoolSubpage next;

    boolean doNotDestroy;
    // 該page切分後每一段的大小
    int elemSize;
    // 該page包含的段數量
    private int maxNumElems;
    private int bitmapLength;
    // 下一個可用的位置
    private int nextAvail;
    // 可用的段數量
    private int numAvail;
    ...
}

假設目前需要申請大小為4096的記憶體：

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

因為4096

private long allocateSubpage(int normCapacity) {
    // Obtain the head of the PoolSubPage pool that is owned by the PoolArena and synchronize on it.
    // This is need as we may add it back and so alter the linked-list structure.
    PoolSubpage head = arena.findSubpagePoolHead(normCapacity);
    synchronized (head) {
        int d = maxOrder; // subpages are only be allocated from pages i.e., leaves
        int id = allocateNode(d);
        if (id 0) {
            return id;
        }

        final PoolSubpage[] subpages = this.subpages;
        final int pageSize = this.pageSize;

        freeBytes -= pageSize;

        int subpageIdx = subpageIdx(id);
        PoolSubpage subpage = subpages[subpageIdx];
        if (subpage == null) {
            subpage = new PoolSubpage(head, this, id, runOffset(id), pageSize, normCapacity);
            subpages[subpageIdx] = subpage;
        } else {
            subpage.init(head, normCapacity);
        }
        return subpage.allocate();
    }
}

1、Arena負責管理PoolChunk和PoolSubpage；
2、allocateNode負責在二叉樹中找到匹配的節點，和poolChunk不同的是，只匹配葉子節點；
3、poolChunk中維護了一個大小為2048的poolSubpage陣列，分別對應二叉樹中2048個葉子節點，假設本次分配到節點2048，則取出poolSubpage陣列第一個元素subpage；
4、如果subpage為空，則進行初始化，並加入到poolSubpage陣列；

subpage初始化實現如下：

PoolSubpage(PoolSubpage head,
    PoolChunk chunk,
    int memoryMapIdx, int runOffset,
    int pageSize, elemSize) {

    this.chunk = chunk;
    this.memoryMapIdx = memoryMapIdx;
    this.runOffset = runOffset;
    this.pageSize = pageSize;
    bitmap = new long[pageSize >>> 10]; // pageSize / 16 / 64
    init(head, elemSize);
}

1、預設初始化bitmap長度為8，這裡解釋一下為什麼只需要8個元素：其中分配記憶體大小都是處理過的，最小為16，說明一個page可以分成8192/16 = 512個記憶體段，一個long有64位，可以描述64個記憶體段，這樣只需要512/64 = 8個long就可以描述全部記憶體段了。
2、init根據當前需要分配的記憶體大小，確定需要多少個bitmap元素，實現如下：

void init(PoolSubpage 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             bitmap[i] = 0;
        }
    }
    addToPool(head);
}

下麵透過分佈申請4096和32大小的記憶體，說明如何確定bitmapLength的值：

1. 比如，當前申請大小4096的記憶體，maxNumElems 和 numAvail 為2，說明一個page被拆分成2個記憶體段，2 >>> 6 = 0，且2 & 63 ！= 0，所以bitmapLength為1，說明只需要一個long就可以描述2個記憶體段狀態。

2. 如果當前申請大小32的記憶體，maxNumElems 和 numAvail 為 256，說明一個page被拆分成256個記憶體段， 256>>> 6 = 4，說明需要4個long描述256個記憶體段狀態。

下麵看看subpage是如何進行記憶體分配的：

long allocate() {
    if (elemSize == 0) {
        return toHandle(0);
    }

    if (numAvail == 0 || !doNotDestroy) {
        return -1;
    }

    final int bitmapIdx = getNextAvail();
    int q = bitmapIdx >>> 6;
    int r = bitmapIdx & 63;
    assert (bitmap[q] >>> r & 1) == 0;
    bitmap[q] |= 1L <
    if (-- numAvail == 0) {
        removeFromPool();
    }

    return toHandle(bitmapIdx);
}

1、方法getNextAvail負責找到當前page中可分配記憶體段的bitmapIdx；
2、q = bitmapIdx >>> 6，確定bitmap陣列下標為q的long數，用來描述 bitmapIdx 記憶體段的狀態；
3、bitmapIdx & 63將超出64的那一部分二進位制數抹掉，得到一個小於64的數r；
4、bitmap[q] |= 1L << r將對應位置q設定為1；

如果以上描述不直觀的話，下麵換一種方式解釋，假設需要分配大小為128的記憶體，這時page會拆分成64個記憶體段，需要1個long型別的數字描述這64個記憶體段的狀態，所以bitmap陣列只會用到第一個元素。

狀態轉換

getNextAvail如何實現找到下一個可分配的記憶體段？

private int getNextAvail() {
    int nextAvail = this.nextAvail;
    if (nextAvail >= 0) {
        this.nextAvail = -1;
        return nextAvail;
    }
    return findNextAvail();
}

1、如果nextAvail大於等於0，說明nextAvail指向了下一個可分配的記憶體段，直接傳回nextAvail值；
2、每次分配完成，nextAvail被置為-1，這時只能透過方法findNextAvail重新計算出下一個可分配的記憶體段位置。

private int findNextAvail() {
    final long[] bitmap = this.bitmap;
    final int bitmapLength = this.bitmapLength;
    for (int i = 0; i         long bits = bitmap[i];
        if (~bits != 0) {
            return findNextAvail0(i, bits);
        }
    }
    return -1;
}

private int findNextAvail0(int i, long bits) {
    final int maxNumElems = this.maxNumElems;
    final int baseVal = i <6;
    for (int j = 0; j 64; j ++) {
        if ((bits & 1) == 0) {
            int val = baseVal | j;
            if (val                 return val;
            } else {
                break;
            }
        }
        bits >>>= 1;
    }
    return -1;
}

1、~bits != 0說明這個long所描述的64個記憶體段還有未分配的；
2、(bits & 1) == 0 用來判斷該位置是否未分配，否則bits又移一位，從左到右遍歷值為0的位置；

至此，subpage記憶體段已經分配完成。

666. 彩蛋

---

---

如果你對 Dubbo 感興趣，歡迎加入我的知識星球一起交流。

目前在知識星球（https://t.zsxq.com/2VbiaEu）更新瞭如下 Dubbo 原始碼解析如下：

01. 除錯環境搭建
02. 專案結構一覽
03. 配置 Configuration
04. 核心流程一覽

05. 拓展機制 SPI

06. 執行緒池

07. 服務暴露 Export

08. 服務取用 Refer

09. 註冊中心 Registry

10. 動態編譯 Compile

11. 動態代理 Proxy

12. 服務呼叫 Invoke

13. 呼叫特性 

14. 過濾器 Filter

15. NIO 伺服器

16. P2P 伺服器

17. HTTP 伺服器

18. 序列化 Serialization

19. 叢集容錯 Cluster

20. 優雅停機

21. 日誌適配

22. 狀態檢查

23. 監控中心 Monitor

24. 管理中心 Admin

25. 運維命令 QOS

26. 鏈路追蹤 Tracing

…
一共 60 篇++

原始碼不易↓↓↓↓↓

點贊支援老艿艿↓↓