public void recycle(Object object) {
    if (object != value) {
        throw new IllegalArgumentException("object does not belong to handle");
    }
    stack.push(this);
}

這部分我們并不陌生, 跟到push方法中:

void push(DefaultHandle<?> item) {
    Thread currentThread = Thread.currentThread();
    if (thread == currentThread) {
        pushNow(item);
    } else {
        pushLater(item, currentThread);
    }
}

上一小節(jié)分析過, 同線程會走到pushNow, 有關(guān)具體邏輯也進(jìn)行了分析

如果不是同線程, 則會走到pushLater方法, 傳入handle對象和當(dāng)前線程對象

跟到pushLater方法中

private void pushLater(DefaultHandle<?> item, Thread thread) {
    Map<Stack<?>, WeakOrderQueue> delayedRecycled = DELAYED_RECYCLED.get();
    WeakOrderQueue queue = delayedRecycled.get(this);
    if (queue == null) {
        if (delayedRecycled.size() >= maxDelayedQueues) {
            delayedRecycled.put(this, WeakOrderQueue.DUMMY);
            return;
        }
        if ((queue = WeakOrderQueue.allocate(this, thread)) == null) {
            return;
        }
        delayedRecycled.put(this, queue);
    } else if (queue == WeakOrderQueue.DUMMY) {
        return;
    }
    queue.add(item);
}

首先通過DELAYED_RECYCLED.get()獲取一個delayedRecycled對象

我們跟到DELAYED_RECYCLED中:

private static final FastThreadLocal<Map<Stack<?>, WeakOrderQueue>> DELAYED_RECYCLED =
        new FastThreadLocal<Map<Stack<?>, WeakOrderQueue>>() {
    @Override
    protected Map<Stack<?>, WeakOrderQueue> initialValue() {
        return new WeakHashMap<Stack<?>, WeakOrderQueue>();
    }
};

這里我們看到DELAYED_RECYCLED是一個FastThreadLocal對象, initialValue方法創(chuàng)建一個WeakHashMap對象, WeakHashMap是一個map, key為stack, value為我們剛才提到過的WeakOrderQueue

從中我們可以分析到, 每個線程都維護(hù)了一個WeakHashMap對象

WeakHashMap中的元素, 是一個stack和WeakOrderQueue的映射, 說明了不同的stack, 對應(yīng)不同的WeakOrderQueue

這里的映射關(guān)系可以舉個例子說明:

比如線程1創(chuàng)建了一個對象, 在線程3進(jìn)行了回收, 線程2創(chuàng)建了一個對象, 同樣也在線程3進(jìn)行了回收, 那么線程3對應(yīng)的WeakHashMap中就會有兩個元素:

線程1的stack和線程2的WeakOrderQueue, 線程2和stack和線程2的WeakOrderQueue

我們回到pushLater方法中:

繼續(xù)往下看:

WeakOrderQueue queue = delayedRecycled.get(this)

拿到了當(dāng)前線程的WeakHashMap對象delayedRecycled之后, 然后通過delayedRecycled創(chuàng)建對象的線程的stack, 拿到WeakOrderQueue

這里的this, 就是創(chuàng)建對象的那個線程所屬的stack, 這個stack是綁定在handle中的, 創(chuàng)建handle對象時候進(jìn)行的綁定

假設(shè)當(dāng)前線程是線程2, 創(chuàng)建handle的線程是線程1, 這里通過handle的stack拿到線程1的WeakOrderQueue

if (queue == null) 說明線程2沒有回收過線程1的對象, 則進(jìn)入if塊的邏輯:

首先看判斷 if (delayedRecycled.size() >= maxDelayedQueues)

delayedRecycled.size() 表示當(dāng)前線程回收其他創(chuàng)建對象的線程的線程個數(shù), 也就是有幾個其他的線程在當(dāng)前線程回收對象

maxDelayedQueues表示最多能回收的線程個數(shù), 這里如果朝超過這個值, 就表示當(dāng)前線程不能在回收其他線程的對象了

通過 delayedRecycled.put(this, WeakOrderQueue.DUMMY) 標(biāo)記, 創(chuàng)建對象的線程的stack, 所對應(yīng)的WeakOrderQueue不可用, DUMMY我們可以理解為不可用

