日韩无码专区无码一级三级片|91人人爱网站中日韩无码电影|厨房大战丰满熟妇|AV高清无码在线免费观看|另类AV日韩少妇熟女|中文日本大黄一级黄色片|色情在线视频免费|亚洲成人特黄a片|黄片wwwav色图欧美|欧亚乱色一区二区三区

RELATEED CONSULTING
相關(guān)咨詢
選擇下列產(chǎn)品馬上在線溝通
服務(wù)時間：8:30-17:00
你可能遇到了下面的問題
關(guān)閉右側(cè)工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
為了帶你精通JavaAQS，我畫了40張圖，從管程模型講起！

大家好，我是君哥。

為烏拉特后等地區(qū)用戶提供了全套網(wǎng)頁設(shè)計制作服務(wù)，及烏拉特后網(wǎng)站建設(shè)行業(yè)解決方案。主營業(yè)務(wù)為成都網(wǎng)站建設(shè)、網(wǎng)站設(shè)計、烏拉特后網(wǎng)站設(shè)計，以傳統(tǒng)方式定制建設(shè)網(wǎng)站，并提供域名空間備案等一條龍服務(wù)，秉承以專業(yè)、用心的態(tài)度為用戶提供真誠的服務(wù)。我們深信只要達到每一位用戶的要求，就會得到認可，從而選擇與我們長期合作。這樣，我們也可以走得更遠！

Java中 AQS 是 AbstractQueuedSynchronizer 類，AQS 依賴 FIFO 隊列來提供一個框架，這個框架用于實現(xiàn)鎖以及鎖相關(guān)的同步器，比如信號量、事件等。

在 AQS 中，主要有兩部分功能，一部分是操作 state 變量，第二部分是實現(xiàn)排隊和阻塞機制。

注意，AQS 并沒有實現(xiàn)任何同步接口，它只是提供了類似 acquireInterruptible 的方法，調(diào)用這些方法可以實現(xiàn)鎖和同步器。

1、管程模型

Java 使用 MESA 管程模型來管理類的成員變量和方法，讓這個類的成員變量和方法的操作是線程安全的。下圖是 MESA 管程模型，里面除了定義共享變量外，還定義了條件變量和條件變量等待隊列：

上圖中有三個知識點：

  • MESA 管程模型封裝了共享變量和對共享變量的操作，線程要進入管程內(nèi)部，必須獲取到鎖，如果獲取鎖失敗就進入入口等待隊列阻塞等待。
  • 如果線程獲取到鎖，就進入到管程內(nèi)部。但是進入到管程內(nèi)部，也不一定能立刻操作共享變量，而是要看條件變量是否滿足，如果不滿足，只能進入條件變量等待隊列阻塞等待。
  • 在條件變量等待隊列中，如果被其他線程喚醒，也不一定能立刻操作共享變量，而是需要去入口等待隊列重新排隊等待獲取鎖。

Java 中的 MESA 管程模型有一點改進，就是管程內(nèi)部只有一個條件變量和一個等待隊列。下圖是 AQS 的管程模型：

AQS 的管程模型依賴 AQS 中的 FIFO 隊列實現(xiàn)入口等待隊列，要進入管程內(nèi)部，就由各種并發(fā)鎖的限制。而 ConditionObject 則實現(xiàn)了條件隊列，這個隊列可以創(chuàng)建多個。

下面就從入口等待隊列、并發(fā)鎖、條件等待隊列三個方面來帶你徹底理解 AQS。

2、入口等待隊列

2.1 獲取獨占鎖

獨占, 忽略 interrupts

