Netty分布式獲取異線程釋放對(duì)象源碼剖析

更新時(shí)間：2022年03月30日 11:06:59 作者：向南是個(gè)萬人迷

這篇文章主要為大家介紹了Netty分布式獲取異線程釋放的對(duì)象源碼剖析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步早日升職加薪

獲取異線程釋放對(duì)象

上一小節(jié)分析了異線程回收對(duì)象, 原理是通過與stack關(guān)聯(lián)的WeakOrderQueue進(jìn)行回收

如果對(duì)象經(jīng)過異線程回收之后, 當(dāng)前線程需要取出對(duì)象進(jìn)行二次利用, 如果當(dāng)前stack中為空, 則會(huì)通過當(dāng)前stack關(guān)聯(lián)的WeakOrderQueue進(jìn)行取出, 這也是這一小寫要分析的, 獲取異線程釋放的對(duì)象

在介紹之前我們首先看Stack類中的兩個(gè)屬性

private WeakOrderQueue cursor, prev;
private volatile WeakOrderQueue head;

這里都是指向WeakOrderQueue的指針, 其中head我們上一小節(jié)分析過, 指向最近創(chuàng)建的和stack關(guān)聯(lián)WeakOrderQueue, 也就是頭結(jié)點(diǎn)

cursor代表的是尋找的當(dāng)前WeakOrderQueue, pre則是cursor上一個(gè)節(jié)點(diǎn), 如圖所示:

8-7-1

我們從獲取對(duì)象的入口方法, handle的get開始分析

public final T get() {
    if (maxCapacityPerThread == 0) {
        return newObject((Handle<T>) NOOP_HANDLE);
    }
    Stack<T> stack = threadLocal.get();
    DefaultHandle<T> handle = stack.pop();
    if (handle == null) {
        handle = stack.newHandle();
        handle.value = newObject(handle);
    }
    return (T) handle.value;
}

這塊邏輯我們并不陌上, stack對(duì)象通過pop彈出一個(gè)handle

我們跟到pop方法中

DefaultHandle<T> pop() {
    int size = this.size;
    if (size == 0) {
        if (!scavenge()) {
            return null;
        }
        size = this.size;
    }
    size --;
    DefaultHandle ret = elements[size];
    elements[size] = null;
    if (ret.lastRecycledId != ret.recycleId) {
        throw new IllegalStateException("recycled multiple times");
    }
    ret.recycleId = 0;
    ret.lastRecycledId = 0;
    this.size = size;
    return ret;
}

這里我們重點(diǎn)關(guān)注, 如果size為空, 也就是當(dāng)前tack為空的情況下, 會(huì)走到scavenge方法, 這個(gè)方法, 就是從WeakOrderQueue獲取對(duì)象的方法

跟進(jìn)scavenge方法

boolean scavenge() {
    if (scavengeSome()) {
        return true;
    }
    prev = null;
    cursor = head;
    return false;
}

scavengeSome方法表示已經(jīng)回收到了對(duì)象, 則直接返回, 如果沒有回收到對(duì)象, 則將prev和cursor兩個(gè)指針進(jìn)行重置

繼續(xù)跟到scavengeSome方法中

boolean scavengeSome() {
    WeakOrderQueue cursor = this.cursor;
    if (cursor == null) {
        cursor = head;
        if (cursor == null) {
            return false;
        }
    }
    boolean success = false;
    WeakOrderQueue prev = this.prev;
    do {
        if (cursor.transfer(this)) {
            success = true;
            break;
        }
        WeakOrderQueue next = cursor.next;
        if (cursor.owner.get() == null) {
            if (cursor.hasFinalData()) {
                for (;;) {
                    if (cursor.transfer(this)) {
                        success = true;
                    } else {
                        break;
                    }
                }
            }
            if (prev != null) {
                prev.next = next;
            }
        } else {
            prev = cursor;
        }
        cursor = next;
    } while (cursor != null && !success);
    this.prev = prev;
    this.cursor = cursor;
    return success;
}

首先拿到cursor指針, cursor指針代表要回收的WeakOrderQueue

如果cursor為空, 則讓其指向頭節(jié)點(diǎn), 如果頭節(jié)點(diǎn)也空, 說明當(dāng)前stack沒有與其關(guān)聯(lián)的WeakOrderQueue, 則返回false

通過一個(gè)布爾值success標(biāo)記回收狀態(tài)

然后拿到pre指針, 也就是cursor的上一個(gè)節(jié)點(diǎn), 之后進(jìn)入一個(gè)do-while循環(huán)

do-while循環(huán)的終止條件是, 如果沒有遍歷到最后一個(gè)節(jié)點(diǎn)并且回收的狀態(tài)為false, 這里我們可以分析到再循環(huán)體里, 是不管遍歷與stack關(guān)聯(lián)的WeakOrderQueue, 直到彈出對(duì)象為止

跟到do-while循環(huán)中:

