JUC.Condition學習筆記[附詳細源碼解析]

目錄

Condition的概念

大體實現流程

　　I.初始化狀態
　　II.await()操作
　　III.signal()操作

3個主要方法

　　Condition的數據結構
　　線程何時阻塞和釋放
　　await()方法
　　signal()和signalAll()方法

Condition示例：生產者和消費者

---

JUC提供了Lock可以方便的進行鎖操作，但是有時候我們也需要對線程進行條件性的阻塞和喚醒，這時我們就需要condition條件變量，它就像是在線程上加了多個開關，可以方便的對持有鎖的線程進行阻塞和喚醒。

Condition的概念

Condition主要是為了在J.U.C框架中提供和Java傳統的監視器風格的wait，notify和notifyAll方法類似的功能。 JDK的官方解釋如下： 條件（也稱為條件隊列 或條件變量）為線程提供了一個含義，以便在某個狀態條件現在可能為 true 的另一個線程通知它之前，一直掛起該線程（即讓其“等待”）。因為訪問此共享狀態信息發生在不同的線程中，所以它必須受保護，因此要將某種形式的鎖與該條件相關聯。等待提供一個條件的主要屬性是：以原子方式 釋放相關的鎖，并掛起當前線程，就像 Object.wait 做的那樣。 Condition實質上是被綁定到一個鎖上。 在JUC鎖機制(Lock)學習筆記中，我們了解到AQS有一個隊列，同樣Condition也有一個等待隊列，兩者是相對獨立的隊列，因此一個Lock可以有多個Condition，Lock(AQS)的隊列主要是阻塞線程的，而Condition的隊列也是阻塞線程，但是它是有阻塞和通知解除阻塞的功能 Condition阻塞時會釋放Lock的鎖，阻塞流程請看下面的Condition的await()方法。

大體實現流程

AQS等待隊列與Condition隊列是兩個相互獨立的隊列 await()就是在當前線程持有鎖的基礎上釋放鎖資源，并新建Condition節點加入到Condition的隊列尾部，阻塞當前線程 signal()就是將Condition的頭節點移動到AQS等待節點尾部，讓其等待再次獲取鎖 以下是AQS隊列和Condition隊列的出入結點的示意圖，可以通過這幾張圖看出線程結點在兩個隊列中的出入關系和條件。 I.初始化狀態：AQS等待隊列有3個Node，Condition隊列有1個Node(也有可能1個都沒有) II.節點1執行Condition.await()
1.將head后移 2.釋放節點1的鎖并從AQS等待隊列中移除 3.將節點1加入到Condition的等待隊列中 4.更新lastWaiter為節點1 III.節點2執行signal()操作 5.將firstWaiter后移 6.將節點4移出Condition隊列 7.將節點4加入到AQS的等待隊列中去 8.更新AQS的等待隊列的tail

3個主要方法

Condition的數據結構

我們知道一個Condition可以在多個地方被await()，那么就需要一個FIFO的結構將這些Condition串聯起來，然后根據需要喚醒一個或者多個（通常是所有）。所以在Condition內部就需要一個FIFO的隊列。 private transient Node firstWaiter; private transient Node lastWaiter; 上面的兩個節點就是描述一個FIFO的隊列。我們再結合前面提到的節點（Node）數據結構。我們就發現Node.nextWaiter就派上用場了！nextWaiter就是將一系列的Condition.await*串聯起來組成一個FIFO的隊列。

線程何時阻塞和釋放

阻塞：await()方法中，在線程釋放鎖資源之后，如果節點不在AQS等待隊列，則阻塞當前線程，如果在等待隊列，則自旋等待嘗試獲取鎖 釋放：signal()后，節點會從condition隊列移動到AQS等待隊列，則進入正常鎖的獲取流程

await方法

ReentrantLock是獨占鎖，一個線程拿到鎖后如果不釋放，那么另外一個線程肯定是拿不到鎖，所以在lock.lock()和lock.unlock()之間可能有一次釋放鎖的操作（同樣也必然還有一次獲取鎖的操作）。在進入lock.lock()后唯一可能釋放鎖的操作就是await()了。也就是說await()操作實際上就是釋放鎖，然后掛起線程，一旦條件滿足就被喚醒，再次獲取鎖！ Java Code?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

