Java 多線程并發(fā)AbstractQueuedSynchronizer詳情

更新時(shí)間：2022年06月16日 11:33:52 作者：? 自動(dòng)化BUG制造器? ?

這篇文章主要介紹了Java 多線程并發(fā)AbstractQueuedSynchronizer詳情，文章圍繞主題展開想象的內(nèi)容介紹，具有一定的參考價(jià)值，感興趣的小伙伴可以參考一下

AbstractQueuedSynchronizer

AbstractQueuedSynchronizer 簡(jiǎn)稱 AQS ，抽象隊(duì)列同步器，用來(lái)實(shí)現(xiàn)依賴于先進(jìn)先出（FIFO）等待隊(duì)列的阻塞鎖和相關(guān)同步器的框架。這個(gè)類旨在為大多數(shù)依賴單個(gè)原子 int 值來(lái)表示同步狀態(tài)的同步器提供基礎(chǔ)的能力封裝。例如 ReentrantLock、Semaphore 和 FutureTask 等等都是基于 AQS 實(shí)現(xiàn)的，我們也可以繼承 AQS 實(shí)現(xiàn)自定義同步器。

核心思想

網(wǎng)絡(luò)上常見的解釋是：

如果被請(qǐng)求的共享資源空閑，則將當(dāng)前請(qǐng)求資源的線程設(shè)置為有效的工作線程，并且將共享資源設(shè)置為鎖定狀態(tài)。如果被請(qǐng)求的共享資源被占用，那么就需要一套線程阻塞等待以及被喚醒時(shí)鎖分配的機(jī)制，這個(gè)機(jī)制AQS是用CLH隊(duì)列鎖實(shí)現(xiàn)的，即將暫時(shí)獲取不到鎖的線程加入到隊(duì)列中。

個(gè)人理解，可以把 AQS 當(dāng)成一把鎖，它內(nèi)部通過(guò)一個(gè)隊(duì)列記錄了所有要使用鎖的請(qǐng)求線程，并且管理鎖自己當(dāng)前的狀態(tài)（鎖定、空閑等狀態(tài)）。相當(dāng)于 AQS 就是共享資源本身，當(dāng)有線程請(qǐng)求這個(gè)資源是，AQS 將請(qǐng)求資源的線程記錄當(dāng)前工作線程，并將自身設(shè)置為鎖定狀態(tài)。后續(xù)其他線程請(qǐng)求這個(gè) AQS 時(shí)，將請(qǐng)求線程記錄到等待隊(duì)列中，其他線程此時(shí)未獲取到鎖，進(jìn)入阻塞等待狀態(tài)。

為什么需要 AQS

在深入 AQS 前，我們應(yīng)該持有一個(gè)疑問(wèn)是為什么需要 AQS ？synchronized 關(guān)鍵字和 CAS 原子類都提供了豐富的同步方案了。

但在實(shí)際的需求中，對(duì)同步的需求是各式各樣的，比如，我們需要對(duì)一個(gè)鎖加上超時(shí)時(shí)間，那么光憑 synchronized 關(guān)鍵字或是 CAS 就無(wú)法實(shí)現(xiàn)了，需要對(duì)其進(jìn)行二次封裝。而 JDK 中提供了豐富的同步方案，比如 ReentrantLock ，而 ReentrantLock 是就是基于 AQS 實(shí)現(xiàn)的。

用法

這部分內(nèi)容來(lái)自 JDK 的注釋

要將此類用作同步器的基礎(chǔ)，請(qǐng)?jiān)谶m用時(shí)重新定義以下方法，方法是使用 getState、setState 和/或 compareAndSetState 檢查和/或修改同步狀態(tài)：

tryAcquire
tryRelease
tryAcquireShared
tryReleaseShared
isHeldExclusively

默認(rèn)情況下，這些方法中的每一個(gè)都會(huì)引發(fā) UnsupportedOperationException。這些方法的實(shí)現(xiàn)必須是內(nèi)部線程安全的，并且通常應(yīng)該是短暫的而不是阻塞的。定義這些方法是使用此類的唯一受支持的方法。所有其他方法都被聲明為最終方法，因?yàn)樗鼈儾荒塥?dú)立變化。

您可能還會(huì)發(fā)現(xiàn)從 AbstractOwnableSynchronizer 繼承的方法對(duì)于跟蹤擁有獨(dú)占同步器的線程很有用。鼓勵(lì)您使用它們——這使監(jiān)視和診斷工具能夠幫助用戶確定哪些線程持有鎖。

即使此類基于內(nèi)部 FIFO 隊(duì)列，它也不會(huì)自動(dòng)執(zhí)行 FIFO 采集策略。

獨(dú)占同步的核心形式為：

 ? Acquire:
 ? ? ? while (!tryAcquire(arg)) {
 ? ? ? ? ?enqueue thread if it is not already queued;
 ? ? ? ? ?possibly block current thread;
 ? ? ? }
 ?
 ? Release:
 ? ? ? if (tryRelease(arg))
 ? ? ? ? ?unblock the first queued thread;

（共享模式類似，但可能涉及級(jí)聯(lián)信號(hào)。）

因?yàn)樵谌腙?duì)之前調(diào)用了獲取中的檢查，所以新獲取的線程可能會(huì)搶在其他被阻塞和排隊(duì)的線程之前。但是，如果需要，您可以定義 tryAcquire 和/或 tryAcquireShared 以通過(guò)內(nèi)部調(diào)用一個(gè)或多個(gè)檢查方法來(lái)禁用插入，從而提供公平的 FIFO 獲取順序。特別是，如果 hasQueuedPredecessors（一種專門為公平同步器使用的方法）返回 true，大多數(shù)公平同步器可以定義 tryAcquire 返回 false。其他變化是可能的。

默認(rèn)插入（也稱為貪婪、放棄和避免護(hù)送）策略的吞吐量和可擴(kuò)展性通常最高。雖然這不能保證公平或無(wú)饑餓，但允許較早排隊(duì)的線程在較晚的排隊(duì)線程之前重新競(jìng)爭(zhēng)，并且每次重新競(jìng)爭(zhēng)都有無(wú)偏見的機(jī)會(huì)成功對(duì)抗傳入線程。此外，雖然獲取不是通常意義上的“旋轉(zhuǎn)”，但它們可能會(huì)在阻塞之前執(zhí)行多次調(diào)用 tryAcquire 并穿插其他計(jì)算。當(dāng)獨(dú)占同步只是短暫地保持時(shí)，這提供了自旋的大部分好處，而沒(méi)有大部分責(zé)任。如果需要，您可以通過(guò)預(yù)先調(diào)用獲取具有“快速路徑”檢查的方法來(lái)增加這一點(diǎn)，可能會(huì)預(yù)先檢查 hasContended 和/或 hasQueuedThreads 以僅在同步器可能不會(huì)被爭(zhēng)用時(shí)才這樣做。