首先cursor指針會(huì)調(diào)用transfer方法, 該方法表示從當(dāng)前指針指向的WeakOrderQueue中將元素放入到當(dāng)前stack中, 如果取出成功則將success設(shè)置為true并跳出循環(huán), transfer我們稍后分析, 我們繼續(xù)往下看

如果沒有獲得元素, 則會(huì)通過next屬性拿到下一個(gè)WeakOrderQueue, 然后會(huì)進(jìn)入一個(gè)判斷 if (cursor.owner.get() == null)

owner屬性我們上一小節(jié)提到過, 就是與當(dāng)前WeakOrderQueue關(guān)聯(lián)的一個(gè)線程, get方法就是獲得關(guān)聯(lián)的線程對(duì)象, 如果這個(gè)對(duì)象為null說明該線程不存在, 則進(jìn)入if塊, 也就是一些清理的工作

if塊中又進(jìn)入一個(gè)判斷 if (cursor.hasFinalData()) , 這里表示當(dāng)前的WeakOrderQueue中是否還有數(shù)據(jù), 如果有數(shù)據(jù)則通過for循環(huán)將數(shù)據(jù)通過transfer方法傳輸?shù)疆?dāng)前stack中, 傳輸成功的, 將success標(biāo)記為true

transfer方法是將WeakOrderQueue中一個(gè)link中的handle往stack進(jìn)行傳輸, 有關(guān)link的相關(guān)內(nèi)容, 我們上一小節(jié)也進(jìn)行過分析

所以這里通過for循環(huán)將每個(gè)link的中的數(shù)據(jù)傳輸?shù)絪tack中

繼續(xù)往下看, 如果pre節(jié)點(diǎn)不為空, 則通過 prev.next = next 將cursor節(jié)點(diǎn)進(jìn)行釋放, 也就是pre的下一個(gè)節(jié)點(diǎn)指向cursor的下一個(gè)節(jié)點(diǎn)

繼續(xù)往下看else塊中的 prev = cursor

這里表示如果當(dāng)前線程還在, 則將prev賦值為cursor, 代表prev后移一個(gè)節(jié)點(diǎn)

最后通過cursor = next將cursor后移一位, 然后再繼續(xù)進(jìn)行循環(huán)

循環(huán)結(jié)束之后, 將stack的prev和cursor屬性進(jìn)行保存

我們跟到transfer方法中, 分析如何將WeakOrderQueue中的handle傳輸?shù)絪tack中:

boolean transfer(Stack<?> dst) {
    Link head = this.head;
    if (head == null) {
        return false;
    }
    if (head.readIndex == LINK_CAPACITY) {
        if (head.next == null) {
            return false;
        }
        this.head = head = head.next;
    }
    final int srcStart = head.readIndex;
    int srcEnd = head.get();
    final int srcSize = srcEnd - srcStart;
    if (srcSize == 0) {
        return false;
    }
    final int dstSize = dst.size;
    final int expectedCapacity = dstSize + srcSize;
    if (expectedCapacity > dst.elements.length) {
        final int actualCapacity = dst.increaseCapacity(expectedCapacity);
        srcEnd = min(srcStart + actualCapacity - dstSize, srcEnd);
    }
    if (srcStart != srcEnd) {
        final DefaultHandle[] srcElems = head.elements;
        final DefaultHandle[] dstElems = dst.elements;
        int newDstSize = dstSize;
        for (int i = srcStart; i < srcEnd; i++) {
            DefaultHandle element = srcElems[i];
            if (element.recycleId == 0) {
                element.recycleId = element.lastRecycledId;
            } else if (element.recycleId != element.lastRecycledId) {
                throw new IllegalStateException("recycled already");
            }
            srcElems[i] = null;
            if (dst.dropHandle(element)) {
                continue;
            }
            element.stack = dst;
            dstElems[newDstSize ++] = element;
        }
        if (srcEnd == LINK_CAPACITY && head.next != null) {
            reclaimSpace(LINK_CAPACITY);
            this.head = head.next;
        }
        head.readIndex = srcEnd;
        if (dst.size == newDstSize) {
            return false;
        }
        dst.size = newDstSize;
        return true;
    } else {
        return false;
    }
}

8-7-2

我們上一小節(jié)分析過, WeakOrderQueue是由多個(gè)link組成, 每個(gè)link通過鏈表的方式進(jìn)行關(guān)聯(lián), 其中head屬性指向第一個(gè)link, tail屬性指向最后一個(gè)link

在每個(gè)link中有多個(gè)handle

在link中維護(hù)了一個(gè)讀指針readIndex, 標(biāo)記著讀取link中handle的位置

我們繼續(xù)分析transfer方法

首先獲取頭結(jié)點(diǎn), 并判斷頭結(jié)點(diǎn)是否為空, 如果頭結(jié)點(diǎn)為空, 說明當(dāng)前WeakOrderQueue并沒有l(wèi)ink, 返回false

if (head.readIndex == LINK_CAPACITY) 這里判斷讀指針是否為16, 因?yàn)閘ink中元素最大數(shù)量就是16, 如果讀指針為16, 說明當(dāng)前l(fā)ink中的數(shù)據(jù)都被取走了