?

public?final?void?await()?throws?InterruptedException?{ ????//?1.如果當前線程被中斷，則拋出中斷異常 ????if?(Thread.interrupted()) ????????throw?new?InterruptedException(); ????//?2.將節點加入到Condition隊列中去，這里如果lastWaiter是cancel狀態，那么會把它踢出Condition隊列。 ????Node?node?=?addConditionWaiter(); ????//?3.調用tryRelease，釋放當前線程的鎖 ????long?savedState?=?fullyRelease(node); ????int?interruptMode?=?0; ????//?4.為什么會有在AQS的等待隊列的判斷？ ????// 解答：signal操作會將Node從Condition隊列中拿出并且放入到等待隊列中去，在不在AQS等待隊列就看signal是否執行了 ????// 如果不在AQS等待隊列中，就park當前線程，如果在，就退出循環，這個時候如果被中斷，那么就退出循環 ????while?(!isOnSyncQueue(node))?{ ????????LockSupport.park(this); ????????if?((interruptMode?=?checkInterruptWhileWaiting(node))?!=?0) ????????????break; ????} ????//?5.這個時候線程已經被signal()或者signalAll()操作給喚醒了，退出了4中的while循環 ????// 自旋等待嘗試再次獲取鎖，調用acquireQueued方法 ????if?(acquireQueued(node,?savedState)?&&?interruptMode?!=?THROW_IE) ????????interruptMode?=?REINTERRUPT; ????if?(node.nextWaiter?!=?null) ????????unlinkCancelledWaiters(); ????if?(interruptMode?!=?0) ????????reportInterruptAfterWait(interruptMode); }

整個await的過程如下： 1.將當前線程加入Condition鎖隊列。特別說明的是，這里不同于AQS的隊列，這里進入的是Condition的FIFO隊列。進行2。 2.釋放鎖。這里可以看到將鎖釋放了，否則別的線程就無法拿到鎖而發生死鎖。進行3。 3.自旋(while)掛起，直到被喚醒或者超時或者CACELLED等。進行4。 4.獲取鎖(acquireQueued)。并將自己從Condition的FIFO隊列中釋放，表明自己不再需要鎖（我已經拿到鎖了）。 可以看到，這個await的操作過程和Object.wait()方法是一樣，只不過await()采用了Condition隊列的方式實現了Object.wait()的功能。

signal和signalAll方法

await*()清楚了，現在再來看signal/signalAll就容易多了。按照signal/signalAll的需求，就是要將Condition.await*()中FIFO隊列中第一個Node喚醒（或者全部Node）喚醒。盡管所有Node可能都被喚醒，但是要知道的是仍然只有一個線程能夠拿到鎖，其它沒有拿到鎖的線程仍然需要自旋等待，就上上面提到的第4步(acquireQueued)。?

Java Code?

1 2 3 4 5 6 7

?

public?final?void?signal()?{ ????if?(!isHeldExclusively()) ????????throw?new?IllegalMonitorStateException(); ????Node?first?=?firstWaiter; ????if?(first?!=?null) ????????doSignal(first); }

這里先判斷當前線程是否持有鎖，如果沒有持有，則拋出異常，然后判斷整個condition隊列是否為空，不為空則調用doSignal方法來喚醒線程，看看doSignal方法都干了一些什么：

Java Code?

1 2 3 4 5 6 7 8

?

private?void?doSignal(Node?first)?{ ????do?{ ????????if?(?(firstWaiter?=?first.nextWaiter)?==?null) ????????????lastWaiter?=?null; ????????first.nextWaiter?=?null; ????}?while?(!transferForSignal(first)?&& ?????????????(first?=?firstWaiter)?!=?null); }

上面的代碼很容易看出來，signal就是喚醒Condition隊列中的第一個非CANCELLED節點線程，而signalAll就是喚醒所有非CANCELLED節點線程。當然了遇到CANCELLED線程就需要將其從FIFO隊列中剔除。 Java Code?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

?

final?boolean?transferForSignal(Node?node)?{ ????/* ?????*?設置node的waitStatus：Condition->0 ?????*/ ????if?(!compareAndSetWaitStatus(node,?Node.CONDITION,?0)) ????????return?false; ????/* ?????*?加入到AQS的等待隊列，讓節點繼續獲取鎖 ???? *?設置前置節點狀態為SIGNAL ?????*/ ????Node?p?=?enq(node); ????int?c?=?p.waitStatus; ????if?(c?>?0?||?!compareAndSetWaitStatus(p,?c,?Node.SIGNAL)) ????????LockSupport.unpark(node.thread); ????return?true; }

上面就是喚醒一個await*()線程的過程，根據前面的介紹，如果要unpark線程，并使線程拿到鎖，那么就需要線程節點進入AQS的隊列。所以可以看到在LockSupport.unpark之前調用了enq(node)操作，將當前節點加入到AQS隊列。 signalAll和signal方法類似，主要的不同在于它不是調用doSignal方法，而是調用doSignalAll方法： Java Code?

1 2 3 4 5 6 7 8 9

?

private?void?doSignalAll(Node?first)?{ ????lastWaiter?=?firstWaiter??=?null; ????do?{ ????????Node?next?=?first.nextWaiter; ????????first.nextWaiter?=?null; ????????transferForSignal(first); ????????first?=?next; ????}?while?(first?!=?null); }

這個方法就相當于把Condition隊列中的所有Node全部取出插入到等待隊列中去。

Condition應用示例：生產者和消費者

Condition 實例實質上被綁定到一個鎖上。要為特定 Lock 實例獲得 Condition 實例，請使用其 newCondition() 方法。在最后我們來看一個應用示例

Java Code?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72

?

/** ?*?生產者、消費者示例 ?*/ public?class?ConditionTest?{ ????private?int?storage; ????private?int?putCounter; ????private?int?getCounter; ????private?Lock?lock?=?new?ReentrantLock(); ????private?Condition?putCondition?=?lock.newCondition(); ????private?Condition?getCondition?=?lock.newCondition(); ????public?void?put()?throws?InterruptedException?{ ????????try?{ ????????????lock.lock(); ????????????if?(storage?>?0)?{ ????????????????putCondition.await(); ????????????} ????????????storage++; ????????????System.out.println("put?=>?"?+?++putCounter?); ????????????getCondition.signal(); ????????}?finally?{ ????????????lock.unlock(); ????????} ????} ????public?void?get()?throws?InterruptedException?{ ????????try?{ ????????????lock.lock(); ????????????lock.lock(); ????????????if?(storage?<=?0)?{ ????????????????getCondition.await(); ????????????} ????????????storage--; ????????????System.out.println("get??=>?"?+?++getCounter); ????????????putCondition.signal(); ????????}?finally?{ ????????????lock.unlock(); ????????????lock.unlock(); ????????} ????} ????public?class?PutThread?extends?Thread?{ ????????@Override ????????public?void?run()?{ ????????????for?(int?i?=?0;?i?<?100;?i++)?{ ????????????????try?{ ????????????????????put(); ????????????????}?catch?(InterruptedException?e)?{ ????????????????} ????????????} ????????} ????} ????public?class?GetThread?extends?Thread?{ ????????@Override ????????public?void?run()?{ ????????????for?(int?i?=?0;?i?<?100;?i++)?{ ????????????????try?{ ????????????????????get(); ????????????????}?catch?(InterruptedException?e)?{ ????????????????} ????????????} ????????} ????} ????public?static?void?main(String[]?args)?{ ????????final?ConditionTest?test?=?new?ConditionTest(); ????????Thread?put?=?test.new?PutThread(); ????????Thread?get?=?test.new?GetThread(); ????????put.start(); ????????get.start(); ????}

原創文章，請注明引用來源：CM4J 參考文章列表： http://www.goldendoc.org/2011/06/juc_condition/ http://www.blogjava.net/xylz/archive/2010/07/08/325540.html posted on 2013-08-28 17:19 CM4J 閱讀(...) 評論(...) 編輯 收藏

轉載于:https://www.cnblogs.com/cm4j/p/juc_condition.html

總結

以上是生活随笔為你收集整理的JUC.Condition学习笔记[附详细源码解析]的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。