如果沒有超過maxDelayedQueues, 則通過if判斷中的 WeakOrderQueue.allocate(this, thread) 這種方式創(chuàng)建一個WeakOrderQueue

allocate傳入this, 也就是創(chuàng)建對象的線程對應(yīng)的stack, 假設(shè)是線程1, thread就是當(dāng)前線程, 假設(shè)是線程2

跟到allocate方法中

static WeakOrderQueue allocate(Stack<?> stack, Thread thread) {
    return reserveSpace(stack.availableSharedCapacity, LINK_CAPACITY)
            ? new WeakOrderQueue(stack, thread) : null;
}

reserveSpace(stack.availableSharedCapacity, LINK_CAPACITY)表示線程1的stack還能不能分配LINK_CAPACITY個元素, 如果可以, 則直接通過new的方式創(chuàng)建一個WeakOrderQueue對象

再跟到reserveSpace方法中:

private static boolean reserveSpace(AtomicInteger availableSharedCapacity, int space) {
    assert space >= 0;
    for (;;) {
        int available = availableSharedCapacity.get();
        if (available < space) {
            return false;
        }
        if (availableSharedCapacity.compareAndSet(available, available - space)) {
            return true;
        }
    }
}

參數(shù)availableSharedCapacity表示線程1的stack允許外部線程給其緩存多少個對象, 之前我們分析過是16384, space默認(rèn)是16

方法中通過一個cas操作, 將16384減去16, 表示stack可以給其他線程緩存的對象數(shù)為16384-16

而這16個元素, 將由線程2緩存

回到pushLater方法中

創(chuàng)建之后通過 delayedRecycled.put(this, queue) 將stack和WeakOrderQueue進(jìn)行關(guān)聯(lián)

最后通過queue.add(item), 將創(chuàng)建的WeakOrderQueue添加一個handle

講解WeakOrderQueue之前, 我們首先了解下WeakOrderQueue的數(shù)據(jù)結(jié)構(gòu)

WeakOrderQueue維護(hù)了多個link, link之間是通過鏈表進(jìn)行連接, 每個link可以盛放16個handle,

我們剛才分析過, 在reserveSpace方法中將 stack.availableSharedCapacity-16 , 其實就表示了先分配16個空間放在link里, 下次回收的時候, 如果這16空間沒有填滿, 則可以繼續(xù)往里盛放

如果16個空間都已填滿, 則通過繼續(xù)添加link的方式繼續(xù)分配16個空間用于盛放handle

WeakOrderQueue和WeakOrderQueue之間也是通過鏈表進(jìn)行關(guān)聯(lián)

可以根據(jù)下圖理解上述邏輯:

8-6-2

根據(jù)以上思路, 我們跟到WeakOrderQueue的構(gòu)造方法中:

private WeakOrderQueue(Stack<?> stack, Thread thread) {
    head = tail = new Link();
    owner = new WeakReference<Thread>(thread);
    synchronized (stack) {
        next = stack.head;
        stack.head = this;
    }
    availableSharedCapacity = stack.availableSharedCapacity;
}

這里有個head和tail, 都指向一個link對象, 這里我們可以分析到, 其實在WeakOrderQueue中維護(hù)了一個鏈表, head分別代表頭結(jié)點和尾節(jié)點, 初始狀態(tài)下, 頭結(jié)點和尾節(jié)點都指向同一個節(jié)點

簡單看下link的類的定義

private static final class Link extends AtomicInteger {
    private final DefaultHandle&lt;?&gt;[] elements = new DefaultHandle[LINK_CAPACITY];
    private int readIndex;
    private Link next;
}

每次創(chuàng)建一個Link, 都會創(chuàng)建一個DefaultHandle類型的數(shù)組用于盛放DefaultHandle對象, 默認(rèn)大小是16個

readIndex是一個讀指針, 我們之后小節(jié)會進(jìn)行分析

next節(jié)點則指向下一個link

回到WeakOrderQueue的構(gòu)造方法中:

owner是對向前線程進(jìn)行一個包裝, 代表了當(dāng)前線程

接下來在一個同步塊中, 將當(dāng)前創(chuàng)建的WeakOrderQueue插入到stack指向的第一個WeakOrderQueue, 也就是stack的head屬性, 指向我們創(chuàng)建的WeakOrderQueue, 如圖所示