接著判斷 head.next == null , 表示是否還有下一個(gè)link, 如果沒有下一個(gè)link, 則說明當(dāng)前WeakOrderQueue沒有元素了, 則返回false

8-7-3

繼續(xù)往下看, 拿到head節(jié)點(diǎn)的讀指針和head中元素的數(shù)量, 接著計(jì)算可以傳輸元素的大小, 如果大小為0, 則返回false

8-7-4

接著, 拿到當(dāng)前stack的大小, 當(dāng)前stack大小加上可以傳輸?shù)拇笮”硎緎tack中所需要的容量

if (expectedCapacity > dst.elements.length) 表示如果需要的容量大于當(dāng)前stack中所維護(hù)的數(shù)組的大小, 則將stack中維護(hù)的數(shù)組進(jìn)行擴(kuò)容, 進(jìn)入if塊中

擴(kuò)容之后會(huì)返回actualCapacity, 表示擴(kuò)容之后的大小

再看 srcEnd = min(srcStart + actualCapacity - dstSize, srcEnd) 這步

srcEnd表示可以從Link中取的最后一個(gè)元素的下標(biāo)

srcStart + actualCapacity - dstSize 這里我們進(jìn)行一個(gè)拆分, actualCapacity - dstSize表示擴(kuò)容后大大小-原stack的大小, 也就是最多能往stack中傳輸多少元素

讀指針+可以往stack傳輸?shù)臄?shù)量, 可以表示往stack中傳輸?shù)淖詈笠粋€(gè)下標(biāo), 這里的下標(biāo)和srcEnd中取一個(gè)較小的值, 也就是既不能超過stack的容量, 也不能造成當(dāng)前l(fā)ink中下標(biāo)越界

繼續(xù)往下看

int newDstSize = dstSize 表示初始化stack的下標(biāo), 表示stack中從這個(gè)下標(biāo)開始添加數(shù)據(jù)

然后判斷 srcStart != srcEnd , 表示能不能同link中獲取內(nèi)容, 如果不能, 則返回false, 如果可以, 則進(jìn)入if塊中

接著拿到當(dāng)前l(fā)ink的數(shù)組elements和stack中的數(shù)組elements

然后通過for循環(huán), 通過數(shù)組下標(biāo)的方式不斷的將當(dāng)前l(fā)ink中的數(shù)據(jù)放入到stack中

for循環(huán)中首先拿到link的第i個(gè)元素

接著我們我們關(guān)注一個(gè)細(xì)節(jié)

if (element.recycleId == 0) {
    element.recycleId = element.lastRecycledId;
} else if (element.recycleId != element.lastRecycledId) {
    throw new IllegalStateException("recycled already");
}

這里 element.recycleId == 0 表示對(duì)象沒有被回收過, 如果沒有被回收過, 則賦值為lastRecycledId, 我們前面分析過lastRecycledId是WeakOrderQueue中的唯一下標(biāo), 通過賦值標(biāo)記element被回收過

然后繼續(xù)判斷 element.recycleId != element.lastRecycledId , 這表示該對(duì)象被回收過, 但是回收的recycleId卻不是最后一次回收lastRecycledId, 這是一種異常情況, 表示一個(gè)對(duì)象在不同的地方被回收過兩次, 這種情況則拋出異常

接著將link的第i個(gè)元素設(shè)置為null

繼續(xù)往下看:

if (dst.dropHandle(element)) {
    continue;
}

這里表示控制回收站回收的頻率, 之前的小節(jié)我們分析過, 這里不再贅述

element.stack = dst 表示將handle的stack屬性設(shè)置到當(dāng)前stack

dstElems[newDstSize ++] = element 這里通過數(shù)組的下標(biāo)的方式將link中的handle賦值到stack的數(shù)組中

繼續(xù)往下看:

if (srcEnd == LINK_CAPACITY && head.next != null) {
    reclaimSpace(LINK_CAPACITY);
    this.head = head.next;
}

這里的if表循環(huán)結(jié)束后, 如果link中的數(shù)據(jù)已經(jīng)回收完畢, 并且還有下一個(gè)節(jié)點(diǎn)則會(huì)進(jìn)到reclaimSpace方法

我們跟到reclaimSpace方法

private void reclaimSpace(int space) {
    assert space >= 0;
    availableSharedCapacity.addAndGet(space);
}

這里將availableSharedCapacity加上16, 表示W(wǎng)eakOrderQueue還可以繼續(xù)插入link

繼續(xù)看transfer方法

this.head = head.next 表示將head節(jié)點(diǎn)后移一個(gè)元素

head.readIndex = srcEnd 表示將讀指針指向srcEnd, 下一次讀取可以從srcEnd開始

if (dst.size == newDstSize) 表示沒有向stack傳輸任何對(duì)象, 則返回false

否則就通過 dst.size = newDstSize 更新stack的大小為newDstSize, 并返回true

以上就是從link中往stack中傳輸數(shù)據(jù)的過程