此類通過(guò)將其使用范圍專門用于可以依賴 int 狀態(tài)、獲取和釋放參數(shù)以及內(nèi)部 FIFO 等待隊(duì)列的同步器，部分地為同步提供了高效且可擴(kuò)展的基礎(chǔ)。如果這還不夠，您可以使用原子類、您自己的自定義 java.util.Queue 類和 LockSupport 阻塞支持從較低級(jí)別構(gòu)建同步器。

用法示例

這是一個(gè)不可重入互斥鎖類，它使用值 0 表示未鎖定狀態(tài)，使用值 1 表示鎖定狀態(tài)。雖然不可重入鎖并不嚴(yán)格要求記錄當(dāng)前所有者線程，但無(wú)論如何，此類都會(huì)這樣做以使使用情況更易于監(jiān)控。

它還支持條件并公開一些檢測(cè)方法：

class Mutex implements Lock, java.io.Serializable {
?
 ? // Our internal helper class
 ? private static class Sync extends AbstractQueuedSynchronizer {
 ? ? // Acquires the lock if state is zero
 ? ? public boolean tryAcquire(int acquires) {
 ? ? ? assert acquires == 1; // Otherwise unused
 ? ? ? if (compareAndSetState(0, 1)) {
 ? ? ? ? setExclusiveOwnerThread(Thread.currentThread());
 ? ? ? ? return true;
 ? ? ? }
 ? ? ? return false;
 ? ? }
?
 ? ? // Releases the lock by setting state to zero
 ? ? protected boolean tryRelease(int releases) {
 ? ? ? assert releases == 1; // Otherwise unused
 ? ? ? if (!isHeldExclusively())
 ? ? ? ? throw new IllegalMonitorStateException();
 ? ? ? setExclusiveOwnerThread(null);
 ? ? ? setState(0);
 ? ? ? return true;
 ? ? }?
 ? ? // Reports whether in locked state
 ? ? public boolean isLocked() {
 ? ? ? return getState() != 0;
 ? ? }
 ? ? public boolean isHeldExclusively() {
 ? ? ? // a data race, but safe due to out-of-thin-air guarantees
 ? ? ? return getExclusiveOwnerThread() == Thread.currentThread();
 ? ? }?
 ? ? // Provides a Condition
 ? ? public Condition newCondition() {
 ? ? ? return new ConditionObject();
 ? ? }
 ? ? // Deserializes properly
 ? ? private void readObject(ObjectInputStream s)
 ? ? ? ? throws IOException, ClassNotFoundException {
 ? ? ? s.defaultReadObject();
 ? ? ? setState(0); // reset to unlocked state
 ? ? }
 ? }
 ? // The sync object does all the hard work. We just forward to it.
 ? private final Sync sync = new Sync();
 ? public void lock() ? ? ? ? ? ?  { sync.acquire(1); }
 ? public boolean tryLock() ? ? ?  { return sync.tryAcquire(1); }
 ? public void unlock() ? ? ? ? ?  { sync.release(1); }
 ? public Condition newCondition() { return sync.newCondition(); }
 ? public boolean isLocked() ? ? ? { return sync.isLocked(); }
 ? public boolean isHeldByCurrentThread() {
 ? ? return sync.isHeldExclusively();
 ? }
 ? public boolean hasQueuedThreads() {
 ? ? return sync.hasQueuedThreads();
 ? }
 ? public void lockInterruptibly() throws InterruptedException {
 ? ? sync.acquireInterruptibly(1);
 ? }
 ? public boolean tryLock(long timeout, TimeUnit unit)
 ? ? ? throws InterruptedException {
 ? ? return sync.tryAcquireNanos(1, unit.toNanos(timeout));
 ? }
 }

這是一個(gè)類似于 CountDownLatch 的鎖存器類，只是它只需要一個(gè)信號(hào)即可觸發(fā)。因?yàn)殒i存器是非獨(dú)占的，所以它使用共享的獲取和釋放方法。

 class BooleanLatch {?
 ? private static class Sync extends AbstractQueuedSynchronizer {
 ? ? boolean isSignalled() { return getState() != 0; }
 ? ? protected int tryAcquireShared(int ignore) {
 ? ? ? return isSignalled() ? 1 : -1;
 ? ? }
 ? ? protected boolean tryReleaseShared(int ignore) {
 ? ? ? setState(1);
 ? ? ? return true;
 ? ? }
 ? }
 ? private final Sync sync = new Sync();
 ? public boolean isSignalled() { return sync.isSignalled(); }
 ? public void signal() ? ? ? ? { sync.releaseShared(1); }
 ? public void await() throws InterruptedException {
 ? ? sync.acquireSharedInterruptibly(1);
 ? }
 }

AQS 底層原理

父類 AbstractOwnableSynchronizer

AbstractQueuedSynchronizer 繼承自 AbstractOwnableSynchronizer ，后者邏輯十分簡(jiǎn)單：

public abstract class AbstractOwnableSynchronizer implements java.io.Serializable {?
 ? ?private static final long serialVersionUID = 3737899427754241961L;?
 ? ?protected AbstractOwnableSynchronizer() { }
 ? ?private transient Thread exclusiveOwnerThread;
    // 設(shè)置當(dāng)前持有鎖的線程
 ? ?protected final void setExclusiveOwnerThread(Thread thread) {
 ? ? ? ?exclusiveOwnerThread = thread;
 ?  }
 ? ?protected final Thread getExclusiveOwnerThread() {
 ? ? ? ?return exclusiveOwnerThread;
 ?  }
}

AbstractOwnableSynchronizer 只是定義了設(shè)置持有鎖的線程的能力。

CLH 隊(duì)列

AQS 的等待隊(duì)列是 CLH (Craig , Landin , and Hagersten) 鎖定隊(duì)列的變體，CLH 鎖通常用于自旋鎖。AQS 將每個(gè)請(qǐng)求共享資源的線程封裝程一個(gè) CLH 節(jié)點(diǎn)來(lái)實(shí)現(xiàn)的，這個(gè)節(jié)點(diǎn)的定義是：

 ? ?/** CLH Nodes */
 ? ?abstract static class Node {
 ? ? ? ?volatile Node prev; ? ? ? // initially attached via casTail
 ? ? ? ?volatile Node next; ? ? ? // visibly nonnull when signallable
 ? ? ? ?Thread waiter; ? ? ? ? ? ?// visibly nonnull when enqueued
 ? ? ? ?volatile int status; ? ? ?// written by owner, atomic bit ops by others
?
 ? ? ? ?// methods for atomic operations
 ? ? ? ?final boolean casPrev(Node c, Node v) { ?// for cleanQueue
 ? ? ? ? ? ?return U.weakCompareAndSetReference(this, PREV, c, v); // 通過(guò) CAS 確保同步設(shè)置 prev 的值
 ? ? ?  }
 ? ? ? ?final boolean casNext(Node c, Node v) { ?// for cleanQueue
 ? ? ? ? ? ?return U.weakCompareAndSetReference(this, NEXT, c, v);
 ? ? ?  }
 ? ? ? ?final int getAndUnsetStatus(int v) { ? ? // for signalling
 ? ? ? ? ? ?return U.getAndBitwiseAndInt(this, STATUS, ~v);
 ? ? ?  }
 ? ? ? ?final void setPrevRelaxed(Node p) { ? ? ?// for off-queue assignment
 ? ? ? ? ? ?U.putReference(this, PREV, p);
 ? ? ?  }
 ? ? ? ?final void setStatusRelaxed(int s) { ? ? // for off-queue assignment
 ? ? ? ? ? ?U.putInt(this, STATUS, s);
 ? ? ?  }
 ? ? ? ?final void clearStatus() { ? ? ? ? ? ? ? // for reducing unneeded signals
 ? ? ? ? ? ?U.putIntOpaque(this, STATUS, 0);
 ? ? ?  }
 ? ? ? ?private static final long STATUS = U.objectFieldOffset(Node.class, "status");
 ? ? ? ?private static final long NEXT = U.objectFieldOffset(Node.class, "next");
 ? ? ? ?private static final long PREV = U.objectFieldOffset(Node.class, "prev");
 ?  }

CLH 的節(jié)點(diǎn)的數(shù)據(jù)結(jié)構(gòu)是一個(gè)雙向鏈表的節(jié)點(diǎn)，只不過(guò)每個(gè)操作都是經(jīng)過(guò) CAS 確保線程安全的。要加入 CLH 鎖隊(duì)列，您可以將其自動(dòng)拼接為新的尾部；要出隊(duì)，需要設(shè)置 head 字段，以便下一個(gè)符合條件的等待節(jié)點(diǎn)成為新的頭節(jié)點(diǎn)：

 +------+  prev +-------+  prev +------+
 | ? ?  | <---- | ? ? ? | <---- | ? ?  |
 | head | next  | first | next  | tail |
 | ? ?  | ----> | ? ? ? | ----> | ? ?  |
 +------+ ? ? ? +-------+ ? ? ? +------+

Node 中的 status 字段表示當(dāng)前節(jié)點(diǎn)代表的線程的狀態(tài)。

status 存在三種狀態(tài)：

 ? ?static final int WAITING ? = 1; ? ? ? ? ?// must be 1
 ? ?static final int CANCELLED = 0x80000000; // must be negative 
 ? ?static final int COND ? ? ?= 2; ? ? ? ? ?// in a condition wait

WAITING：表示等待狀態(tài)，值為 1。
CANCELLED：表示當(dāng)前線程被取消，為 0x80000000。
COND：表示當(dāng)前節(jié)點(diǎn)在等待條件，也就是在條件等待隊(duì)列中，值為 2。

在上面的 COND 中，提到了一個(gè)條件等待隊(duì)列的概念。

首先，Node 是一個(gè)靜態(tài)抽象類，它在 AQS 中存在三種實(shí)現(xiàn)類：

ExclusiveNode
SharedNode
ConditionNode

前兩者都是空實(shí)現(xiàn)：

 ? ?static final class ExclusiveNode extends Node { }
 ? ?static final class SharedNode extends Node { }

而最后的 ConditionNode 多了些內(nèi)容：

 ? ?static final class ConditionNode extends Node implements ForkJoinPool.ManagedBlocker {
 ? ? ? ?ConditionNode nextWaiter; 
 ? ? ? ?// 檢查線程是否中斷或當(dāng)前線程的狀態(tài)已取消等待。
 ? ? ? ?public final boolean isReleasable() {
 ? ? ? ? ? ?return status <= 1 || Thread.currentThread().isInterrupted();
 ? ? ?  }
?
 ? ? ? ?public final boolean block() {
 ? ? ? ? ? ?while (!isReleasable()) LockSupport.park();
 ? ? ? ? ? ?return true;
 ? ? ?  }
 ?  }

ConditionNode 拓展了兩個(gè)方法：

檢查線程狀態(tài)是否處于等待。
阻塞當(dāng)前線程：當(dāng)前線程正在等待執(zhí)行，通過(guò) LockSupport.park() 阻塞當(dāng)前線程。這里通過(guò) while 循環(huán)持續(xù)重試，嘗試阻塞線程。

而到這一步，所有的信息都指向了一個(gè)相關(guān)的類 Condition 。

Condition

AQS 中的 Condition 的實(shí)現(xiàn)是內(nèi)部類 ConditionObject ：

public class ConditionObject implements Condition, java.io.Serializable

ConditionObject 實(shí)現(xiàn)了 Condition 接口和序列化接口，后者說(shuō)明了該類型的對(duì)象可以進(jìn)行序列化。而前者 Condition 接口，定義了一些行為能力：

public interface Condition {
 ? ?void await() throws InterruptedException;?
 ? ?void awaitUninterruptibly();?
 ? ?long awaitNanos(long nanosTimeout) throws InterruptedException;?
 ? ?boolean await(long time, TimeUnit unit) throws InterruptedException;?
 ? ?boolean awaitUntil(Date deadline) throws InterruptedException;?
 ? ?void signal();
 ? ?void signalAll();
}

Condition 中定義的能力與 Java 的 Object 類中提供的同步相關(guān)方法（wait、notify 和 notifyAll）代表的能力極為相似。前者提供了更豐富的等待方法。類比的角度來(lái)看，如果 Object 是配合 synchronized 關(guān)鍵字使用的，那么 Condition 就是用來(lái)配合基于 AQS 實(shí)現(xiàn)的鎖來(lái)使用的接口。

可以將 Condition 的方法分為兩組：等待和喚醒。

用于等待的方法

// 等待，當(dāng)前線程在接到信號(hào)或被中斷之前一直處于等待狀態(tài) ? ?
void await() throws InterruptedException;
// 等待，當(dāng)前線程在接到信號(hào)之前一直處于等待狀態(tài)，不響應(yīng)中斷
void awaitUninterruptibly();
//等待，當(dāng)前線程在接到信號(hào)、被中斷或到達(dá)指定等待時(shí)間之前一直處于等待狀態(tài) 
long awaitNanos(long nanosTimeout) throws InterruptedException;
// 等待，當(dāng)前線程在接到信號(hào)、被中斷或到達(dá)指定等待時(shí)間之前一直處于等待狀態(tài)。
// 此方法在行為上等效于: awaitNanos(unit.toNanos(time)) > 0
boolean await(long time, TimeUnit unit) throws InterruptedException;
// 等待，當(dāng)前線程在接到信號(hào)、被中斷或到達(dá)指定最后期限之前一直處于等待狀態(tài) ? ?
boolean awaitUntil(Date deadline) throws InterruptedException;

用于喚醒的方法

// 喚醒一個(gè)等待線程。如果所有的線程都在等待此條件，則選擇其中的一個(gè)喚醒。在從 await 返回之前，該線程必須重新獲取鎖。
void signal();
// 喚醒所有等待線程。如果所有的線程都在等待此條件，則喚醒所有線程。在從 await 返回之前，每個(gè)線程都必須重新獲取鎖。
void signalAll();

ConditionObject

分析完 Condition ，繼續(xù)來(lái)理解 ConditionObject。 ConditionObject 是 Condition 在 AQS 中的實(shí)現(xiàn)：

public class ConditionObject implements Condition, java.io.Serializable {
 ? ?/** condition 隊(duì)列頭節(jié)點(diǎn) */
 ? ?private transient ConditionNode firstWaiter;
 ? ?/** condition 隊(duì)列尾節(jié)點(diǎn) */
 ? ?private transient ConditionNode lastWaiter;
 ? ?// ---- Signalling methods ----
 ? ?// 移除一個(gè)或所有等待者并將其轉(zhuǎn)移到同步隊(duì)列。
 ? ?private void doSignal(ConditionNode first, boolean all)
 ? ?public final void signal()
 ? ?public final void signalAll()?
 ? ?// ---- Waiting methods ----
 ? ?// 將節(jié)點(diǎn)添加到條件列表并釋放鎖定。
 ? ?private int enableWait(ConditionNode node)
 ? ?// 如果最初放置在條件隊(duì)列中的節(jié)點(diǎn)現(xiàn)在準(zhǔn)備好重新獲取同步隊(duì)列，則返回 true。
 ? ?private boolean canReacquire(ConditionNode node) ?
 ? ?// 從條件隊(duì)列中取消鏈接給定節(jié)點(diǎn)和其他非等待節(jié)點(diǎn)，除非已經(jīng)取消鏈接。
 ? ?private void unlinkCancelledWaiters(ConditionNode node) 
 ? ?// 實(shí)現(xiàn)不可中斷的條件等待
 ? ?public final void awaitUninterruptibly()?
 ? ?public final void await()?
 ? ?public final long awaitNanos(long nanosTimeout)?
 ? ?public final boolean awaitUntil(Date deadline)?
 ? ?public final boolean await(long time, TimeUnit unit)?
 ? ?//  ---- support for instrumentation ----?
 ? ?// 如果此條件是由給定的同步對(duì)象創(chuàng)建的，則返回 true。
 ? ?final boolean isOwnedBy(AbstractQueuedSynchronizer sync)?
 ? ?// 查詢是否有線程在此條件下等待。
 ? ?protected final boolean hasWaiters()?
 ? ?// 返回在此條件下等待的線程數(shù)的估計(jì)值。
 ? ?protected final int getWaitQueueLength()
 ? ?// 返回一個(gè)集合，其中包含可能正在等待此 Condition 的那些線程。
 ? ?protected final Collection<Thread> getWaitingThreads()
}

ConditionObject 實(shí)現(xiàn)了 Condition 能力的基礎(chǔ)上，拓展了對(duì) ConditionNode 相關(guān)的操作，方法通過(guò)其用途可以劃分為三組：

Signalling
Waiting
其他方法

Signalling methods

 ? ? ? ?public final void signal() {
 ? ? ? ? ? ?ConditionNode first = firstWaiter;
 ? ? ? ? ? ?if (!isHeldExclusively())
 ? ? ? ? ? ? ? ?throw new IllegalMonitorStateException();
 ? ? ? ? ? ?if (first != null)
 ? ? ? ? ? ? ? ?doSignal(first, false);
 ? ? ?  }
 ? ? ? ?public final void signalAll() {
 ? ? ? ? ? ?ConditionNode first = firstWaiter;
 ? ? ? ? ? ?if (!isHeldExclusively())
 ? ? ? ? ? ? ? ?throw new IllegalMonitorStateException();
 ? ? ? ? ? ?if (first != null)
 ? ? ? ? ? ? ? ?doSignal(first, true);
 ? ? ?  }

喚醒方法主要邏輯是通過(guò) doSignal(ConditionNode first, boolean all) 實(shí)現(xiàn)的。doSignal 方法根據(jù)參數(shù)，進(jìn)行一個(gè) while 循環(huán)，

兩個(gè)方法傳遞進(jìn)來(lái)的都是頭節(jié)點(diǎn)，也就是從 ConditionNode 雙向鏈表的頭節(jié)點(diǎn)開始遍歷，如果第二個(gè)參數(shù) all 設(shè)置為 false ，只執(zhí)行一次遍歷中邏輯。循環(huán)中的邏輯是：

// 最終都調(diào)用了這個(gè)方法
private void doSignal(ConditionNode first, boolean all) {
    while (first != null) {
 ? ?    // 取出 first 的下一個(gè)節(jié)點(diǎn)，設(shè)置為 next
        ConditionNode next = first.nextWaiter; 
 ? ?    // 如果 first 是鏈表中唯一的一個(gè)節(jié)點(diǎn)，設(shè)置 lastWaiter 為 null
        if ((firstWaiter = next) == null) // 
            lastWaiter = null;
 ? ?    // 讀取 first 的 status ，檢查是否是 COND
        if ((first.getAndUnsetStatus(COND) & COND) != 0) { 
 ? ? ? ?    // first 處于 COND 狀態(tài)，出隊(duì)
            enqueue(first); 
 ? ? ? ?    // 通過(guò) all 來(lái)判斷是否將等待的線程都進(jìn)行喚醒邏輯。
            if (!all)
                break; ?
        }
        first = next; // 循環(huán)指向下一個(gè)
    }
}

關(guān)鍵方法 enqueue(ConditionNode) 是 AQS 中的方法：

 ? ?final void enqueue(Node node) {
 ? ? ? ?if (node != null) {
 ? ? ? ? ? ?for (;;) {
 ? ? ? ? ? ?    // 獲取尾節(jié)點(diǎn)
 ? ? ? ? ? ? ? ?Node t = tail; 
 ? ? ? ? ? ?    // 避免不必要的內(nèi)存屏障
 ? ? ? ? ? ? ? ?node.setPrevRelaxed(t); 
 ? ? ? ? ? ? ? ?if (t == null) ? ? ?
 ? ? ? ? ? ? ? ?    // 空隊(duì)列首先初始化一個(gè)頭節(jié)點(diǎn)
 ? ? ? ? ? ? ? ? ? ?tryInitializeHead();  
 ? ? ? ? ? ? ? ?else if (casTail(t, node)) { // 更新 tail 指針為 node （這里不是將 t = node）
 ? ? ? ? ? ? ? ? ? ?t.next = node; // 為節(jié)點(diǎn) t 的 next 指針指向 node
 ? ? ? ? ? ? ? ? ? ?if (t.status < 0) ?// t 的狀態(tài) < 0 一般代表后續(xù)節(jié)點(diǎn)需要運(yùn)行了
 ? ? ? ? ? ? ? ? ? ? ? ?LockSupport.unpark(node.waiter);
 ? ? ? ? ? ? ? ? ? ?break;
 ? ? ? ? ? ? ?  }
 ? ? ? ? ?  }
 ? ? ?  }
 ?  }

可以看出 enqueue(ConditionNode) 中本質(zhì)上是通過(guò)調(diào)用 LockSupport.unpark(node.waiter); 來(lái)喚醒線程的。

Waiting methods

對(duì)外提供的等待能力的方法包括：

 ? ?// 實(shí)現(xiàn)不可中斷的條件等待
 ? ?public final void awaitUninterruptibly()
 ? ?public final void await()?
 ? ?public final long awaitNanos(long nanosTimeout)?
 ? ?public final boolean awaitUntil(Date deadline)
 ? ?public final boolean await(long time, TimeUnit unit)

它們內(nèi)部都用到了公共的邏輯：

 ? ?// 添加節(jié)點(diǎn)到 condition 列表并釋放鎖
    private int enableWait(ConditionNode node)
 ? ?private boolean canReacquire(ConditionNode node) 
 ? ?private void unlinkCancelledWaiters(ConditionNode node)

enableWait

 ? ? ? ?private int enableWait(ConditionNode node) {
 ? ? ? ? ? ?if (isHeldExclusively()) { // 如果是當(dāng)前線程持有鎖資源
 ? ? ? ? ? ? ? ?node.waiter = Thread.currentThread(); ?// 將節(jié)點(diǎn)的綁定的線程設(shè)置為當(dāng)前線程
 ? ? ? ? ? ? ? ?node.setStatusRelaxed(COND | WAITING); // 設(shè)置節(jié)點(diǎn)狀態(tài)
 ? ? ? ? ? ? ? ?ConditionNode last = lastWaiter;       // 獲取 尾節(jié)點(diǎn)
 ? ? ? ? ? ? ? ?if (last == null)
 ? ? ? ? ? ? ? ? ? ?firstWaiter = node;                // 如果列表為空， node 就是頭節(jié)點(diǎn)
 ? ? ? ? ? ? ? ?else
 ? ? ? ? ? ? ? ? ? ?last.nextWaiter = node;            // 否則，將尾節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)設(shè)置為 node
 ? ? ? ? ? ? ? ?lastWaiter = node;                     // 更新 lastWaiter 指針
 ? ? ? ? ? ? ? ?int savedState = getState();           // 獲取當(dāng)前線程的同步狀態(tài)
 ? ? ? ? ? ? ? ?if (release(savedState))               // 在當(dāng)前持有鎖資源的線程嘗試釋放鎖
 ? ? ? ? ? ? ? ? ? ?return savedState;
 ? ? ? ? ?  }
 ? ? ? ? ? ?node.status = CANCELLED; // 當(dāng)前線程未持有鎖資源，更新 node 的狀態(tài)為 CANCELLED
 ? ? ? ? ? ?throw new IllegalMonitorStateException(); // 并拋出 IllegalMonitorStateException
 ? ? ?  }

這個(gè)方法對(duì)傳入的節(jié)點(diǎn)插入到等待隊(duì)列的隊(duì)尾，并根據(jù)當(dāng)前線程的狀態(tài)進(jìn)行了檢查。關(guān)鍵方法的 release(int) ：

 ? ?public final boolean release(int arg) {
 ? ? ? ?if (tryRelease(arg)) { // 嘗試釋放鎖資源
 ? ? ? ? ? ?signalNext(head); ?// 釋放成功，喚醒下一個(gè)等待中的線程
 ? ? ? ? ? ?return true;
 ? ? ?  }
 ? ? ? ?return false;
 ?  }

喚醒給定節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)（如果存在），通過(guò)調(diào)用 LockSupport.unpark(s.waiter) 喚醒節(jié)點(diǎn)對(duì)應(yīng)的線程。

 ? ?private static void signalNext(Node h) {
 ? ? ? ?Node s;
 ? ? ? ?if (h != null && (s = h.next) != null && s.status != 0) {
 ? ? ? ? ? ?s.getAndUnsetStatus(WAITING);
 ? ? ? ? ? ?LockSupport.unpark(s.waiter);
 ? ? ?  }
 ?  }

canReacquire

檢查傳入的 node 是否在鏈表中，且不為頭節(jié)點(diǎn)：

// 如果最初放置在條件隊(duì)列中的節(jié)點(diǎn)現(xiàn)在準(zhǔn)備好重新獲取同步隊(duì)列，則返回 true。
private boolean canReacquire(ConditionNode node) {
    // 檢查傳入的 node 是否在鏈表中，且不為頭節(jié)點(diǎn)
    return node != null && node.prev != null && isEnqueued(node);
}

// in AQS 
final boolean isEnqueued(Node node) {
    // 從 Node 雙向鏈表尾部開始遍歷，是否存在 node
    for (Node t = tail; t != null; t = t.prev)
        if (t == node)
 ? ? ?      return true;
 ? ?return false;
}

unlinkCancelledWaiters

 ? ? ? ?private void unlinkCancelledWaiters(ConditionNode node) {
 ? ? ? ?    // node 為空 / node 不是隊(duì)尾 / node 是最后一個(gè)節(jié)點(diǎn)
 ? ? ? ? ? ?if (node == null || node.nextWaiter != null || node == lastWaiter) {
 ? ? ? ? ? ? ? ?ConditionNode w = firstWaiter, trail = null; // w = first ， trail = null
 ? ? ? ? ? ?    // /從鏈表頭節(jié)點(diǎn)開始遍歷
 ? ? ? ? ? ? ? ?while (w != null) { 
 ? ? ? ? ? ? ? ? ? ?ConditionNode next = w.nextWaiter;  // 取出下一個(gè)節(jié)點(diǎn)
 ? ? ? ? ? ? ? ? ? ?if ((w.status & COND) == 0) {       // 當(dāng)前節(jié)點(diǎn)的狀態(tài)包含 COND
 ? ? ? ? ? ? ? ? ? ? ? ?w.nextWaiter = null;            // 當(dāng)前節(jié)點(diǎn)的 next 設(shè)置為 null 
 ? ? ? ? ? ? ? ? ? ? ? ?if (trail == null)              // 如果 trail 指針為空
 ? ? ? ? ? ? ? ? ? ? ? ? ? ?firstWaiter = next;         // firstWaiter 指向 next
 ? ? ? ? ? ? ? ? ? ? ? ?else
 ? ? ? ? ? ? ? ? ? ? ? ? ? ?trail.nextWaiter = next;    // trail 指針不為空，尾指針的 next 指向當(dāng)前節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn) 
 ? ? ? ? ? ? ? ? ? ? ? ?if (next == null)
 ? ? ? ? ? ? ? ? ? ? ? ? ? ?lastWaiter = trail; // 最后將 lastWaiter 設(shè)置為 trail （過(guò)濾后的 trail 鏈表插入到隊(duì)尾）
 ? ? ? ? ? ? ? ? ?  } else
 ? ? ? ? ? ? ? ? ? ? ? ?trail = w; // 頭節(jié)點(diǎn)狀態(tài)不是 COND，當(dāng)前節(jié)點(diǎn)設(shè)置為 trail 指針。
 ? ? ? ? ? ? ? ? ? ?w = next; // 下一個(gè)循環(huán)
 ? ? ? ? ? ? ?  } 
 ? ? ? ? ?  }
 ? ? ?  }

這個(gè)方法遍歷 ConditionNode 隊(duì)列，過(guò)濾掉狀態(tài)不包含 COND 的節(jié)點(diǎn)。

對(duì)外提供的等待方法

上面三個(gè)方法是內(nèi)部處理邏輯。而對(duì)外暴露的是以下五個(gè)方法：

 ? ?public final void awaitUninterruptibly()?
 ? ?public final void await()?
 ? ?public final long awaitNanos(long nanosTimeout)?
 ? ?public final boolean awaitUntil(Date deadline)?
 ? ?public final boolean await(long time, TimeUnit unit)

除了awaitUninterruptibly() ，其他方法所代表的能力和 Condition 接口中定義的所代表的能力基本一致。

awaitUninterruptibly

awaitUninterruptibly() 是用于實(shí)現(xiàn)不可中斷的條件等待：

 ? ? ? ?public final void awaitUninterruptibly() {
 ? ? ? ? ? ?ConditionNode node = new ConditionNode(); // 創(chuàng)建一個(gè)新的 node
 ? ? ? ? ? ?int savedState = enableWait(node);        // 將這個(gè)新 node 插入，并返回 node 的狀態(tài)
 ? ? ? ? ? ?LockSupport.setCurrentBlocker(this);      // 設(shè)置 blocker
 ? ? ? ? ? ?boolean interrupted = false, rejected = false;  // flag：中斷和拒絕
 ? ? ? ? ? ?while (!canReacquire(node)) {             // 當(dāng)前線程關(guān)聯(lián)的 node 不再等待隊(duì)列      
 ? ? ? ? ? ? ? ?if (Thread.interrupted())             // 嘗試中斷線程
 ? ? ? ? ? ? ? ? ? ?interrupted = true;
 ? ? ? ? ? ? ? ?else if ((node.status & COND) != 0) { ?// 中斷線程不成功的情況下，如果 node 狀態(tài)包含 COND
 ? ? ? ? ? ? ? ? ? ?// 嘗試阻塞線程
 ? ? ? ? ? ? ? ? ?  try {
 ? ? ? ? ? ? ? ? ? ? ? ?if (rejected) ?
 ? ? ? ? ? ? ? ? ? ? ? ? ? ?node.block(); // 實(shí)際上也是 LockSupport.park
 ? ? ? ? ? ? ? ? ? ? ? ?else
 ? ? ? ? ? ? ? ? ? ? ? ? ? ?ForkJoinPool.managedBlock(node); 
 ? ? ? ? ? ? ? ? ?  } catch (RejectedExecutionException ex) {
 ? ? ? ? ? ? ? ? ? ? ? ?rejected = true;    // 拒絕執(zhí)行
 ? ? ? ? ? ? ? ? ?  } catch (InterruptedException ie) {
 ? ? ? ? ? ? ? ? ? ? ? ?interrupted = true;   // 中斷
 ? ? ? ? ? ? ? ? ?  }
 ? ? ? ? ? ? ?  } else
 ? ? ? ? ? ? ? ? ? ?Thread.onSpinWait(); ?      // 當(dāng)前線程無(wú)法繼續(xù)執(zhí)行
 ? ? ? ? ?  }
 ? ? ? ?    // 不是隊(duì)列中的唯一節(jié)點(diǎn)時(shí)執(zhí)行下面邏輯
 ? ? ? ? ? ?LockSupport.setCurrentBlocker(null); 
 ? ? ? ? ? ?node.clearStatus();   // 清除 node 的 status 
 ? ? ? ? ? ?acquire(node, savedState, false, false, false, 0L); // 【*】重點(diǎn)方法
 ? ? ? ? ? ?if (interrupted)
 ? ? ? ? ? ? ? ?Thread.currentThread().interrupt();
 ? ? ?  }

在這個(gè)方法中，首先講解兩個(gè)方法：

Thread.onSpinWait() 表示調(diào)用者暫時(shí)無(wú)法繼續(xù)，直到其他活動(dòng)發(fā)生一個(gè)或多個(gè)動(dòng)作。通過(guò)在自旋等待循環(huán)構(gòu)造的每次迭代中調(diào)用此方法，調(diào)用線程向運(yùn)行時(shí)指示它正忙于等待。運(yùn)行時(shí)可能會(huì)采取措施來(lái)提高調(diào)用自旋等待循環(huán)構(gòu)造的性能。
ForkJoinPool.managedBlock(node) 則是通過(guò) Blocker 來(lái)檢查線程的運(yùn)行狀態(tài)，然后嘗試阻塞線程。

最后是最關(guān)鍵的方法 acquire ，它的詳細(xì)邏輯放到最后講解，這個(gè)方法的作用就是，當(dāng)前線程進(jìn)入等待后，需要將關(guān)聯(lián)的線程開啟一個(gè)自旋，掛起后能夠持續(xù)去嘗試獲取鎖資源。

await

 ? ? ? ?public final void await() throws InterruptedException {
 ? ? ? ? ? ?if (Thread.interrupted())
 ? ? ? ? ? ? ? ?throw new InterruptedException();
 ? ? ? ? ? ?ConditionNode node = new ConditionNode();
 ? ? ? ? ? ?int savedState = enableWait(node);
 ? ? ? ? ? ?LockSupport.setCurrentBlocker(this); // for back-compatibility
 ? ? ? ? ? ?boolean interrupted = false, cancelled = false, rejected = false;
 ? ? ? ? ? ?while (!canReacquire(node)) {
 ? ? ? ? ? ? ? ?if (interrupted |= Thread.interrupted()) {
 ? ? ? ? ? ? ? ? ? ?if (cancelled = (node.getAndUnsetStatus(COND) & COND) != 0)
 ? ? ? ? ? ? ? ? ? ? ? ?break; ? ? ? ? ? ? ?// else interrupted after signal
 ? ? ? ? ? ? ?  } else if ((node.status & COND) != 0) {
 ? ? ? ? ? ? ? ? ? ?try {
 ? ? ? ? ? ? ? ? ? ? ? ?if (rejected)
 ? ? ? ? ? ? ? ? ? ? ? ? ? ?node.block();
 ? ? ? ? ? ? ? ? ? ? ? ?else
 ? ? ? ? ? ? ? ? ? ? ? ? ? ?ForkJoinPool.managedBlock(node);
 ? ? ? ? ? ? ? ? ?  } catch (RejectedExecutionException ex) {
 ? ? ? ? ? ? ? ? ? ? ? ?rejected = true;
 ? ? ? ? ? ? ? ? ?  } catch (InterruptedException ie) {
 ? ? ? ? ? ? ? ? ? ? ? ?interrupted = true;
 ? ? ? ? ? ? ? ? ?  }
 ? ? ? ? ? ? ?  } else
 ? ? ? ? ? ? ? ? ? ?Thread.onSpinWait(); ? ?// awoke while enqueuing
 ? ? ? ? ?  }
 ? ? ? ? ? ?LockSupport.setCurrentBlocker(null);
 ? ? ? ? ? ?node.clearStatus();
 ? ? ? ? ? ?acquire(node, savedState, false, false, false, 0L);
 ? ? ? ? ? ?if (interrupted) {
 ? ? ? ? ? ? ? ?if (cancelled) {
 ? ? ? ? ? ? ? ? ? ?unlinkCancelledWaiters(node);
 ? ? ? ? ? ? ? ? ? ?throw new InterruptedException();
 ? ? ? ? ? ? ?  }
 ? ? ? ? ? ? ? ?Thread.currentThread().interrupt();
 ? ? ? ? ?  }
 ? ? ?  }

await() 方法相較于 awaitUninterruptibly()，while 邏輯基本一致，最后多了一步 cancelled 狀態(tài)檢查，如果 cancelled = true ，調(diào)用 unlinkCancelledWaiters(node)，去清理等待隊(duì)列。

awaitNanos

awaitNanos(long) 在 await() 之上多了對(duì)超時(shí)時(shí)間的計(jì)算和處理邏輯：

 ? ? ? ?public final long awaitNanos(long nanosTimeout)
 ? ? ? ? ? ? ? ?throws InterruptedException {
 ? ? ? ? ? ?if (Thread.interrupted())
 ? ? ? ? ? ? ? ?throw new InterruptedException();
 ? ? ? ? ? ?ConditionNode node = new ConditionNode();
 ? ? ? ? ? ?int savedState = enableWait(node);
 ? ? ? ? ? ?long nanos = (nanosTimeout < 0L) ? 0L : nanosTimeout;
 ? ? ? ? ? ?long deadline = System.nanoTime() + nanos;
 ? ? ? ? ? ?boolean cancelled = false, interrupted = false;
 ? ? ? ? ? ?while (!canReacquire(node)) {
 ? ? ? ? ? ? ? ?if ((interrupted |= Thread.interrupted()) ||
 ? ? ? ? ? ? ? ? ?  (nanos = deadline - System.nanoTime()) <= 0L) { // 多了一個(gè)超時(shí)條件
 ? ? ? ? ? ? ? ? ? ?if (cancelled = (node.getAndUnsetStatus(COND) & COND) != 0)
 ? ? ? ? ? ? ? ? ? ? ? ?break;
 ? ? ? ? ? ? ?  } else
 ? ? ? ? ? ? ? ? ? ?LockSupport.parkNanos(this, nanos);
 ? ? ? ? ?  }
 ? ? ? ? ? ?node.clearStatus();
 ? ? ? ? ? ?acquire(node, savedState, false, false, false, 0L);
 ? ? ? ? ? ?if (cancelled) {
 ? ? ? ? ? ? ? ?unlinkCancelledWaiters(node);
 ? ? ? ? ? ? ? ?if (interrupted)
 ? ? ? ? ? ? ? ? ? ?throw new InterruptedException();
 ? ? ? ? ?  } else if (interrupted)
 ? ? ? ? ? ? ? ?Thread.currentThread().interrupt();
 ? ? ? ? ? ?long remaining = deadline - System.nanoTime(); // avoid overflow
 ? ? ? ? ? ?return (remaining <= nanosTimeout) ? remaining : Long.MIN_VALUE;
 ? ? ?  }

awaitUntil

awaitUntil(Date) 和 awaitNanos(long) 同理，只是將超時(shí)計(jì)算改成了日期計(jì)算：

 ? ? ? ? ? ?long abstime = deadline.getTime();
 ? ? ? ? ? ?// ...
 ? ? ? ? ? ?boolean cancelled = false, interrupted = false;
 ? ? ? ? ? ?while (!canReacquire(node)) {
 ? ? ? ? ? ? ? ?if ((interrupted |= Thread.interrupted()) ||
 ? ? ? ? ? ? ? ? ? ?System.currentTimeMillis() >= abstime) { // 時(shí)間檢查
 ? ? ? ? ? ? ? ? ? ?if (cancelled = (node.getAndUnsetStatus(COND) & COND) != 0)
 ? ? ? ? ? ? ? ? ? ? ? ?break;
 ? ? ? ? ? ? ?  } else
 ? ? ? ? ? ? ? ? ? ?LockSupport.parkUntil(this, abstime);
 ? ? ? ? ?  }

await(long, TimeUnit)

await(long, TimeUnit) 則是邏輯更加與 awaitNanos(long) 相似了, 只是多了一步計(jì)算 awaitNanos(long nanosTimeout) 中的參數(shù) nanosTimeout 的操作：

long nanosTimeout = unit.toNanos(time);

acquire 方法

在 wait 方法組中，最終都會(huì)調(diào)用到這個(gè)邏輯：

 ? ?final int acquire(Node node, int arg, boolean shared, boolean interruptible, boolean timed, long time) {
 ? ? ? ?Thread current = Thread.currentThread();
 ? ? ? ?byte spins = 0, postSpins = 0; ? // 在取消第一個(gè)線程時(shí)重試
 ? ? ? ?boolean interrupted = false, first = false;
 ? ? ? ?Node pred = null; ? ? ? ? ? ? ? ?// 入隊(duì)時(shí)節(jié)點(diǎn)的前一個(gè)指針
 ? ? ? ?/*
 ? ? ? ? * 反復(fù)執(zhí)行:
 ? ? ? ? *  檢查當(dāng)前節(jié)點(diǎn)是否是 first
 ? ? ? ? *  若是, 確保 head 穩(wěn)定，否則確保有效的 prev
 ? ? ? ? *  如果節(jié)點(diǎn)是第一個(gè)或尚未入隊(duì)，嘗試獲取
 ? ? ? ? *  否則，如果節(jié)點(diǎn)尚未創(chuàng)建，則創(chuàng)建這個(gè)它
 ? ? ? ? *  否則，如果節(jié)點(diǎn)尚未入隊(duì)，嘗試入隊(duì)一次
 ? ? ? ? *  否則，如果通過(guò) park 喚醒，重試，最多 postSpins 次
 ? ? ? ? *  否則，如果 WAITING 狀態(tài)未設(shè)置，設(shè)置并重試
 ? ? ? ? *  否則，park 并且清除 WAITING 狀態(tài), 檢查取消邏輯
 ? ? ? ? */
 ? ? ? ?for (;;) {
 ? ? ? ? ? ?if (!first && (pred = (node == null) ? null : node.prev) != null && !(first = (head == pred))) {
 ? ? ? ? ? ? ? ?if (pred.status < 0) {
 ? ? ? ? ? ? ? ? ? ?cleanQueue(); ? ? ? ? ? // predecessor cancelled
 ? ? ? ? ? ? ? ? ? ?continue;
 ? ? ? ? ? ? ?  } else if (pred.prev == null) {
 ? ? ? ? ? ? ? ? ? ?Thread.onSpinWait(); ? ?// ensure serialization
 ? ? ? ? ? ? ? ? ? ?continue;
 ? ? ? ? ? ? ?  }
 ? ? ? ? ?  }
 ? ? ? ? ? ?if (first || pred == null) {
 ? ? ? ? ? ? ? ?boolean acquired;
 ? ? ? ? ? ? ? ?try {
 ? ? ? ? ? ? ? ? ? ?if (shared)
 ? ? ? ? ? ? ? ? ? ? ? ?acquired = (tryAcquireShared(arg) >= 0);
 ? ? ? ? ? ? ? ? ? ?else
 ? ? ? ? ? ? ? ? ? ? ? ?acquired = tryAcquire(arg);
 ? ? ? ? ? ? ?  } catch (Throwable ex) {
 ? ? ? ? ? ? ? ? ? ?cancelAcquire(node, interrupted, false);
 ? ? ? ? ? ? ? ? ? ?throw ex;
 ? ? ? ? ? ? ?  }
 ? ? ? ? ? ? ? ?if (acquired) {
 ? ? ? ? ? ? ? ? ? ?if (first) {
 ? ? ? ? ? ? ? ? ? ? ? ?node.prev = null;
 ? ? ? ? ? ? ? ? ? ? ? ?head = node;
 ? ? ? ? ? ? ? ? ? ? ? ?pred.next = null;
 ? ? ? ? ? ? ? ? ? ? ? ?node.waiter = null;
 ? ? ? ? ? ? ? ? ? ? ? ?if (shared)
 ? ? ? ? ? ? ? ? ? ? ? ? ? ?signalNextIfShared(node);
 ? ? ? ? ? ? ? ? ? ? ? ?if (interrupted)
 ? ? ? ? ? ? ? ? ? ? ? ? ? ?current.interrupt();
 ? ? ? ? ? ? ? ? ?  }
 ? ? ? ? ? ? ? ? ? ?return 1;
 ? ? ? ? ? ? ?  }
 ? ? ? ? ?  }
 ? ? ? ? ? ?if (node == null) { ? ? ? ? ? ? ? ? // allocate; retry before enqueue
 ? ? ? ? ? ? ? ?if (shared)
 ? ? ? ? ? ? ? ? ? ?node = new SharedNode();
 ? ? ? ? ? ? ? ?else
 ? ? ? ? ? ? ? ? ? ?node = new ExclusiveNode();
 ? ? ? ? ?  } else if (pred == null) { ? ? ? ? ?// try to enqueue
 ? ? ? ? ? ? ? ?node.waiter = current;
 ? ? ? ? ? ? ? ?Node t = tail;
 ? ? ? ? ? ? ? ?node.setPrevRelaxed(t); ? ? ? ? // avoid unnecessary fence
 ? ? ? ? ? ? ? ?if (t == null)
 ? ? ? ? ? ? ? ? ? ?tryInitializeHead();
 ? ? ? ? ? ? ? ?else if (!casTail(t, node))
 ? ? ? ? ? ? ? ? ? ?node.setPrevRelaxed(null); ?// back out
 ? ? ? ? ? ? ? ?else
 ? ? ? ? ? ? ? ? ? ?t.next = node;
 ? ? ? ? ?  } else if (first && spins != 0) {
 ? ? ? ? ? ? ? ?--spins; ? ? ? ? ? ? ? ? ? ? ? ?// reduce unfairness on rewaits
 ? ? ? ? ? ? ? ?Thread.onSpinWait();
 ? ? ? ? ?  } else if (node.status == 0) {
 ? ? ? ? ? ? ? ?node.status = WAITING; ? ? ? ? ?// enable signal and recheck
 ? ? ? ? ?  } else {
 ? ? ? ? ? ? ? ?long nanos;
 ? ? ? ? ? ? ? ?spins = postSpins = (byte)((postSpins << 1) | 1);
 ? ? ? ? ? ? ? ?if (!timed)
 ? ? ? ? ? ? ? ? ? ?LockSupport.park(this);
 ? ? ? ? ? ? ? ?else if ((nanos = time - System.nanoTime()) > 0L)
 ? ? ? ? ? ? ? ? ? ?LockSupport.parkNanos(this, nanos);
 ? ? ? ? ? ? ? ?else
 ? ? ? ? ? ? ? ? ? ?break;
 ? ? ? ? ? ? ? ?node.clearStatus();
 ? ? ? ? ? ? ? ?if ((interrupted |= Thread.interrupted()) && interruptible)
 ? ? ? ? ? ? ? ? ? ?break;
 ? ? ? ? ?  }
 ? ? ?  }
 ? ? ? ?return cancelAcquire(node, interrupted, interruptible);
 ?  }

這個(gè)方法會(huì)在 Node 關(guān)聯(lián)的線程讓出鎖資源后，開啟一個(gè)死循環(huán)嘗試通過(guò) tryAcquire 嘗試獲取鎖資源，最后如果超時(shí)或嘗試次數(shù)超出限制，會(huì)通過(guò) LockSupport.park 阻塞自身。

到此這篇關(guān)于Java 多線程并發(fā)AbstractQueuedSynchronizer詳情的文章就介紹到這了,更多相關(guān)Java AbstractQueuedSynchronizer內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

亚洲乱码中文字幕综合,中国熟女仑乱hd,亚洲精品乱拍国产一区二区三区,一本大道卡一卡二卡三乱码全集资源,又粗又黄又硬又爽的免费视频

軟件下載

源碼下載

軟件編程

網(wǎng)絡(luò)編程

在線工具

數(shù)據(jù)庫(kù)

CMS

常用工具