Netty分布式高性能工具類異線程下回收對象分析

這篇文章主要介紹“Netty分布式高性能工具類異線程下回收對象分析”，在日常操作中，相信很多人在Netty分布式高性能工具類異線程下回收對象分析問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”Netty分布式高性能工具類異線程下回收對象分析”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

異線程回收對象

就是創建對象和回收對象不在同一條線程的情況下, 對象回收的邏輯

我們之前小節簡單介紹過, 異線程回收對象, 是不會放在當前線程的stack中的, 而是放在一個WeakOrderQueue的數據結構中, 回顧我們之前的一個圖:

8-6-1

相關的邏輯, 我們跟到源碼中:

首先從回收對象的入口方法開始, DefualtHandle的recycle方法:

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);
    }
}

上一小節分析過, 同線程會走到pushNow, 有關具體邏輯也進行了分析

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

跟到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方法創建一個WeakHashMap對象, WeakHashMap是一個map, key為stack, value為我們剛才提到過的WeakOrderQueue

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

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

這里的映射關系可以舉個例子說明:

比如線程1創建了一個對象, 在線程3進行了回收, 線程2創建了一個對象, 同樣也在線程3進行了回收, 那么線程3對應的WeakHashMap中就會有兩個元素:

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

我們回到pushLater方法中:

繼續往下看:

WeakOrderQueue queue = delayedRecycled.get(this)

拿到了當前線程的WeakHashMap對象delayedRecycled之后, 然后通過delayedRecycled創建對象的線程的stack, 拿到WeakOrderQueue

這里的this, 就是創建對象的那個線程所屬的stack, 這個stack是綁定在handle中的, 創建handle對象時候進行的綁定

假設當前線程是線程2, 創建handle的線程是線程1, 這里通過handle的stack拿到線程1的WeakOrderQueue

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

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

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

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

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

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

allocate傳入this, 也就是創建對象的線程對應的stack, 假設是線程1, thread就是當前線程, 假設是線程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的方式創建一個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;
        }
    }
}

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

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

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

回到pushLater方法中

創建之后通過 delayedRecycled.put(this, queue) 將stack和WeakOrderQueue進行關聯

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

講解WeakOrderQueue之前, 我們首先了解下WeakOrderQueue的數據結構

WeakOrderQueue維護了多個link, link之間是通過鏈表進行連接, 每個link可以盛放16個handle,

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

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

WeakOrderQueue和WeakOrderQueue之間也是通過鏈表進行關聯

可以根據下圖理解上述邏輯:

8-6-2

根據以上思路, 我們跟到WeakOrderQueue的構造方法中:

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中維護了一個鏈表, head分別代表頭結點和尾節點, 初始狀態下, 頭結點和尾節點都指向同一個節點

簡單看下link的類的定義

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

每次創建一個Link, 都會創建一個DefaultHandle類型的數組用于盛放DefaultHandle對象, 默認大小是16個

readIndex是一個讀指針, 我們之后小節會進行分析

next節點則指向下一個link

回到WeakOrderQueue的構造方法中:

owner是對向前線程進行一個包裝, 代表了當前線程

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

8-6-3

如果線程2創建一個和stack關聯的WeakOrderQueue, stack的head節點就就會指向線程2創建WeakOrderQueue

如果之后線程3也創建了一個和stack關聯的WeakOrderQueue, stack的head節點就會指向新創建的線程3的WeakOrderQueue

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

也就是無論哪個線程創建一個和同一個stack關聯的WeakOrderQueue的時候, 都插入到stack指向的WeakOrderQueue列表的頭部

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

回到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);
}

根據之前分析的WeakOrderQueue的數據結構, 我們分析最后一步, 也就是WeakOrderQueue的add方法

我們跟進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表示WeakOrderQueue的id, weakOrderQueue每次創建的時候, 會為自增一個唯一的id

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

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

tail.get()表示獲取當前link中已經填充元素的個數, 如果等于16, 說明元素已經填充滿

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

如果還能繼續緩存, 則在創建一個link, 并將尾節點指向新創建的link, 并且原來尾節點的next的節點指向新創建的link

然后拿到當前link的writeIndex, 也就是寫指針, 如果是新創建的link中沒有元素, writeIndex為0

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

然后將handle的stack屬性設置為null, 表示當前handle不是通過stack進行回收的

最后將tail節點的元素個數進行+1, 表示下一次將從writeIndex+1的位置往里寫

到此，關于“Netty分布式高性能工具類異線程下回收對象分析”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續學習更多相關知識，請繼續關注億速云網站，小編會繼續努力為大家帶來更多實用的文章！