第Netty分布式高性能工具類異線程下回收對(duì)象解析就是創(chuàng)建對(duì)象和回收對(duì)象不在同一條線程的情況下,對(duì)象回收的邏輯

我們之前小節(jié)簡(jiǎn)單介紹過,異線程回收對(duì)象,是不會(huì)放在當(dāng)前線程的stack中的,而是放在一個(gè)WeakOrderQueue的數(shù)據(jù)結(jié)構(gòu)中,回顧我們之前的一個(gè)圖:

8-6-1

相關(guān)的邏輯,我們跟到源碼中:

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

publicvoidrecycle(Objectobject){

if(object!=value){

thrownewIllegalArgumentException(objectdoesnotbelongtohandle

stack.push(this);

}

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

voidpush(DefaultHandleitem){

ThreadcurrentThread=Thread.currentThread();

if(thread==currentThread){

pushNow(item);

}else{

pushLater(item,currentThread);

}

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

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

跟到pushLater方法中

privatevoidpushLater(DefaultHandleitem,Threadthread){

MapStack,WeakOrderQueuedelayedRecycled=DELAYED_RECYCLED.get();

WeakOrderQueuequeue=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);

}elseif(queue==WeakOrderQueue.DUMMY){

return;

queue.add(item);

}

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

我們跟到DELAYED_RECYCLED中:

privatestaticfinalFastThreadLocalMapStack,WeakOrderQueueDELAYED_RECYCLED=

newFastThreadLocalMapStack,WeakOrderQueue(){

@Override

protectedMapStack,WeakOrderQueueinitialValue(){

returnnewWeakHashMapStack,WeakOrderQueue

};

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

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

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

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

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

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

我們回到pushLater方法中:

繼續(xù)往下看:

WeakOrderQueuequeue=delayedRecycled.get(this)

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

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

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

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

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

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

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

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

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

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

跟到allocate方法中

staticWeakOrderQueueallocate(Stackstack,Threadthread){

returnreserveSpace(stack.availableSharedCapacity,LINK_CAPACITY)

newWeakOrderQueue(stack,thread):null;

}

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

再跟到reserveSpace方法中:

privatestaticbooleanreserveSpace(AtomicIntegeravailableSharedCapacity,intspace){

assertspace

for(;;){

intavailable=availableSharedCapacity.get();

if(availablespace){

returnfalse;

if(availableSharedCpareAndSet(available,available-space)){

returntrue;

}

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

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

而這16個(gè)元素,將由線程2緩存

回到pushLater方法中

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

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

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

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

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

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

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

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

8-6-2

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

privateWeakOrderQueue(Stackstack,Threadthread){

head=tail=newLink();

owner=newWeakReferenceThread(thread);

synchronized(stack){

next=stack.head;

stack.head=this;

availableSharedCapacity=stack.availableSharedCapacity;

}

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

簡(jiǎn)單看下link的類的定義

privatestaticfinalclassLinkextendsAtomicInteger{

privatefinalDefaultHandle[]elements=newDefaultHandle[LINK_CAPACITY];

privateintreadIndex;

privateLinknext;

}

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

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

next節(jié)點(diǎn)則指向下一個(gè)link

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

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

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

8-6-3

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

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

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

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

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

回到pushLater方法中

privatevoidpushLater(DefaultHandleitem,Threadthread){

MapStack,WeakOrderQueuedelayedRecycled=DELAYED_RECYCLED.get();

WeakOrderQueuequeue=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);

}elseif(queue==WeakOrderQueue.DUMMY){

return;

queue.add(item);

}

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

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

voidadd(DefaultHandlehandle){

handle.lastRecycledId=id;

Linktail=this.tail;

intwriteIndex;

if((writeIndex=tail.get())==LINK_CAPACITY){

if(!reserveSpace(availableSharedCapacity,LINK_CAPACITY)){

return;

this.tail=tail=tail.next=newLink();

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)建的時(shí)候,會(huì)為自增一個(gè)唯一的id

Linktail=t