Condition介紹
網(wǎng)站建設(shè)哪家好，找成都創(chuàng)新互聯(lián)公司！專注于網(wǎng)頁設(shè)計(jì)、網(wǎng)站建設(shè)、微信開發(fā)、小程序定制開發(fā)、集團(tuán)企業(yè)網(wǎng)站建設(shè)等服務(wù)項(xiàng)目。為回饋新老客戶創(chuàng)新互聯(lián)還提供了沙灣免費(fèi)建站歡迎大家使用！
上篇文章講了ReentrantLock的加鎖和釋放鎖的使用，這篇文章是對ReentrantLock的補(bǔ)充。ReentrantLock#newCondition()可以創(chuàng)建Condition，在ReentrantLock加鎖過程中可以利用Condition阻塞當(dāng)前線程并臨時釋放鎖，待另外線程獲取到鎖并在邏輯后通知阻塞線程"激活"。Condition常用在基于異步通信的同步機(jī)制實(shí)現(xiàn)中，比如dubbo中的請求和獲取應(yīng)答結(jié)果的實(shí)現(xiàn)。
常用方法
Condition中主要的方法有2個
這里的await()，signal()必須在ReentrantLock#lock()和ReentrantLock#unlock()之間調(diào)用。
Condition實(shí)現(xiàn)分析
Condition的實(shí)現(xiàn)也是利用AbstractQueuedSynchronizer隊(duì)列來實(shí)現(xiàn)，await()在被調(diào)用后先將當(dāng)前線程加入到等待隊(duì)列中，然后釋放鎖，最后阻塞當(dāng)前線程。signal()在被調(diào)用后會先獲取等待隊(duì)列中第一個節(jié)點(diǎn)，并將這個節(jié)點(diǎn)轉(zhuǎn)化成ReentrantLock中的節(jié)點(diǎn)并加入到同步阻塞隊(duì)列的結(jié)尾，這樣此節(jié)點(diǎn)的上個節(jié)點(diǎn)線程釋放鎖后會激活此節(jié)點(diǎn)線程取來獲取鎖。
await()方法源碼分析
await()源碼如下
重慶分公司
重慶分公司
public final void await() throws InterruptedException {
//判斷是否當(dāng)前線程是否被中斷中斷則拋出中斷異常
if (Thread.interrupted())
throw new InterruptedException();
//加入等待隊(duì)列
Node node = addConditionWaiter();
//釋放當(dāng)前線程鎖
int savedState = fullyRelease(node);
int interruptMode = 0;
//判斷是否在同步阻塞隊(duì)列，如果不在一直循環(huán)到被加入
while (!isOnSyncQueue(node)) {
//阻塞當(dāng)前線程
LockSupport.park(this);
//判斷是否被中斷
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
//獲取鎖，如果獲取中被中斷則設(shè)置中斷狀態(tài)
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
//清除等待隊(duì)列中被"激活"的節(jié)點(diǎn)
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
//如果當(dāng)前線程被中斷，處理中斷邏輯
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}主要分以下幾步
下面來看下addConditionWaiter()實(shí)現(xiàn)
private Node addConditionWaiter() {
//獲取等待隊(duì)列尾部節(jié)點(diǎn)
Node t = lastWaiter;
//如果尾部狀態(tài)不為CONDITION，如果已經(jīng)被"激活"，清理之，然后重新獲取尾部節(jié)點(diǎn)
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
//創(chuàng)建以當(dāng)前線程為基礎(chǔ)的節(jié)點(diǎn)，并將節(jié)點(diǎn)模式設(shè)置成CONDITION
Node node = new Node(Thread.currentThread(), Node.CONDITION);
//如果尾節(jié)點(diǎn)不存在，說明隊(duì)列為空，將頭節(jié)點(diǎn)設(shè)置成當(dāng)前節(jié)點(diǎn)
if (t == null)
firstWaiter = node;
//如果尾節(jié)點(diǎn)存在，將此節(jié)點(diǎn)設(shè)置成尾節(jié)點(diǎn)的下個節(jié)點(diǎn)
else
t.nextWaiter = node;
//將尾節(jié)點(diǎn)設(shè)置成當(dāng)前節(jié)點(diǎn)
lastWaiter = node;
return node;
}addConditionWaiter()的邏輯很簡單，就是創(chuàng)建以當(dāng)前線程為基礎(chǔ)的節(jié)點(diǎn)并把節(jié)點(diǎn)加入等待隊(duì)列的尾部待其他線程處理。
下面來看下fullyRelease(Node node)實(shí)現(xiàn)
final int fullyRelease(Node node) {
boolean failed = true;
try {
//獲取阻塞隊(duì)列中當(dāng)前線程節(jié)點(diǎn)的鎖狀態(tài)值
int savedState = getState();
//釋放當(dāng)前線程節(jié)點(diǎn)鎖
if (release(savedState)) {
failed = false;
return savedState;
} else {
throw new IllegalMonitorStateException();
}
} finally {
//釋放失敗講節(jié)點(diǎn)等待狀態(tài)設(shè)置成關(guān)閉
if (failed)
node.waitStatus = Node.CANCELLED;
}
}調(diào)用getState()先獲取阻塞隊(duì)列中當(dāng)前線程節(jié)點(diǎn)的鎖狀態(tài)值，這個值可能大于1表示多次重入，然后調(diào)用release(savedState)釋放所有鎖，如果釋放成功返回鎖狀態(tài)值。
下面來看下isOnSyncQueue(Node node)實(shí)現(xiàn)
final boolean isOnSyncQueue(Node node) {
//判斷當(dāng)前節(jié)點(diǎn)是否是CONDITION或者前置節(jié)點(diǎn)是否為空如果為空直接返回false
if (node.waitStatus == Node.CONDITION || node.prev == null)
return false;
//如果下個節(jié)點(diǎn)存在，則在同步阻塞隊(duì)列中返回true
if (node.next != null) // If has successor, it must be on queue
return true;
//遍歷查找當(dāng)前節(jié)點(diǎn)是否在同步阻塞隊(duì)列中
return findNodeFromTail(node);
}
private boolean findNodeFromTail(Node node) {
Node t = tail;
for (;;) {
if (t == node)
return true;
if (t == null)
return false;
t = t.prev;
}
}此方法的功能是查找當(dāng)前節(jié)點(diǎn)是否在同步阻塞隊(duì)列中，方法先是快速判斷，判斷不了再進(jìn)行遍歷查找。
下面看下checkInterruptWhileWaiting(Node node)實(shí)現(xiàn)
private int checkInterruptWhileWaiting(Node node) {
return Thread.interrupted() ?
(transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
0;
}
final boolean transferAfterCancelledWait(Node node) {
if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
enq(node);
return true;
}
while (!isOnSyncQueue(node))
Thread.yield();
return false;
}此方法在線程被激活后被調(diào)用，主要功能就是判斷被激活的線程是否被中斷。此方法會返回2種中斷狀態(tài)THROW_IE和REINTERRUPT，THROW_IE是調(diào)用signal()前被中斷返回，REINTERRUPT在調(diào)用signal()后被中斷返回。 此方法先判斷是否被標(biāo)記中斷，是的話再調(diào)用transferAfterCancelledWait(node)取判斷是那種中斷狀態(tài)，transferAfterCancelledWait(node)方法分2步
如果使用await()方法上面2步其實(shí)是沒什么作用其最后一定會返回false，因?yàn)閍wait()被激活只能調(diào)用 signal()方法，而signal()方法肯定已經(jīng)將節(jié)點(diǎn)加入到同步阻塞隊(duì)列中。所以以上邏輯是給await(long time, TimeUnit unit)等帶超時激活方法用的。
acquireQueued(node, savedState)方法再上一章節(jié)已經(jīng)講過這邊就不重復(fù)了，下面分析下unlinkCancelledWaiters()方法
private void unlinkCancelledWaiters() {
//獲取等待隊(duì)列頭節(jié)點(diǎn)
Node t = firstWaiter;
Node trail = null;
while (t != null) {
//獲取下個節(jié)點(diǎn)
Node next = t.nextWaiter;
//如果狀態(tài)不為CONDITION說明已經(jīng)加入阻塞隊(duì)列需要清理掉
if (t.waitStatus != Node.CONDITION) {
t.nextWaiter = null;
if (trail == null)
firstWaiter = next;
else
//獲取下個節(jié)點(diǎn)
trail.nextWaiter = next;
if (next == null)
lastWaiter = trail;
}
else
trail = t;
t = next;
}
}此方法就是從頭開始查找狀態(tài)不為CONDITION的節(jié)點(diǎn)并清理，狀態(tài)不為CONDITION節(jié)點(diǎn)說明此節(jié)點(diǎn)已經(jīng)加入到阻塞隊(duì)列，已經(jīng)不需要維護(hù)。
下面來看下reportInterruptAfterWait(interruptMode)方法
private void reportInterruptAfterWait(int interruptMode)
throws InterruptedException {
//如果是THROW_IE模式直接拋出異常
if (interruptMode == THROW_IE)
throw new InterruptedException();
//如果是REINTERRUPT模式標(biāo)記線程中斷由上層處理中斷
else if (interruptMode == REINTERRUPT)
selfInterrupt();
}此方法處理中斷邏輯。如果是THROW_IE模式直接拋出異常,如果是REINTERRUPT模式標(biāo)記線程中斷由上層處理中斷。
signal()方法源碼分析
signal()源碼如下
public final void signal() {
//是否當(dāng)前線程持有鎖
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
//通知"激活"頭節(jié)點(diǎn)線程
if (first != null)
doSignal(first);
}先調(diào)用isHeldExclusively()判斷鎖是否被當(dāng)前線程持有，然后檢查等待隊(duì)列是否為空，不為空就是可以取第一個節(jié)點(diǎn)調(diào)用doSignal(first)去"激活",這里激活不是真正的激活而只是將節(jié)點(diǎn)加入到同步阻塞隊(duì)列尾部,所以上下文中帶""的激活都是這種解釋。
下面看下isHeldExclusively()實(shí)現(xiàn)
protected final boolean isHeldExclusively() {
return getExclusiveOwnerThread() == Thread.currentThread();
}實(shí)現(xiàn)就是比較下當(dāng)前線程和持有鎖的線程是否同一個
下面看下doSignal(first)的實(shí)現(xiàn)
private void doSignal(Node first) {
do {
//頭指頭后移一位，如果后面的節(jié)點(diǎn)為空，則將尾指頭也指向空，說明隊(duì)列為空了
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
//清空頭節(jié)點(diǎn)的下個節(jié)點(diǎn)
first.nextWaiter = null;
//如果"激活"失敗者取下個繼續(xù)，直到成功或者遍歷完
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}此方法就是取當(dāng)前頭節(jié)點(diǎn)一直去嘗試"激活"，直到成功或者遍歷完。
下面來看下transferForSignal(first)方法
final boolean transferForSignal(Node node) {
//將CONDITION狀態(tài)設(shè)置成0
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
//加入到同步阻塞隊(duì)列
Node p = enq(node);
int ws = p.waitStatus;
//狀態(tài)異常直接激活
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}（1）此方法先先將CONDITION狀態(tài)設(shè)置成0，因?yàn)槿绻荂ONDITION狀態(tài)加入到同步阻塞隊(duì)列，激活的時候是不識別的。
（2）加入到同步阻塞隊(duì)列的尾部。所以同步阻塞隊(duì)列中前面如果有多個在排隊(duì)，調(diào)用unlock()不會馬上激活此節(jié)點(diǎn)。
（3）狀態(tài)異常直接調(diào)用unpark激活，這邊按理說如果狀態(tài)異常情況下激活，await()在調(diào)用unlock()被激活后會進(jìn)行相應(yīng)的異常處理，但看await()代碼沒有處理則是正常執(zhí)行。
這個方法主要就是把節(jié)點(diǎn)加入到同步阻塞隊(duì)列的，真正的激活則是調(diào)用unlock()去處理。
以上所述是小編給大家介紹的Java并發(fā)編程之Condition源碼分析詳解整合，希望對大家有所幫助，如果大家有任何疑問請給我留言，小編會及時回復(fù)大家的。在此也非常感謝大家對創(chuàng)新互聯(lián)網(wǎng)站的支持！