public final void acquire(int arg){
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

這里的 tryAcquire 是抽象方法，由 AQS 的子類來實現(xiàn)，因為每個子類實現(xiàn)的鎖是不一樣的。

2.1.1 入隊

上面的代碼可以看到，獲取鎖失敗后，會先執(zhí)行 addWaiter 方法加入隊列，然后執(zhí)行 acquireQueued 方法自旋地獲取鎖直到成功。

addWaiter 代碼邏輯如下圖，簡單說就是把 node 入隊，入隊后返回 node 參數(shù)給 acquireQueued 方法：

這里有一個點需要注意，如果隊列為空，則新建一個 Node 作為隊頭。

2.1.2 入隊后獲取鎖

acquireQueued 自旋獲取鎖邏輯如下圖：

這里有幾個細節(jié)：

（1）waitStatus

  • CANCELLED(1)：當前節(jié)點取消獲取鎖。當?shù)却瑫r或被中斷(響應(yīng)中斷)，會觸發(fā)變更為此狀態(tài)，進入該狀態(tài)后節(jié)點狀態(tài)不再變化；
  • SIGNAL(-1)：后面節(jié)點等待當前節(jié)點喚醒；
  • CONDITION(-2)：Condition 中使用，當前線程阻塞在 Condition，如果其他線程調(diào)用了 Condition 的 signal 方法，這個結(jié)點將從等待隊列轉(zhuǎn)移到同步隊列隊尾，等待獲取同步鎖；
  • PROPAGATE(-3)：共享模式，前置節(jié)點喚醒后面節(jié)點后，喚醒操作無條件傳播下去；
  • 0：中間狀態(tài)，當前節(jié)點后面的節(jié)點已經(jīng)喚醒，但是當前節(jié)點線程還沒有執(zhí)行完成。

（2）獲取鎖失敗后掛起

如果前置節(jié)點不是頭節(jié)點，或者前置節(jié)點是頭節(jié)點但當前節(jié)點獲取鎖失敗，這時當前節(jié)點需要掛起，分三種情況：

  • 前置節(jié)點 waitStatus=-1，如下圖：

  • 前置節(jié)點 waitStatus > 0，如下圖：

  • 前置節(jié)點 waitStatus < 0 但不等于 -1，如下圖：

（3）取消獲取鎖

如果獲取鎖拋出異常，則取消獲取鎖，如果當前節(jié)點是 tail 節(jié)點，分兩種情況如下圖：

如果當前節(jié)點不是 tail 節(jié)點，也分兩種情況，如下圖：

4.對中斷狀態(tài)忽略

5.如果前置節(jié)點的狀態(tài)是 0 或 PROPAGATE，會被當前節(jié)點自旋過程中更新成 -1，以便之后通知當前節(jié)點。

2.1.3 獨占 + 響應(yīng)中斷

對應(yīng)方法 acquireInterruptibly(int arg)。

跟忽略中斷(acquire方法)不同的是要響應(yīng)中斷，下面兩個地方響應(yīng)中斷：

  • 獲取鎖之前會檢查當前線程是否中斷。
  • 獲取鎖失敗入隊，在隊列中自旋獲取鎖的過程中也會檢查當前線程是否中斷。如果檢查到當前線程已經(jīng)中斷，則拋出 InterruptedException，當前線程退出。

2.1.4 獨占 + 響應(yīng)中斷 + 考慮超時

對應(yīng)方法 tryAcquireNanos(int arg, long nanosTimeout)。

這個方法具備了獨占 + 響應(yīng)中斷 + 超時的功能，下面2個地方要判斷是否超時：

  • 自旋獲取鎖的過程中每次獲取鎖失敗都要判斷是否超時;
  • 獲取鎖失敗 park 之前要判斷超時時間是否大于自旋的閾值時間 (spinForTimeoutThreshold = 1ns) 另外，park 線程的操作使用 parkNanos 傳入阻塞時間。

2.2 釋放獨占鎖

獨占鎖釋放分兩步：釋放鎖，喚醒后繼節(jié)點。

釋放鎖的方法 tryRelease 是抽象的，由子類去實現(xiàn)。

我們看一下喚醒后繼節(jié)點的邏輯，首先需要滿足兩個條件：

  • head 節(jié)點不等于 null;
  • head 節(jié)點 waitStatus 不等于 0。這里有兩種情況(在方法 unparkSuccessor)：
  • 情況一，后繼節(jié)點 waitStatus <= 0，直接喚醒后繼節(jié)點，如下圖：

  • 情況二：后繼節(jié)點為空或者 waitStatus > 0，從后往前查找最接近當前節(jié)點的節(jié)點進行喚醒，如下圖：

2.3 獲取共享鎖

之前我們講了獨占鎖，這一小節(jié)我們談共享鎖，有什么不同呢？

2.3.1 共享，忽略 interrupts

對應(yīng)方法 acquireShared，代碼如下：

public final void acquireShared(int arg){
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

2.3.2 tryAcquireShared

這里獲取鎖使用的方法是 tryAcquireShared，獲取的是共享鎖。獲取共享鎖跟獲取獨占鎖不同的是，會返回一個整數(shù)值，說明如下:

  • 返回負數(shù)：獲取鎖失敗。
  • 返回 0：獲取鎖成功但是之后再由線程來獲取共享鎖時就會失敗。
  • 返回正數(shù):獲取鎖成功而且之后再有線程來獲取共享鎖時也可能會成功。所以需要把喚醒操作傳播下去。tryAcquireShared 獲取鎖失敗后(返回負數(shù))，就需要入隊后自旋獲取，也就是執(zhí)行方法 doAcquireShared。

2.3.3 doAcquireShared

怎么判斷隊列中等待節(jié)點是在等待共享鎖呢？nextWaiter == SHARED，這個參數(shù)值是入隊新建節(jié)點的時候構(gòu)造函數(shù)傳入的。

自旋過程中，如果獲取鎖成功(返回正數(shù))，首先把自己設(shè)置成新的 head 節(jié)點，然后把通知傳播下去。如下圖：

之后會喚醒后面節(jié)點并保證喚醒操作可以傳播下去。但是需要滿足四個條件中的一個：

  • tryAcquireShared 返回值大于0，有多余的鎖，可以繼續(xù)喚醒后繼節(jié)點。
  • 舊的 head 節(jié)點 waitStatus < 0，應(yīng)該是其他線程釋放共享鎖過程中把它的狀態(tài)更新成了 -3。
  • 新的 hade 節(jié)點 waitStatus < 0，只要不是 tail 節(jié)點，就可能是 -1。這里會造成不必要的喚醒，因為喚醒后獲取不到鎖只能繼續(xù)入隊等待。
  • 當前節(jié)點的后繼節(jié)點是空或者非空但正在等待共享鎖。

喚醒后面節(jié)點的操作，其實就是釋放共享鎖，對應(yīng)方法是 doReleaseShared，見釋放共享鎖一節(jié)。

2.3.4 共享 + 響應(yīng)中斷

對應(yīng)方法 acquireSharedInterruptibly(int arg)。

跟共享忽略中斷(acquireShared 方法)不同的是要響應(yīng)中斷，下面兩個地方響應(yīng)中斷：

  • 獲取鎖之前會檢查當前線程是否中斷。
  • 獲取鎖失敗入隊，在隊列中自旋獲取鎖的過程中也會檢查當前線程是否中斷。

如果檢查到當前線程已經(jīng)中斷，則拋出 InterruptedException，當前線程退出。

2.3.5 共享 + 響應(yīng)中斷 + 考慮超時

對應(yīng)方法 tryAcquireSharedNanos(int arg, long nanosTimeout)。

這個方法具備了共享 + 響應(yīng)中斷 + 超時的功能，下面兩個個地方要判斷是否超時：

  • 自旋獲取鎖的過程中每次獲取鎖失敗都要判斷是否超時。
  • 獲取鎖失敗 park 之前要判斷超時時間是否大于自旋的閾值時間(spinForTimeoutThreshold = 1ns)。

另外，park 線程的操作使用 parkNanos 傳入阻塞時間。

2.4 釋放共享鎖

釋放共享鎖代碼如下：

public final boolean releaseShared(int arg){
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

首先嘗試釋放共享鎖，tryReleaseShared 代碼由子類來實現(xiàn)。釋放成功后執(zhí)行AQS中的 doReleaseShared 方法，是一個自旋操作。

自旋的條件是隊列中至少有兩個節(jié)點，這里分三種情況。

情況一：當前節(jié)點 waitStatus 是 -1，如下圖：

情況二：當前節(jié)點 waitStatus 是 0(被其他線程更新新成了中間狀態(tài))，如下圖：

情況三：當前節(jié)點 waitStatus 是 -3，為什么會這樣呢？需要解釋一下，head節(jié)點喚醒后繼節(jié)點之前 waitStatus 已經(jīng)被更新中間態(tài) 0 了，喚醒后繼節(jié)點動作還沒有執(zhí)行，又被其他線程更成了 -3，也就是其他線程釋放鎖執(zhí)行了上面情況二。這時需要先把 waitStatus 再更成 0 (在方法 unparkSuccessor)，如下圖：

2.5 抽象方法

上面的講解可以看出，如果要基于 AQS 來實現(xiàn)并發(fā)鎖，可以根據(jù)需求重寫下面四個方法來實現(xiàn)，這四個方法在 AQS 中沒有具體實現(xiàn)：

  • tryAcquire(int arg)：獲取獨占鎖
  • tryRelease(int arg)：釋放獨占鎖
  • tryAcquireShared(int arg)：獲取共享鎖
  • tryReleaseShared(int arg)：釋放共享鎖

AQS 的子類需要重寫上面的方法來修改 state 值，并且定義獲取鎖或者釋放鎖時 state 值的變化。子類也可以定義自己的 state 變量，但是只有更新 AQS 中的 state變量才會對同步起作用。

還有一個判斷當前線程是否持有獨占鎖的方法 isHeldExclusively，也可以供子類重寫后使用。

獲取/釋放鎖的具體實現(xiàn)放到下篇文章講解。

2.6 總結(jié)

AQS 使用 FIFO 隊列實現(xiàn)了一個鎖相關(guān)的并發(fā)器模板，可以基于這個模板來實現(xiàn)各種鎖，包括獨占鎖、共享鎖、信號量等。

AQS 中，有一個核心狀態(tài)是 waitStatus，這個代表節(jié)點的狀態(tài)，決定了當前節(jié)點的后續(xù)操作，比如是否等待喚醒，是否要喚醒后繼節(jié)點。

3 并發(fā)鎖

這一章節(jié)講解 Java AQS 中的并發(fā)鎖。其實 Java AQS 中的并發(fā)鎖主要是基于 state 這個變量值來實現(xiàn)的。

3.1 ReentrantLock

我們先來看一下 UML 類圖：

從圖中可以看到，ReentrantLock 使用抽象內(nèi)部類 Sync 來實現(xiàn)了 AQS 的方法，然后基于 Sync 這個同步器實現(xiàn)了公平鎖和非公平鎖。主要實現(xiàn)了下面 3 個方法：

  • tryAcquire(int arg)：獲取獨占鎖
  • tryRelease(int arg)：釋放獨占鎖
  • isHeldExclusively：當前線程是否占有獨占鎖。ReentrantLock 默認實現(xiàn)的是非公平鎖，可以在構(gòu)造函數(shù)指定。

從實現(xiàn)的方法可以看到，ReentrantLock 中獲取的鎖是獨占鎖，我們再來看一下獲取和釋放獨占鎖的代碼：

public final void acquire(int arg){
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

獨占鎖的特點是調(diào)用上面 acquire 方法，傳入的參數(shù)是 1。

3.1.1 獲取公平鎖

獲取鎖首先判斷同步狀態(tài)(state)的值。

3.1.1.1 state 等于 0

這說明沒有線程占用鎖，當前線程如果符合下面兩個條件，就可以獲取到鎖：

沒有前任節(jié)點，如下圖：

CAS 的方式更新 state 值(把 0 更新成 1)成功。如果獲取獨占鎖成功，會更新 AQS 中 exclusiveOwnerThread 為當前線程，這個很容易理解。

3.1.1.2 state 不等于 0

這說明已經(jīng)有線程占有鎖，判斷占有鎖的線程是不是當前線程，如下圖：

state += 1 值如果小于 0，會拋出異常。

如果獲取鎖失敗，則進入 AQS 隊列等待喚醒。

3.1.2 獲取非公平鎖

跟公平鎖相比，非公平鎖的唯一不同是如果判斷到 state 等于 0，不用判斷有沒有前任節(jié)點，只要 CAS 設(shè)置 state 值(把 0 更新成 1)成功，就獲取到了鎖。

3.1.3 釋放鎖

公平鎖和非公平鎖，釋放邏輯完全一樣，都是在內(nèi)部類 Sync 中實現(xiàn)的。釋放鎖需要注意兩點，如下圖：

為什么 state 會大于 1，因為是可以重入的，占有鎖的線程可以多次獲取鎖。

3.1.4 總結(jié)

公平鎖的特點是每個線程都要進行排隊，不用擔心線程永遠獲取不到鎖，但有個缺點是每個線程入隊后都需要阻塞和被喚醒，這一定程度上影響了效率。非公平鎖的特點是每個線程入隊前都會先嘗試獲取鎖，如果獲取成功就不會入隊了，這比公平鎖效率高。但也有一個缺點，隊列中的線程有可能等待很長時間，高并發(fā)下甚至可能永遠獲取不到鎖。

3.2 ReentrantReadWriteLock

我們先來看一下 UML 類圖：

從圖中可以看到，ReentrantReadWriteLock 使用抽象內(nèi)部類Sync來實現(xiàn)了 AQS 的方法，然后基于 Sync 這個同步器實現(xiàn)了公平鎖和非公平鎖。主要實現(xiàn)了下面 3 個方法：

  • tryAcquire(int arg)：獲取獨占鎖
  • tryRelease(int arg)：釋放獨占鎖
  • tryAcquireShared(int arg)：獲取共享鎖
  • tryReleaseShared(int arg)：釋放共享鎖
  • isHeldExclusively：當前線程是否占有獨占鎖 可見ReentrantReadWriteLock里面同時用到了共享鎖和獨占鎖。

下圖是定義的幾個常用變量：

下面這 2 個方法用戶獲取共享鎖和獨占鎖的數(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 進行與運算，得到的是低 16 位的值，所以獨占鎖占了低 16 位。如下圖：

這樣上面獲取鎖數(shù)量的方法就很好理解了。

3.2.1 讀鎖

讀鎖的實現(xiàn)對應(yīng)內(nèi)部類 ReadLock。

3.2.1.1 獲取讀鎖

獲取讀鎖實際上是 ReadLock 調(diào)用了 AQS 的下面方法,傳入?yún)?shù)是 1：

public final void acquireShared(int arg){
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

ReentrantReadWriteLock 內(nèi)部類 Sync 實現(xiàn)了 tryAcquireShared 方法，主要包括如下三種情況：

  1. 使用 exclusiveCount 方法查看 state 中是否有獨占鎖，如果有并且獨占線程不是當前線程，返回 -1，獲取失敗;
  2. 使用 sharedCount 查看 state 中共享鎖數(shù)量，如果讀鎖數(shù)量小于最大值(MAX_COUNT=65535)，則再滿足下面 3 個條件就可以獲取成功并返回 1：

a.當前線程不需要阻塞(readerShouldBlock)。在公平鎖中，需要判斷是否有前置節(jié)點，如下圖就需要阻塞:

在非公平鎖中，則是判斷第一個節(jié)點是不是有獨占鎖，如下圖就需要阻塞:

b.使用 CAS 把 state 的值加 SHARED_UNIT(65536)。這里是不是就更理解讀鎖占高位的說法了，獲取一個讀鎖，state 的值就要加 SHARED_UNIT 這么多個。

c.給當前線程的 holdCount 加 1。

  1. 如果 2 失敗，自旋，重復(fù)上面的步驟直到獲取到鎖。tryAcquireShared (獲取共享鎖)會返回一個整數(shù)，如下：
  • 返回負數(shù)：獲取鎖失敗。
  • 返回 0：獲取鎖成功但是之后再由線程來獲取共享鎖時就會失敗。
  • 返回正數(shù):獲取鎖成功而且之后再有線程來獲取共享鎖時也可能會成功。
3.2.1.2 釋放讀鎖

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 實現(xiàn)了 releaseShared 方法，具體邏輯分為下面兩步：

  1. 當前線程 holdCounter 值減 1。
  2. CAS的方式將 state 的值減去 SHARED_UNIT。

3.2.2 寫鎖

寫鎖的實現(xiàn)對應(yīng)內(nèi)部類 WriteLock。

3.2.2.1 獲取寫鎖

ReentrantReadWriteLock 獲取寫鎖其實是在 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 實現(xiàn)了 tryAcquire 方法，首先獲取 state 值和獨占鎖數(shù)量(exclusiveCount),之后分如下兩種情況，如下圖：

1.state 不等于 0:

  • 獨占鎖數(shù)量等于 0，這時說明有線程占用了共享鎖，如果當前線程不是獨占線程，獲取鎖失敗。
  • 獨占鎖數(shù)量不等于 0，獨占鎖數(shù)量加 1 后大于 MAX_COUNT，獲取鎖失敗。
  • 上面 2 種情況不符合，獲取鎖成功，state 值加 1。2.state 等于 0，判斷當前線程是否需要阻塞(writerShouldBlock)。在公平鎖中，跟 readerShouldBlock 的邏輯完全一樣，就是判斷隊列中 head 節(jié)點的后繼節(jié)點是不是當前線程。在非公平鎖中，直接返回 false，即可以直接嘗試獲取鎖。

如果當前線程不需要阻塞，并且給 state 賦值成功，使用 CAS 方式把 state 值加 1，把獨占線程置為當前線程。

3.2.2.2 釋放寫鎖

ReentrantReadWriteLock 釋放寫鎖其實是在 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 中實現(xiàn)了 tryRelease(arg) 方法，邏輯如下：

  1. 判斷當前線程是不是獨占線程，如果不是，拋出異常。
  2. state值減1后，用新state值判斷獨占鎖數(shù)量是否等于0
  • 如果等于0，則把獨占線程置為空，返回true，這樣上面的代碼就可以喚醒隊列中的后置節(jié)點了
  • 如果不等于0，返回false，不喚醒后繼節(jié)點。

3.3 CountDownLatch

我們先來看一下UML類圖：

從上面的圖中看出，CountDownLatch 的內(nèi)部類 Sync 實現(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。

3.3.1 await

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 中實現(xiàn)了 tryAcquireShared 方法，await 邏輯如下圖：

上面的自旋過程就是等待 state 的值不斷減小，只有 state 值成為 0 的時候，主線程才會跳出自旋執(zhí)行之后的邏輯。

3.3.2 countDown

CountDownLatch 的 countDown 方法調(diào)用了 AQS 的 releaseShared(int arg)，傳入?yún)?shù) 1，不過這個參數(shù)并沒有用。內(nèi)部類 Sync 實現(xiàn)了 tryReleaseShared 方法，邏輯如下圖：

3.3.3 總結(jié)

CountDownLatch 的構(gòu)造函數(shù)入?yún)⒅禃x值給 state 變量，入隊操作是主線程入隊，每個子線程調(diào)用了countDown 后 state 值減 1，當 state 值成為 0 后喚醒主線程。

3.4 Semaphore

Semaphore 是一個信號量，用來保護共享資源。如果線程要訪問共享資源，首先從 Semaphore 獲取鎖(信號量)，如果信號量的計數(shù)器等于 0，則當前線程進入 AQS 隊列阻塞等待。否則，線程獲取鎖成功，信號量減 1。使用完共享資源后，釋放鎖(信號量加 1)。

Semaphore 跟管程模型不一樣的是，允許多個(構(gòu)造函數(shù)的 permits)線程進入管程內(nèi)部，因此也常用它來做限流。

UML 類圖如下：

Semaphore的構(gòu)造函數(shù)會傳入一個int類型參數(shù)，用來初始化state的值。

3.4.1 acquire

獲取鎖的操作調(diào)用了 AQS 中的 acquireSharedInterruptibly 方法，傳入?yún)?shù) 1，代碼見 CountDownLatch 中 await 小節(jié)。Semaphore 在公平鎖和非公平鎖中分別實現(xiàn)了 tryAcquireShared 方法。

3.4.1.1 公平鎖

Semaphore 默認使用非公平鎖，如果使用公平鎖，需要在構(gòu)造函數(shù)指定。獲取公平鎖邏輯比較簡單，如下圖：

3.4.1.2 非公平鎖

acquire 在非公平的鎖唯一的區(qū)別就是不會判斷 AQS 隊列是否有前置節(jié)點(hasQueuedPredecessors)，而是直接嘗試獲取鎖。

除了 acquire 方法外，還有其他幾個獲取鎖的方法，原理類似，只是調(diào)用了 AQS 中的不同方法。

3.4.2 release

釋放鎖的操作調(diào)用了 AQS 中的 releaseShared(int arg) 方法，傳入?yún)?shù) 1，在內(nèi)部類 Sync 中實現(xiàn)了 tryReleaseShared 方法，邏輯很簡單：使用 CAS 的方式將 state 的值加 1，之后喚醒隊列中的后繼節(jié)點。

3.5 ThreadPoolExecutor

ThreadPoolExecutor 中也用到了 AQS，看下面的 UML 類圖：

Worker 主要在 ThreadPoolExecutor 中斷線程的時候使用。Worker 自己實現(xiàn)了獨占鎖，在中斷線程時首先進行加鎖，中斷操作后釋放鎖。按照官方說法，這里不直接使用 ReentrantLock 的原因是防止調(diào)用控制線程池的方法(類似 setCorePoolSize)時能夠重新獲取到鎖，

3.5.1 tryAcquire

使用 CAS 的方式把 AQS 中 state 從 0 改為 1，把當前線程置為獨占線程。

3.5.2 tryRelease

把獨占線程置為空，把 AQS 中 state 改為 0。

Worker 初始化的時候會把 state 置為 -1，這樣是不能獲取鎖成功的。只有調(diào)用了 runWorker 方法，才會通過釋放鎖操作把 state 更為 0。這樣保證了只中斷運行中的線程，而不會中斷等待中的線程。

3.6 總結(jié)

AQS 基于雙向隊列實現(xiàn)了入口等待隊列，基于 state 變量實現(xiàn)了各種并發(fā)鎖，上篇文章講了入口等待隊列，而這篇文章主要講了基于 AQS 的并發(fā)鎖原理。

4、條件變量等待隊列

本章節(jié)主要講解管程模型中條件變量等待隊列。

4.1 官方示例

首先我們看一下官方給出的示例代碼：

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，說明如下：

  1. 如果 items 數(shù)組已經(jīng)滿了，則 notFull 變量不滿足，線程需要進入 notFull 條件等待隊列進行等待。當 take 方法取走一個數(shù)組元素時，notFull 條件滿足了，喚醒 notFull 條件等待隊列中等待線程。
  2. 如果 items 數(shù)組為空，則 notEmpty 變量不滿足，線程需要進入 notEmpty 條件等待隊列進行等待。當 put 方法加入一個數(shù)組元素時，notEmpty 條件滿足了，喚醒 notEmpty 條件等待隊列中等待線程。
  3. 條件變量是綁定在 Lock 上的，示例代碼使用了 ReentrantLock。在執(zhí)行 await 和 signal 方法時首先要獲取到鎖。

4.2 原理簡介

Java AQS 的條件變量等待隊列是基于接口 Condition 和 ConditionObject 來實現(xiàn)的，URM 類圖如下：

Condition 接口主要定義了下面3個方法：

  • await：進入條件等待隊列
  • signal：喚醒條件等待隊列中的元素
  • signalAll：喚醒條件等待隊列中的所有元素

4.3 await

條件等待隊列跟入口等待隊列有兩個不同：

  • 雖然二者共用了 Node 類，但是條件等待隊列是單向隊列，入口等待隊列是雙向隊列，條件隊列中下一個節(jié)點的引用是 nextWaiter，入口等待隊列中下一個節(jié)點的引用是 next。
  • 條件等待隊列中元素的 waitStatus 必須是 -2。await 方法的流程如下圖：

4.3.1 進入條件等待隊列

入隊方法對應(yīng)方法 addConditionWaiter，這里有三種情況：

  • 隊列為空，則新建一個節(jié)點，如下圖：

  • 隊列非空，最后一個元素的 waitStatus 是 -2，如下圖：

  • 隊列非空，最后一個元素的 waitStatus 不是 -2，如下圖：

可以看到，這種情況會從隊列第一個元素開始檢查 waitStatus 不是 -2 的元素，并從隊列中移除。

4.3.2 釋放鎖

AQS 的并發(fā)鎖是基于 state 變量實現(xiàn)的，線程進入條件等待隊列后，要釋放鎖，即 state 會變?yōu)?0，釋放操作會喚醒入口等待隊列中的線程。對應(yīng)方法 fullyRelease，返回值是釋放鎖減掉的 state 值 savedState。

4.3.3 阻塞等待

釋放鎖后，線程阻塞，自旋等待被喚醒。

4.3.4 喚醒之后

喚醒之后，當前線程主要有四個動作：

  • 轉(zhuǎn)入入口等待隊列，并把 waitStatus 改為 0。waitStatus 等于 0 表示中間狀態(tài)，當前節(jié)點后面的節(jié)點已經(jīng)喚醒，但是當前節(jié)點線程還沒有執(zhí)行完成。
  • 重新獲取鎖，如果獲取成功，則當前線程成為入口等待隊列頭結(jié)點，interruptMode 置為 1。
  • 如果當前節(jié)點在條件等待隊列中有后繼節(jié)點，則剔除條件等待隊列中 waitStatus!=-2 的節(jié)點，即隊列中狀態(tài)為取消的節(jié)點。
  • interruptMode 如果不等于 0，則處理中斷。

4.3.5 一個細節(jié)

上面提到了 interruptMode，這個屬性有三個值：

  • 0：沒有被中斷
  • -1：中斷后拋出 InterruptedException，這種情況是當前線程阻塞，沒有被 signal 之前發(fā)生了中斷
  • 1：重新進入中斷狀態(tài)，這種情況是指當前線程阻塞，被 signal 之后發(fā)生了中斷

4.3.6 擴展

AQS 還提供了其他幾個 await 方法，如下：

  • awaitUninterruptibly：不用處理中斷。
  • awaitNanos：自旋等待喚醒過程中有超時時間限制，超時則轉(zhuǎn)入入口等待隊列。
  • awaitUntil：自旋等待喚醒過程中有截止時間，時間到則轉(zhuǎn)入入口等待隊列。

4.4 signal

喚醒條件等待隊列中的元素，首先判斷當前線程是否持有獨占鎖，如果沒有，拋出異常。

喚醒條件隊列中的元素，會從第一個元素也就是 firstWaiter 開始，根據(jù) firstWaiter 的 waitStatus 是不是 -2，分兩種情況。

4.4.1 waitStatus==-2

條件隊列第一個節(jié)點進入入口等待隊列，等待獲取鎖，如下圖：

這里有兩個注意點：

  • 如果入口等待隊列中 tail 節(jié)點的 waitStatus 小于等于 0，則 firstWaiter 加入后需要把舊 tail 節(jié)點置為 -1 (表示后面節(jié)點等待當前節(jié)點喚醒)，如下圖：

如果重置 waitStatus 狀態(tài)失敗,則 unpark 節(jié)點 firstWaiter。

  • 如果入口等待隊列中 tail 節(jié)點的 waitStatus 大于 0，則 unpark 節(jié)點 firstWaiter。

4.4.2 waitStatus!=-2

如果 firstWaiter 的 waitStatus 不等于 -2，則查找 firstWaiter 的 nextWaiter，直到找到一個 waitStatus 等于 -2 的節(jié)點，然后將這個節(jié)點加入入口等待隊列隊尾，如下圖：

4.4.3 waitStatus 修改

上面的兩種情況無論哪種，進入入口等待隊列之前都要用 CAS 的方式把 waitStatus 改為 0。

4.5 signalAll

理解了 signal 的邏輯，signalAll 的邏輯就非常容易理解了。首先判斷當前線程是否持有獨占鎖，如果沒有，拋出異常。

將條件等待隊列中的所有節(jié)點依次加入入口等待隊列。如下圖：

4.6 使用案例

4.6.1 示例代碼

Java 并發(fā)包下有很多類使用到了 AQS 中的 Condition，如下圖：

這里我們以 CyclicBarrier 為例來講解。CyclicBarrier 是讓一組線程相互等待共同達到一個屏障點。從 Cyclic 可以看出 Barrier 可以循環(huán)利用，也就是當線程釋放之后可以繼續(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í)行完成

4.6.2 原理講解

CyclicBarrier 初始化的時候，會指定線程的數(shù)量 count，每個線程執(zhí)行完邏輯后，調(diào)用 CyclicBarrier 的 await 方法，這個方法首先將 count 減 1，然后調(diào)用 Condition的 await，讓當前線程進入條件等待隊列。當最后一個線程將 count 減 1 后，count 數(shù)量等于 0，這時就會調(diào)用 Condition 的 signalAll 方法喚醒所有線程。

4.7 總結(jié)

Java 的管程模型使用了 MESA 模型，基于 AQS 實現(xiàn)的 MESA 模型中，使用雙向隊列實現(xiàn)了入口等待隊列，使用變量 state 實現(xiàn)了并發(fā)鎖，使用 Condition 實現(xiàn)了條件等待隊列。

在 AQS 的實現(xiàn)中，使用同步隊列這個術(shù)語來表示雙向隊列，本文中使用入口等待隊列來描述是為了更好的配合管程模型來講解。

AQS 的 Condition 中，使用 await 方法將當前線程放入條件變量等待隊列阻塞等待，使用 notify 來喚醒條件等待隊列中的線程，被喚醒之后，線程并不能立刻執(zhí)行，而是進入入口等待隊列等待獲取鎖。


網(wǎng)站名稱：為了帶你精通JavaAQS，我畫了40張圖，從管程模型講起！
新聞來源：http://www.5511xx.com/article/dhogpog.html