掃二維碼與項(xiàng)目經(jīng)理溝通
我們在微信上24小時期待你的聲音
解答本文疑問/技術(shù)咨詢/運(yùn)營咨詢/技術(shù)建議/互聯(lián)網(wǎng)交流
大家好，我是君哥。
為烏拉特后等地區(qū)用戶提供了全套網(wǎng)頁設(shè)計(jì)制作服務(wù)，及烏拉特后網(wǎng)站建設(shè)行業(yè)解決方案。主營業(yè)務(wù)為成都網(wǎng)站建設(shè)、網(wǎng)站設(shè)計(jì)、烏拉特后網(wǎng)站設(shè)計(jì)，以傳統(tǒng)方式定制建設(shè)網(wǎng)站，并提供域名空間備案等一條龍服務(wù)，秉承以專業(yè)、用心的態(tài)度為用戶提供真誠的服務(wù)。我們深信只要達(dá)到每一位用戶的要求，就會得到認(rèn)可，從而選擇與我們長期合作。這樣，我們也可以走得更遠(yuǎn)！
Java中 AQS 是 AbstractQueuedSynchronizer 類，AQS 依賴 FIFO 隊(duì)列來提供一個框架，這個框架用于實(shí)現(xiàn)鎖以及鎖相關(guān)的同步器，比如信號量、事件等。
在 AQS 中，主要有兩部分功能，一部分是操作 state 變量，第二部分是實(shí)現(xiàn)排隊(duì)和阻塞機(jī)制。
注意，AQS 并沒有實(shí)現(xiàn)任何同步接口，它只是提供了類似 acquireInterruptible 的方法，調(diào)用這些方法可以實(shí)現(xiàn)鎖和同步器。
Java 使用 MESA 管程模型來管理類的成員變量和方法，讓這個類的成員變量和方法的操作是線程安全的。下圖是 MESA 管程模型，里面除了定義共享變量外，還定義了條件變量和條件變量等待隊(duì)列：
上圖中有三個知識點(diǎn)：
Java 中的 MESA 管程模型有一點(diǎn)改進(jìn)，就是管程內(nèi)部只有一個條件變量和一個等待隊(duì)列。下圖是 AQS 的管程模型：
AQS 的管程模型依賴 AQS 中的 FIFO 隊(duì)列實(shí)現(xiàn)入口等待隊(duì)列，要進(jìn)入管程內(nèi)部，就由各種并發(fā)鎖的限制。而 ConditionObject 則實(shí)現(xiàn)了條件隊(duì)列，這個隊(duì)列可以創(chuàng)建多個。
下面就從入口等待隊(duì)列、并發(fā)鎖、條件等待隊(duì)列三個方面來帶你徹底理解 AQS。
獨(dú)占, 忽略 interrupts
public final void acquire(int arg){
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}這里的 tryAcquire 是抽象方法，由 AQS 的子類來實(shí)現(xiàn)，因?yàn)槊總€子類實(shí)現(xiàn)的鎖是不一樣的。
上面的代碼可以看到，獲取鎖失敗后，會先執(zhí)行 addWaiter 方法加入隊(duì)列，然后執(zhí)行 acquireQueued 方法自旋地獲取鎖直到成功。
addWaiter 代碼邏輯如下圖，簡單說就是把 node 入隊(duì)，入隊(duì)后返回 node 參數(shù)給 acquireQueued 方法：
acquireQueued 自旋獲取鎖邏輯如下圖：
這里有幾個細(xì)節(jié)：
（1）waitStatus
（2）獲取鎖失敗后掛起
如果前置節(jié)點(diǎn)不是頭節(jié)點(diǎn)，或者前置節(jié)點(diǎn)是頭節(jié)點(diǎn)但當(dāng)前節(jié)點(diǎn)獲取鎖失敗，這時當(dāng)前節(jié)點(diǎn)需要掛起，分三種情況：
（3）取消獲取鎖
如果獲取鎖拋出異常，則取消獲取鎖，如果當(dāng)前節(jié)點(diǎn)是 tail 節(jié)點(diǎn)，分兩種情況如下圖：
如果當(dāng)前節(jié)點(diǎn)不是 tail 節(jié)點(diǎn)，也分兩種情況，如下圖：
4.對中斷狀態(tài)忽略
5.如果前置節(jié)點(diǎn)的狀態(tài)是 0 或 PROPAGATE，會被當(dāng)前節(jié)點(diǎn)自旋過程中更新成 -1，以便之后通知當(dāng)前節(jié)點(diǎn)。
對應(yīng)方法 acquireInterruptibly(int arg)。
跟忽略中斷(acquire方法)不同的是要響應(yīng)中斷，下面兩個地方響應(yīng)中斷：
對應(yīng)方法 tryAcquireNanos(int arg, long nanosTimeout)。
這個方法具備了獨(dú)占 + 響應(yīng)中斷 + 超時的功能，下面2個地方要判斷是否超時：
獨(dú)占鎖釋放分兩步：釋放鎖，喚醒后繼節(jié)點(diǎn)。
釋放鎖的方法 tryRelease 是抽象的，由子類去實(shí)現(xiàn)。
我們看一下喚醒后繼節(jié)點(diǎn)的邏輯，首先需要滿足兩個條件：
之前我們講了獨(dú)占鎖，這一小節(jié)我們談共享鎖，有什么不同呢？
對應(yīng)方法 acquireShared，代碼如下：
public final void acquireShared(int arg){
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}這里獲取鎖使用的方法是 tryAcquireShared，獲取的是共享鎖。獲取共享鎖跟獲取獨(dú)占鎖不同的是，會返回一個整數(shù)值，說明如下:
怎么判斷隊(duì)列中等待節(jié)點(diǎn)是在等待共享鎖呢？nextWaiter == SHARED，這個參數(shù)值是入隊(duì)新建節(jié)點(diǎn)的時候構(gòu)造函數(shù)傳入的。
自旋過程中，如果獲取鎖成功(返回正數(shù))，首先把自己設(shè)置成新的 head 節(jié)點(diǎn)，然后把通知傳播下去。如下圖：
之后會喚醒后面節(jié)點(diǎn)并保證喚醒操作可以傳播下去。但是需要滿足四個條件中的一個：
喚醒后面節(jié)點(diǎn)的操作，其實(shí)就是釋放共享鎖，對應(yīng)方法是 doReleaseShared，見釋放共享鎖一節(jié)。
對應(yīng)方法 acquireSharedInterruptibly(int arg)。
跟共享忽略中斷(acquireShared 方法)不同的是要響應(yīng)中斷，下面兩個地方響應(yīng)中斷：
如果檢查到當(dāng)前線程已經(jīng)中斷，則拋出 InterruptedException，當(dāng)前線程退出。
對應(yīng)方法 tryAcquireSharedNanos(int arg, long nanosTimeout)。
這個方法具備了共享 + 響應(yīng)中斷 + 超時的功能，下面兩個個地方要判斷是否超時：
另外，park 線程的操作使用 parkNanos 傳入阻塞時間。
釋放共享鎖代碼如下：
public final boolean releaseShared(int arg){
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}首先嘗試釋放共享鎖，tryReleaseShared 代碼由子類來實(shí)現(xiàn)。釋放成功后執(zhí)行AQS中的 doReleaseShared 方法，是一個自旋操作。
自旋的條件是隊(duì)列中至少有兩個節(jié)點(diǎn)，這里分三種情況。
情況一：當(dāng)前節(jié)點(diǎn) waitStatus 是 -1，如下圖：
情況二：當(dāng)前節(jié)點(diǎn) waitStatus 是 0(被其他線程更新新成了中間狀態(tài))，如下圖：
情況三：當(dāng)前節(jié)點(diǎn) waitStatus 是 -3，為什么會這樣呢？需要解釋一下，head節(jié)點(diǎn)喚醒后繼節(jié)點(diǎn)之前 waitStatus 已經(jīng)被更新中間態(tài) 0 了，喚醒后繼節(jié)點(diǎn)動作還沒有執(zhí)行，又被其他線程更成了 -3，也就是其他線程釋放鎖執(zhí)行了上面情況二。這時需要先把 waitStatus 再更成 0 (在方法 unparkSuccessor)，如下圖：
上面的講解可以看出，如果要基于 AQS 來實(shí)現(xiàn)并發(fā)鎖，可以根據(jù)需求重寫下面四個方法來實(shí)現(xiàn)，這四個方法在 AQS 中沒有具體實(shí)現(xiàn)：
AQS 的子類需要重寫上面的方法來修改 state 值，并且定義獲取鎖或者釋放鎖時 state 值的變化。子類也可以定義自己的 state 變量，但是只有更新 AQS 中的 state變量才會對同步起作用。
還有一個判斷當(dāng)前線程是否持有獨(dú)占鎖的方法 isHeldExclusively，也可以供子類重寫后使用。
獲取/釋放鎖的具體實(shí)現(xiàn)放到下篇文章講解。
AQS 使用 FIFO 隊(duì)列實(shí)現(xiàn)了一個鎖相關(guān)的并發(fā)器模板，可以基于這個模板來實(shí)現(xiàn)各種鎖，包括獨(dú)占鎖、共享鎖、信號量等。
AQS 中，有一個核心狀態(tài)是 waitStatus，這個代表節(jié)點(diǎn)的狀態(tài)，決定了當(dāng)前節(jié)點(diǎn)的后續(xù)操作，比如是否等待喚醒，是否要喚醒后繼節(jié)點(diǎn)。
這一章節(jié)講解 Java AQS 中的并發(fā)鎖。其實(shí) Java AQS 中的并發(fā)鎖主要是基于 state 這個變量值來實(shí)現(xiàn)的。
我們先來看一下 UML 類圖：
從圖中可以看到，ReentrantLock 使用抽象內(nèi)部類 Sync 來實(shí)現(xiàn)了 AQS 的方法，然后基于 Sync 這個同步器實(shí)現(xiàn)了公平鎖和非公平鎖。主要實(shí)現(xiàn)了下面 3 個方法：
從實(shí)現(xiàn)的方法可以看到，ReentrantLock 中獲取的鎖是獨(dú)占鎖，我們再來看一下獲取和釋放獨(dú)占鎖的代碼：
public final void acquire(int arg){
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}獨(dú)占鎖的特點(diǎn)是調(diào)用上面 acquire 方法，傳入的參數(shù)是 1。
獲取鎖首先判斷同步狀態(tài)(state)的值。
這說明沒有線程占用鎖，當(dāng)前線程如果符合下面兩個條件，就可以獲取到鎖：
沒有前任節(jié)點(diǎn)，如下圖：
CAS 的方式更新 state 值(把 0 更新成 1)成功。如果獲取獨(dú)占鎖成功，會更新 AQS 中 exclusiveOwnerThread 為當(dāng)前線程，這個很容易理解。
這說明已經(jīng)有線程占有鎖，判斷占有鎖的線程是不是當(dāng)前線程，如下圖：
state += 1 值如果小于 0，會拋出異常。
如果獲取鎖失敗，則進(jìn)入 AQS 隊(duì)列等待喚醒。
跟公平鎖相比，非公平鎖的唯一不同是如果判斷到 state 等于 0，不用判斷有沒有前任節(jié)點(diǎn)，只要 CAS 設(shè)置 state 值(把 0 更新成 1)成功，就獲取到了鎖。
公平鎖和非公平鎖，釋放邏輯完全一樣，都是在內(nèi)部類 Sync 中實(shí)現(xiàn)的。釋放鎖需要注意兩點(diǎn)，如下圖：
公平鎖的特點(diǎn)是每個線程都要進(jìn)行排隊(duì)，不用擔(dān)心線程永遠(yuǎn)獲取不到鎖，但有個缺點(diǎn)是每個線程入隊(duì)后都需要阻塞和被喚醒，這一定程度上影響了效率。非公平鎖的特點(diǎn)是每個線程入隊(duì)前都會先嘗試獲取鎖，如果獲取成功就不會入隊(duì)了，這比公平鎖效率高。但也有一個缺點(diǎn)，隊(duì)列中的線程有可能等待很長時間，高并發(fā)下甚至可能永遠(yuǎn)獲取不到鎖。
我們先來看一下 UML 類圖：
從圖中可以看到，ReentrantReadWriteLock 使用抽象內(nèi)部類Sync來實(shí)現(xiàn)了 AQS 的方法，然后基于 Sync 這個同步器實(shí)現(xiàn)了公平鎖和非公平鎖。主要實(shí)現(xiàn)了下面 3 個方法：
下圖是定義的幾個常用變量：
下面這 2 個方法用戶獲取共享鎖和獨(dú)占鎖的數(shù)量：
static int sharedCount(int c){ return c >>> SHARED_SHIFT; }
static int exclusiveCount(int c){ return c & EXCLUSIVE_MASK; }從sharedCount 可以看到，共享鎖的數(shù)量要右移 16 位獲取，也就是說共享鎖占了高 16 位。從上圖 EXCLUSIVE_MASK 的定義看到，跟 EXCLUSIVE_MASK 進(jìn)行與運(yùn)算，得到的是低 16 位的值，所以獨(dú)占鎖占了低 16 位。如下圖：
這樣上面獲取鎖數(shù)量的方法就很好理解了。
讀鎖的實(shí)現(xiàn)對應(yīng)內(nèi)部類 ReadLock。
獲取讀鎖實(shí)際上是 ReadLock 調(diào)用了 AQS 的下面方法,傳入?yún)?shù)是 1：
public final void acquireShared(int arg){
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}ReentrantReadWriteLock 內(nèi)部類 Sync 實(shí)現(xiàn)了 tryAcquireShared 方法，主要包括如下三種情況：
a.當(dāng)前線程不需要阻塞(readerShouldBlock)。在公平鎖中，需要判斷是否有前置節(jié)點(diǎn)，如下圖就需要阻塞:
在非公平鎖中，則是判斷第一個節(jié)點(diǎn)是不是有獨(dú)占鎖，如下圖就需要阻塞:
b.使用 CAS 把 state 的值加 SHARED_UNIT(65536)。這里是不是就更理解讀鎖占高位的說法了，獲取一個讀鎖，state 的值就要加 SHARED_UNIT 這么多個。
c.給當(dāng)前線程的 holdCount 加 1。
ReentrantReadWriteLock 釋放讀鎖是在 ReadLock 中調(diào)用了 AQS 下面方法，傳入的參數(shù)是1：
public final boolean releaseShared(int arg){
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}ReentrantReadWriteLock 內(nèi)部類 Sync 實(shí)現(xiàn)了 releaseShared 方法，具體邏輯分為下面兩步：
寫鎖的實(shí)現(xiàn)對應(yīng)內(nèi)部類 WriteLock。
ReentrantReadWriteLock 獲取寫鎖其實(shí)是在 WriteLock 中調(diào)用了 AQS 的下面方法，傳入?yún)?shù) 1：
public final void acquire(int arg){
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}在ReentrantReadWriteLock 內(nèi)部類 Sync 實(shí)現(xiàn)了 tryAcquire 方法，首先獲取 state 值和獨(dú)占鎖數(shù)量(exclusiveCount),之后分如下兩種情況，如下圖：
1.state 不等于 0:
如果當(dāng)前線程不需要阻塞，并且給 state 賦值成功，使用 CAS 方式把 state 值加 1，把獨(dú)占線程置為當(dāng)前線程。
ReentrantReadWriteLock 釋放寫鎖其實(shí)是在 WriteLock 中調(diào)用了 AQS 的下面方法，傳入?yún)?shù) 1：
public final boolean release(int arg){
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}ReentrantReadWriteLock 在 Sync 中實(shí)現(xiàn)了 tryRelease(arg) 方法，邏輯如下：
我們先來看一下UML類圖：
從上面的圖中看出，CountDownLatch 的內(nèi)部類 Sync 實(shí)現(xiàn)了獲取共享鎖和釋放共享鎖的邏輯。
使用 CountDownLatch 時，構(gòu)造函數(shù)會傳入一個 int 類型的參數(shù) count，表示調(diào)動 count 次的 countDown 后主線程才可以被喚醒。
public CountDownLatch(int count){
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}上面的 Sync(count) 就是將 AQS 中的 state 賦值為 count。
CountDownLatch 的 await 方法調(diào)用了 AQS 中的 acquireSharedInterruptibly(int arg)，傳入?yún)?shù) 1，不過這個參數(shù)并沒有用。代碼如下：
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
Sync 中實(shí)現(xiàn)了 tryAcquireShared 方法，await 邏輯如下圖：
上面的自旋過程就是等待 state 的值不斷減小，只有 state 值成為 0 的時候，主線程才會跳出自旋執(zhí)行之后的邏輯。
CountDownLatch 的 countDown 方法調(diào)用了 AQS 的 releaseShared(int arg)，傳入?yún)?shù) 1，不過這個參數(shù)并沒有用。內(nèi)部類 Sync 實(shí)現(xiàn)了 tryReleaseShared 方法，邏輯如下圖：
CountDownLatch 的構(gòu)造函數(shù)入?yún)⒅禃x值給 state 變量，入隊(duì)操作是主線程入隊(duì)，每個子線程調(diào)用了countDown 后 state 值減 1，當(dāng) state 值成為 0 后喚醒主線程。
Semaphore 是一個信號量，用來保護(hù)共享資源。如果線程要訪問共享資源，首先從 Semaphore 獲取鎖(信號量)，如果信號量的計(jì)數(shù)器等于 0，則當(dāng)前線程進(jìn)入 AQS 隊(duì)列阻塞等待。否則，線程獲取鎖成功，信號量減 1。使用完共享資源后，釋放鎖(信號量加 1)。
Semaphore 跟管程模型不一樣的是，允許多個(構(gòu)造函數(shù)的 permits)線程進(jìn)入管程內(nèi)部，因此也常用它來做限流。
UML 類圖如下：
Semaphore的構(gòu)造函數(shù)會傳入一個int類型參數(shù)，用來初始化state的值。
獲取鎖的操作調(diào)用了 AQS 中的 acquireSharedInterruptibly 方法，傳入?yún)?shù) 1，代碼見 CountDownLatch 中 await 小節(jié)。Semaphore 在公平鎖和非公平鎖中分別實(shí)現(xiàn)了 tryAcquireShared 方法。
Semaphore 默認(rèn)使用非公平鎖，如果使用公平鎖，需要在構(gòu)造函數(shù)指定。獲取公平鎖邏輯比較簡單，如下圖：
acquire 在非公平的鎖唯一的區(qū)別就是不會判斷 AQS 隊(duì)列是否有前置節(jié)點(diǎn)(hasQueuedPredecessors)，而是直接嘗試獲取鎖。
除了 acquire 方法外，還有其他幾個獲取鎖的方法，原理類似，只是調(diào)用了 AQS 中的不同方法。
釋放鎖的操作調(diào)用了 AQS 中的 releaseShared(int arg) 方法，傳入?yún)?shù) 1，在內(nèi)部類 Sync 中實(shí)現(xiàn)了 tryReleaseShared 方法，邏輯很簡單：使用 CAS 的方式將 state 的值加 1，之后喚醒隊(duì)列中的后繼節(jié)點(diǎn)。
ThreadPoolExecutor 中也用到了 AQS，看下面的 UML 類圖：
Worker 主要在 ThreadPoolExecutor 中斷線程的時候使用。Worker 自己實(shí)現(xiàn)了獨(dú)占鎖，在中斷線程時首先進(jìn)行加鎖，中斷操作后釋放鎖。按照官方說法，這里不直接使用 ReentrantLock 的原因是防止調(diào)用控制線程池的方法(類似 setCorePoolSize)時能夠重新獲取到鎖，
使用 CAS 的方式把 AQS 中 state 從 0 改為 1，把當(dāng)前線程置為獨(dú)占線程。
把獨(dú)占線程置為空，把 AQS 中 state 改為 0。
Worker 初始化的時候會把 state 置為 -1，這樣是不能獲取鎖成功的。只有調(diào)用了 runWorker 方法，才會通過釋放鎖操作把 state 更為 0。這樣保證了只中斷運(yùn)行中的線程，而不會中斷等待中的線程。
AQS 基于雙向隊(duì)列實(shí)現(xiàn)了入口等待隊(duì)列，基于 state 變量實(shí)現(xiàn)了各種并發(fā)鎖，上篇文章講了入口等待隊(duì)列，而這篇文章主要講了基于 AQS 的并發(fā)鎖原理。
本章節(jié)主要講解管程模型中條件變量等待隊(duì)列。
首先我們看一下官方給出的示例代碼：
public class BoundedBuffer {
final Lock lock = new ReentrantLock();
final Condition notFull = lock.newCondition();
final Condition notEmpty = lock.newCondition();
final Object[] items = new Object[100];
int putptr, takeptr, count;
public void put(Object x) throws InterruptedException {
lock.lock();
try {
while (count == items.length)
notFull.await();
items[putptr] = x;
if (++putptr == items.length) putptr = 0;
++count;
notEmpty.signal();
} finally {
lock.unlock();
}
}
public Object take() throws InterruptedException {
lock.lock();
try {
while (count == 0)
notEmpty.await();
Object x = items[takeptr];
if (++takeptr == items.length) takeptr = 0;
--count;
notFull.signal();
return x;
} finally {
lock.unlock();
}
}
}這個代碼定義了兩個條件變量，notFull 和 notEmpty，說明如下：
Java AQS 的條件變量等待隊(duì)列是基于接口 Condition 和 ConditionObject 來實(shí)現(xiàn)的，URM 類圖如下：
Condition 接口主要定義了下面3個方法：
條件等待隊(duì)列跟入口等待隊(duì)列有兩個不同：
入隊(duì)方法對應(yīng)方法 addConditionWaiter，這里有三種情況：
可以看到，這種情況會從隊(duì)列第一個元素開始檢查 waitStatus 不是 -2 的元素，并從隊(duì)列中移除。
AQS 的并發(fā)鎖是基于 state 變量實(shí)現(xiàn)的，線程進(jìn)入條件等待隊(duì)列后，要釋放鎖，即 state 會變?yōu)?0，釋放操作會喚醒入口等待隊(duì)列中的線程。對應(yīng)方法 fullyRelease，返回值是釋放鎖減掉的 state 值 savedState。
釋放鎖后，線程阻塞，自旋等待被喚醒。
喚醒之后，當(dāng)前線程主要有四個動作：
上面提到了 interruptMode，這個屬性有三個值：
AQS 還提供了其他幾個 await 方法，如下：
喚醒條件等待隊(duì)列中的元素，首先判斷當(dāng)前線程是否持有獨(dú)占鎖，如果沒有，拋出異常。
喚醒條件隊(duì)列中的元素，會從第一個元素也就是 firstWaiter 開始，根據(jù) firstWaiter 的 waitStatus 是不是 -2，分兩種情況。
條件隊(duì)列第一個節(jié)點(diǎn)進(jìn)入入口等待隊(duì)列，等待獲取鎖，如下圖：
這里有兩個注意點(diǎn)：
如果重置 waitStatus 狀態(tài)失敗,則 unpark 節(jié)點(diǎn) firstWaiter。
如果 firstWaiter 的 waitStatus 不等于 -2，則查找 firstWaiter 的 nextWaiter，直到找到一個 waitStatus 等于 -2 的節(jié)點(diǎn)，然后將這個節(jié)點(diǎn)加入入口等待隊(duì)列隊(duì)尾，如下圖：
上面的兩種情況無論哪種，進(jìn)入入口等待隊(duì)列之前都要用 CAS 的方式把 waitStatus 改為 0。
理解了 signal 的邏輯，signalAll 的邏輯就非常容易理解了。首先判斷當(dāng)前線程是否持有獨(dú)占鎖，如果沒有，拋出異常。
將條件等待隊(duì)列中的所有節(jié)點(diǎn)依次加入入口等待隊(duì)列。如下圖：
Java 并發(fā)包下有很多類使用到了 AQS 中的 Condition，如下圖：
這里我們以 CyclicBarrier 為例來講解。CyclicBarrier 是讓一組線程相互等待共同達(dá)到一個屏障點(diǎn)。從 Cyclic 可以看出 Barrier 可以循環(huán)利用，也就是當(dāng)線程釋放之后可以繼續(xù)使用。
看下面這段示例代碼：
public static void main(String[] args){
CyclicBarrier cyclicBarrier = new CyclicBarrier(2, () -> {
System.out.println("柵欄中的線程執(zhí)行完成");
});
ExecutorService executorService = Executors.newFixedThreadPool(2);
executorService.submit(() -> {
try {
System.out.println("線程1：" + Thread.currentThread().getName());
cyclicBarrier.await();
} catch (Exception e) {
e.printStackTrace();
}
});
executorService.submit(() -> {
try {
System.out.println("線程2：" + Thread.currentThread().getName());
cyclicBarrier.await();
} catch (Exception e) {
e.printStackTrace();
}
});
executorService.shutdown();
}執(zhí)行結(jié)果：
線程1：pool-1-thread-1
線程2：pool-1-thread-2
柵欄中的線程執(zhí)行完成
CyclicBarrier 初始化的時候，會指定線程的數(shù)量 count，每個線程執(zhí)行完邏輯后，調(diào)用 CyclicBarrier 的 await 方法，這個方法首先將 count 減 1，然后調(diào)用 Condition的 await，讓當(dāng)前線程進(jìn)入條件等待隊(duì)列。當(dāng)最后一個線程將 count 減 1 后，count 數(shù)量等于 0，這時就會調(diào)用 Condition 的 signalAll 方法喚醒所有線程。
Java 的管程模型使用了 MESA 模型，基于 AQS 實(shí)現(xiàn)的 MESA 模型中，使用雙向隊(duì)列實(shí)現(xiàn)了入口等待隊(duì)列，使用變量 state 實(shí)現(xiàn)了并發(fā)鎖，使用 Condition 實(shí)現(xiàn)了條件等待隊(duì)列。
在 AQS 的實(shí)現(xiàn)中，使用同步隊(duì)列這個術(shù)語來表示雙向隊(duì)列，本文中使用入口等待隊(duì)列來描述是為了更好的配合管程模型來講解。
AQS 的 Condition 中，使用 await 方法將當(dāng)前線程放入條件變量等待隊(duì)列阻塞等待，使用 notify 來喚醒條件等待隊(duì)列中的線程，被喚醒之后，線程并不能立刻執(zhí)行，而是進(jìn)入入口等待隊(duì)列等待獲取鎖。
我們在微信上24小時期待你的聲音
解答本文疑問/技術(shù)咨詢/運(yùn)營咨詢/技術(shù)建議/互聯(lián)網(wǎng)交流
微信二維碼
微信二維碼
移動版官網(wǎng)