8-6-3

如果線程2創(chuàng)建一個和stack關(guān)聯(lián)的WeakOrderQueue, stack的head節(jié)點就就會指向線程2創(chuàng)建WeakOrderQueue

如果之后線程3也創(chuàng)建了一個和stack關(guān)聯(lián)的WeakOrderQueue, stack的head節(jié)點就會指向新創(chuàng)建的線程3的WeakOrderQueue

然后線程3的WeakOrderQueue再指向線程2的WeakOrderQueue

也就是無論哪個線程創(chuàng)建一個和同一個stack關(guān)聯(lián)的WeakOrderQueue的時候, 都插入到stack指向的WeakOrderQueue列表的頭部

這樣就可以將stack和其他線程釋放對象的容器WeakOrderQueue進(jìn)行綁定

回到pushLater方法中

private void pushLater(DefaultHandle<?> item, Thread thread) {
    Map<Stack<?>, WeakOrderQueue> delayedRecycled = DELAYED_RECYCLED.get();
    WeakOrderQueue queue = delayedRecycled.get(this);
    if (queue == null) {
        if (delayedRecycled.size() >= maxDelayedQueues) {
            delayedRecycled.put(this, WeakOrderQueue.DUMMY);
            return;
        }
        if ((queue = WeakOrderQueue.allocate(this, thread)) == null) {
            return;
        }
        delayedRecycled.put(this, queue);
    } else if (queue == WeakOrderQueue.DUMMY) {
        return;
    }
    queue.add(item);
}

根據(jù)之前分析的WeakOrderQueue的數(shù)據(jù)結(jié)構(gòu), 我們分析最后一步, 也就是WeakOrderQueue的add方法

我們跟進(jìn)WeakOrderQueue的add方法:

void add(DefaultHandle<?> handle) {
    handle.lastRecycledId = id;
    Link tail = this.tail;
    int writeIndex;
    if ((writeIndex = tail.get()) == LINK_CAPACITY) {
        if (!reserveSpace(availableSharedCapacity, LINK_CAPACITY)) {
            return;
        }
        this.tail = tail = tail.next = new Link();
        writeIndex = tail.get();
    }
    tail.elements[writeIndex] = handle;
    handle.stack = null;
    tail.lazySet(writeIndex + 1);
}

首先, 看 handle.lastRecycledId = id

lastRecycledId表示handle上次回收的id, 而id表示W(wǎng)eakOrderQueue的id, weakOrderQueue每次創(chuàng)建的時候, 會為自增一個唯一的id

Link tail = this.tail 表示拿到當(dāng)前WeakOrderQueue的中指向最后一個link的指針, 也就是尾指針

再看 if ((writeIndex = tail.get()) == LINK_CAPACITY)

tail.get()表示獲取當(dāng)前l(fā)ink中已經(jīng)填充元素的個數(shù), 如果等于16, 說明元素已經(jīng)填充滿

然后通過eserveSpace方法判斷當(dāng)前WeakOrderQueue是否還能緩存stack的對象, eserveSpace方法我們剛才已經(jīng)分析過, 會根據(jù)stack的屬性availableSharedCapacity-16的方式判斷還能否緩存stack的對象, 如果不能再緩存stack的對象, 則返回

如果還能繼續(xù)緩存, 則在創(chuàng)建一個link, 并將尾節(jié)點指向新創(chuàng)建的link, 并且原來尾節(jié)點的next的節(jié)點指向新創(chuàng)建的link

然后拿到當(dāng)前l(fā)ink的writeIndex, 也就是寫指針, 如果是新創(chuàng)建的link中沒有元素, writeIndex為0

之后將尾部的link的elements屬性, 也就是一個DefaultHandle類型的數(shù)組, 通過數(shù)組下標(biāo)的方式將第writeIndex個節(jié)點賦值為要回收的handle

然后將handle的stack屬性設(shè)置為null, 表示當(dāng)前handle不是通過stack進(jìn)行回收的

最后將tail節(jié)點的元素個數(shù)進(jìn)行+1, 表示下一次將從writeIndex+1的位置往里寫

以上就是異線程回收對象的邏輯

以上就是Netty分布式高性能工具類異線程下回收對象解析的詳細(xì)內(nèi)容，更多關(guān)于Netty分布式異線程回收對象的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: