源碼級深挖AQS隊(duì)列同步器
源 /碼農(nóng)參上 文/ Dr Hydra
我們知道，在java中提供了兩類鎖的實(shí)現(xiàn)，一種是在jvm層級上實(shí)現(xiàn)的synchrinized隱式鎖，另一類是jdk在代碼層級實(shí)現(xiàn)的，juc包下的Lock顯示鎖，而提到Lock就不得不提一下它的核心隊(duì)列同步器（AQS）了，它的全稱是AbstractQueuedSynchronizer，是用來構(gòu)建鎖或者其他一些同步組件的基礎(chǔ)，除了ReentrantLock、ReentrantReadWriteLock外，它還在CountDownLatch、Semaphore以及ThreadPoolExecutor中被使用，通過理解隊(duì)列同步器的工作原理，對我們了解和使用這些工具類會有很大的幫助。
1、AQS 基礎(chǔ)
為了便于理解AQS的概念，這里首先摘錄部分AbstractQueuedSynchronizer的注釋進(jìn)行了簡要翻譯：
它提供了一個框架，對于依賴先進(jìn)先出等待隊(duì)列的阻塞鎖和同步器（例如信號量和事件），可以用它來實(shí)現(xiàn)。這個類的設(shè)計(jì)，對于大多數(shù)依賴于單個原子值來表示狀態(tài)（state）的同步器，可以提供有力的基礎(chǔ)。子類需要重寫被protected修飾的方法，例如更改狀態(tài)（state），定義在獲取或釋放對象時這些狀態(tài)表示的含義?；谶@些，類中的其他方法實(shí)現(xiàn)了隊(duì)列和阻塞機(jī)制。在子類中可以維護(hù)其他的狀態(tài)字段，但是只有使用getState，setState，compareAndSetState方法原子更新的狀態(tài)值變量，才與同步有關(guān)。
子類被推薦定義為非public的內(nèi)部類，用來實(shí)現(xiàn)封閉類的屬性同步。同步器本身沒有實(shí)現(xiàn)任何同步接口，它僅僅定義了一些方法，供具體的鎖和同步組件中的public方法調(diào)用。
隊(duì)列同步器支持獨(dú)占模式和共享模式，當(dāng)一個線程在獨(dú)占模式下獲取時，其他線程不能獲取成功，在共享模式下多線程的獲取可能成功。在不同模式下，等待的線程使用的是相同的先進(jìn)先出隊(duì)列。通常，實(shí)現(xiàn)子類只支持其中的一種模式，但是在ReadWriteLock中兩者都可以發(fā)揮作用。只支持一種模式的子類在實(shí)現(xiàn)時不需要重寫另一種模式中的方法。
閱讀這些注釋，可以知道AbstractQueuedSynchronizer是一個抽象類，它基于內(nèi)部先進(jìn)先出（FIFO）的雙向隊(duì)列、以及內(nèi)置的一些protected方法來實(shí)現(xiàn)同步器，完成同步狀態(tài)的管理，并且我們可以通過子類繼承AQS抽象類的方式，在共享模式或獨(dú)占模式下，實(shí)現(xiàn)自定義的同步組件。
通過上面的描述，可以看出AQS中的兩大核心是同步狀態(tài)和雙向的同步隊(duì)列，來看一下源碼中是如何對它們進(jìn)行定義的：
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer implements java.io.Serializable {
static final class Node {
volatile int waitStatus;
volatile Node prev;
volatile Node next;
volatile Thread thread;
//...
}
private transient volatile Node head;
private transient volatile Node tail;
private volatile int state;
//...
}
下面針對這兩個核心內(nèi)容分別進(jìn)行研究。
同步隊(duì)列
AQS內(nèi)部靜態(tài)類Node用于表示同步隊(duì)列中的節(jié)點(diǎn)，變量表示的意義如下：
prev：當(dāng)前節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)，如果當(dāng)前節(jié)點(diǎn)是同步隊(duì)列的頭節(jié)點(diǎn)，那么prev為nullnext：當(dāng)前節(jié)點(diǎn)的后繼節(jié)點(diǎn)，如果當(dāng)前節(jié)點(diǎn)是同步隊(duì)列的尾節(jié)點(diǎn)，那么next為nullthread：獲取同步狀態(tài)的線程waitStatus：等待狀態(tài)，取值可為以下情況CANCELLED(1)：表示當(dāng)前節(jié)點(diǎn)對應(yīng)的線程被取消，當(dāng)線程等待超時或被中斷時會被修改為此狀態(tài)SIGNAL(-1)：當(dāng)前節(jié)點(diǎn)的后繼節(jié)點(diǎn)的線程被阻塞，當(dāng)前線程在釋放同步狀態(tài)或取消時，需要喚醒后繼節(jié)點(diǎn)的線程CONDITION(-2)：節(jié)點(diǎn)處于等待隊(duì)列中，節(jié)點(diǎn)線程等待在Condition上，當(dāng)其他線程調(diào)用Condition的signal方法后，會將節(jié)點(diǎn)從等待隊(duì)列移到同步隊(duì)列PROPAGATE(-3)：表示下一次共享式同步狀態(tài)獲取能夠被執(zhí)行，即同步狀態(tài)的獲取可以向后繼節(jié)點(diǎn)的后繼進(jìn)行無條件的傳播0：初始值，表示當(dāng)前節(jié)點(diǎn)等待獲取同步狀態(tài)
每個節(jié)點(diǎn)的prev和next指針在加入隊(duì)列的時候進(jìn)行賦值，通過這些指針就形成了一個雙向列表，另外AQS還保存了同步隊(duì)列的頭節(jié)點(diǎn)head和尾節(jié)點(diǎn)tail，通過這樣的結(jié)構(gòu)，就能夠通過頭節(jié)點(diǎn)或尾節(jié)點(diǎn)，找到隊(duì)列中的任何一個節(jié)點(diǎn)。使用圖來表示同步隊(duì)列的結(jié)構(gòu)如下：
另外可以看到，在源碼中為了保證可見性，同步器中的head、tail、state，以及節(jié)點(diǎn)中的prev，next屬性都加了關(guān)鍵字volatile修飾。
同步狀態(tài)
AQS的另一核心同步狀態(tài)，在代碼中是使用int類型的變量state來表示的，通過原子操作修改同步狀態(tài)的值，來實(shí)現(xiàn)對同步組件的狀態(tài)進(jìn)行修改。在子類中，主要通過AQS提供的下面3個方法對同步狀態(tài)的訪問和轉(zhuǎn)換進(jìn)行操作：
getState()：獲取當(dāng)前的同步狀態(tài)setState(int newState)：設(shè)置新的同步狀態(tài)compareAndSetState(int expect,int update)：調(diào)用Unsafe類的compareAndSwapInt方法，使用CAS操作更新同步狀態(tài)，保證了狀態(tài)修改的原子性
線程會試圖修改state的值，如果修改成功那么表示線程得到或釋放了同步狀態(tài)，如果失敗就會將當(dāng)前線程封裝成一個Node節(jié)點(diǎn)，然后將其加入到同步隊(duì)列中，并阻塞當(dāng)前線程。
設(shè)計(jì)思想
AQS的設(shè)計(jì)使用了模板方法的設(shè)計(jì)模式，模板方法一般在父類中封裝不變的部分（如算法骨架），把擴(kuò)展的可變部分交給子類進(jìn)行擴(kuò)展，子類的執(zhí)行結(jié)果會影響父類的結(jié)果，是一種反向的控制結(jié)構(gòu)。AQS中應(yīng)用了這種設(shè)計(jì)模式，將一部分方法交給子類進(jìn)行重寫，而自定義的同步組件在調(diào)用同步器提供的模板方法（父類中的方法）時，又會調(diào)用子類重寫的方法。
以AQS類中常用于獲取鎖的acquire方法為例，它的代碼如下：
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
acquire方法被final修飾，不可以在子類中重寫，因?yàn)樗菍ν馓峁┑哪０宸椒?，有相對具體和固定的執(zhí)行邏輯。在acquire方法中調(diào)用了tryAcquire方法：
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
可以看到帶有protected修飾的tryAcquire方法是一個空殼方法，并沒有定義實(shí)際獲取同步狀態(tài)的邏輯，這就需要我們在繼承AQS的子類中對齊進(jìn)行重寫，從而達(dá)到擴(kuò)展目的。在重寫過程中，就會用到上面提到的獲取和修改同步狀態(tài)的3個方法getState、setState和compareAndSetState。
以ReentrantLock中的方法調(diào)用為例，當(dāng)調(diào)用ReentrantLock中的lock方法時，會調(diào)用繼承了AQS的內(nèi)部類Sync的父類中的acquire方法，acquire方法再調(diào)用子類Sync的tryAcquire方法并返回boolean類型結(jié)果。
除了tryAcquire方法外，子類中還提供了其他可以重寫的方法，列出如下：
tryAcquire：獨(dú)占式獲取同步狀態(tài)tryRelease：獨(dú)占式釋放同步狀態(tài)tryAcquireShared：共享式獲取同步狀態(tài)tryReleaseShared：共享式釋放同步狀態(tài)isHeldExclusively：當(dāng)前線程是否獨(dú)占式的占用同步狀態(tài)
而我們在實(shí)現(xiàn)自定義的同步組件時，可以直接調(diào)用AQS提供的下面這些模板方法：
acquire：獨(dú)占式獲取同步狀態(tài)，如果線程獲取同步狀態(tài)成功那么方法返回，否則線程阻塞，進(jìn)入同步隊(duì)列中acquireInterruptibly：在acquire基礎(chǔ)上，添加了響應(yīng)中斷功能tryAcquireNanos：在acquireInterruptibly基礎(chǔ)上，添加了超時限制，超時會返回falseacquireShared：共享式獲取同步狀態(tài)，如果線程獲取同步狀態(tài)成功那么方法返回，否則線程進(jìn)入同步隊(duì)列中阻塞。與acquire不同，該方法允許多個線程同時獲取鎖acquireSharedInterruptibly：在acquireShared基礎(chǔ)上，可響應(yīng)中斷tryAcquireSharedNanos：在acquireSharedInterruptibly基礎(chǔ)上，添加了超時限制release：獨(dú)占式釋放同步狀態(tài)，將喚醒同步隊(duì)列中第一個節(jié)點(diǎn)的線程releaseShared：共享式釋放同步狀態(tài)getQueuedThreads：獲取等待在同步隊(duì)列上的線程集合
從模板方法中可以看出，大多方法都是獨(dú)占模式和共享模式對稱出現(xiàn)的，除去查詢等待線程方法外，可以將他們分為兩類：獨(dú)占式獲取或釋放同步狀態(tài)、共享式獲取或釋放同步狀態(tài)，并且它們的核心都是acquire與release方法，其他方法只是在它們實(shí)現(xiàn)的基礎(chǔ)上做了部分的邏輯改動，增加了中斷和超時功能的支持。下面對主要的4個方法進(jìn)行分析。
2、源碼分析
acquire
分析上面acquire方法中源碼的執(zhí)行流程：
1.首先調(diào)用tryAcquire嘗試獲取同步狀態(tài)，如果獲取成功，那么直接返回
2.如果獲取同步狀態(tài)失敗，調(diào)用addWaiter方法生成新Node節(jié)點(diǎn)并加入同步隊(duì)列：
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
方法中使用當(dāng)前線程和等待狀態(tài)構(gòu)造了一個新的Node節(jié)點(diǎn)，在同步隊(duì)列的隊(duì)尾節(jié)點(diǎn)不為空的情況下（說明同步隊(duì)列非空），調(diào)用compareAndSetTail方法以CAS的方式把新節(jié)點(diǎn)設(shè)置為同步隊(duì)列的隊(duì)尾節(jié)點(diǎn)。如果隊(duì)尾節(jié)點(diǎn)為空或添加新節(jié)點(diǎn)失敗，則調(diào)用enq方法：
private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
在同步隊(duì)列為空的情況下，會先創(chuàng)建一個新的空節(jié)點(diǎn)作為頭節(jié)點(diǎn)，然后通過CAS的方式將當(dāng)前線程創(chuàng)建的Node設(shè)為尾節(jié)點(diǎn)。在for循環(huán)中，只有通過CAS將節(jié)點(diǎn)插入到隊(duì)尾后才會返回，否則就會重復(fù)循環(huán)，通過這樣的方式，能夠?qū)⒉l(fā)添加節(jié)點(diǎn)的操作變?yōu)榇刑砑?，保證了線程的安全性。這一過程可以使用下圖表示：
3.添加新節(jié)點(diǎn)完成后，調(diào)用acquireQueued方法，嘗試以自旋的方式獲取同步狀態(tài)：
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
當(dāng)新添加的Node的前驅(qū)節(jié)點(diǎn)是同步隊(duì)列的頭節(jié)點(diǎn)并且嘗試獲取同步狀態(tài)成功時，線程將Node設(shè)為頭節(jié)點(diǎn)并從自旋中退出，否則調(diào)用shouldParkAfterFailedAcquire方法判斷是否需要掛起：
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
return true;
if (ws > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
在該方法中，傳入的第一個Node類型的參數(shù)是當(dāng)前節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)，對其等待狀態(tài)進(jìn)行判斷：
如果為
SIGNAL狀態(tài)，那么前驅(qū)節(jié)點(diǎn)釋放同步狀態(tài)或取消時都會通知后繼節(jié)點(diǎn)，因此可以將當(dāng)前線程阻塞，返回true如果大于0，那么為
CANCEL狀態(tài)，表示前驅(qū)節(jié)點(diǎn)被取消，那么一直向前回溯，找到一個不為CANCEL狀態(tài)的節(jié)點(diǎn)，并將當(dāng)前節(jié)點(diǎn)的前驅(qū)指向它如果不是上面的兩種情況，那么將前驅(qū)節(jié)點(diǎn)的等待狀態(tài)設(shè)為
SIGNAL。這里的目的是在每個節(jié)點(diǎn)進(jìn)入阻塞狀態(tài)前將前驅(qū)節(jié)點(diǎn)的等待狀態(tài)設(shè)為SIGNAL，否則節(jié)點(diǎn)將無法被喚醒在后兩種情況下，都會返回
false，然后在acquireQueued方法中進(jìn)行循環(huán)，直到進(jìn)入shouldParkAfterFailedAcquire方法時為第一種情況，阻塞線程
當(dāng)返回為true時，調(diào)用parkAndCheckInterrupt方法：
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
在方法內(nèi)部調(diào)用了LockSupport的park方法，阻塞當(dāng)前線程，并返回當(dāng)前線程是否被中斷的狀態(tài)。
在上面的代碼中，各節(jié)點(diǎn)通過自旋的方式檢測自己的前驅(qū)節(jié)點(diǎn)是否頭節(jié)點(diǎn)的過程，可用下圖表示：
4.當(dāng)滿足條件，返回acquire方法后，調(diào)用selfInterrupt方法。方法內(nèi)部使用interrupt方法，喚醒被阻塞的線程，繼續(xù)向下執(zhí)行：
static void selfInterrupt() {
Thread.currentThread().interrupt();
}
最后，使用流程圖的方式總結(jié)acquire方法獨(dú)占式獲取鎖的整體流程：
release
與acquire方法對應(yīng)，release方法負(fù)責(zé)獨(dú)占式釋放同步狀態(tài)，流程也相對簡單。在ReentrantLock中，unlock方法就是直接調(diào)用的AQS的release方法。先來直接看一下它的源碼：
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
1.方法中首先調(diào)用子類重寫的tryRelease方法，嘗試釋放當(dāng)前線程持有的同步狀態(tài)，如果成功則向下執(zhí)行，失敗返回false
2.如果同步隊(duì)列的頭節(jié)點(diǎn)不為空且等待狀態(tài)不為初始狀態(tài)，那么將調(diào)用unparkSuccessor方法喚醒它的后繼節(jié)點(diǎn)：
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
Node s = node.next;
if (s == null || s.waitStatus > 0) {
s = null;
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
if (s != null)
LockSupport.unpark(s.thread);
}
方法主要實(shí)現(xiàn)的功能有：
如果頭節(jié)點(diǎn)的等待狀態(tài)小于0，使用CAS將它置為0
如果后續(xù)節(jié)點(diǎn)為空、或它的等待狀態(tài)為
CANCEL被取消，那么從隊(duì)尾開始，向前尋找最靠近隊(duì)列頭部的一個等待狀態(tài)小于0 的節(jié)點(diǎn)找到符合條件的節(jié)點(diǎn)后，調(diào)用
LockSupport工具類的unpark方法，喚醒后繼節(jié)點(diǎn)中對應(yīng)的線程
同步隊(duì)列新頭節(jié)點(diǎn)的設(shè)置過程如下圖所示：
在上面的過程中，采用的是從后向前遍歷尋找未取消節(jié)點(diǎn)的方式，這是因?yàn)锳QS的同步隊(duì)列是一個弱一致性的雙向列表，在下面的情況中，存在next指針為null的情況：
在
enq方法插入新節(jié)點(diǎn)時，可能存在舊尾節(jié)點(diǎn)的next指針還未指向新節(jié)點(diǎn)的情況在
shouldParkAfterFailedAcquire方法中，當(dāng)移除CANCEL狀態(tài)的節(jié)點(diǎn)時，也存在next指針還未指向后續(xù)節(jié)點(diǎn)的情況
acquireShared
在了解了獨(dú)占式獲取同步狀態(tài)后，再來看一下共享式獲取同步狀態(tài)。在共享模式下，允許多個線程同時獲取到同步狀態(tài)，來看一下它的源碼：
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
首先調(diào)用子類重寫的tryAcquireShared方法，返回值為int類型，如果值大于等于0表示獲取同步狀態(tài)成功，那么直接返回。如果小于0表示獲取失敗，執(zhí)行下面的doAcquireShared方法，將線程放入等待隊(duì)列使用自旋嘗試獲取，直到獲取同步狀態(tài)成功：
private void doAcquireShared(int arg) {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
對上面的代碼進(jìn)行簡要的解釋：
1.調(diào)用addWaiter方法，封裝新節(jié)點(diǎn)，并以共享模式（Node.SHARED）將節(jié)點(diǎn)放入同步隊(duì)列中隊(duì)尾
2.在for循環(huán)中，獲取當(dāng)前節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)，如果前驅(qū)節(jié)點(diǎn)是同步隊(duì)列的頭節(jié)點(diǎn)，那么就以共享模式去嘗試獲取同步狀態(tài)，判斷tryAcquireShared的返回值，如果返回值大于等于0，表示獲取同步狀態(tài)成功，修改新的頭節(jié)點(diǎn)，并將信息傳播給同步隊(duì)列中的后繼節(jié)點(diǎn)，然后檢查中斷標(biāo)志位，如果線程被阻塞，那么進(jìn)行喚醒
3.如果前驅(qū)節(jié)點(diǎn)不是頭節(jié)點(diǎn)、或獲取同步狀態(tài)失敗時，調(diào)用shouldParkAfterFailedAcquire判斷是否需要阻塞，如果需要則調(diào)用parkAndCheckInterrupt，在前驅(qū)節(jié)點(diǎn)的等待狀態(tài)為SIGNAL時，將節(jié)點(diǎn)對應(yīng)的線程阻塞
可以看到，共享式的獲取同步狀態(tài)的調(diào)用過程和acquire方法非常相似，但不同的是在獲取同步狀態(tài)成功后，會調(diào)用setHeadAndPropagate方法進(jìn)行共享式同步狀態(tài)的傳播：
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
setHead(node);
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
doReleaseShared();
}
}
因?yàn)楣蚕硎酵綘顟B(tài)是允許多個線程共享的，所以在一個線程獲取到同步狀態(tài)后，需要在第一時間通知后繼節(jié)點(diǎn)的線程可以嘗試獲取同步資源，這樣就可以避免其他線程阻塞時間過長。在方法中，把當(dāng)前節(jié)點(diǎn)設(shè)置為頭節(jié)點(diǎn)后，需要根據(jù)情況判斷后繼節(jié)點(diǎn)是否需要釋放：
propagate>0：表示還擁有剩余的同步資源，從doAcquireShared方法中執(zhí)行到這時，取值是大于等于0的，在等于0的情況下，會繼續(xù)下面的判斷h == null：原頭節(jié)點(diǎn)為空，一般情況下不滿足，有可能發(fā)生在原頭節(jié)點(diǎn)被gc回收的情況，此條不滿足情況則向下繼續(xù)判斷h.waitStatus < 0：原頭節(jié)點(diǎn)的等待狀態(tài)可能取值為0或-3當(dāng)某個線程釋放同步資源或者前一個節(jié)點(diǎn)共享式獲取同步狀態(tài)時（會執(zhí)行下面的
doReleaseShared方法），會將自己的waitStatus從-1改變?yōu)?這時可能后繼節(jié)點(diǎn)還沒有來的及將自己更新為頭節(jié)點(diǎn)，如果有其他的線程在這個時候再調(diào)用
doReleaseShared方法，那么取到的還是原頭節(jié)點(diǎn)，會把它的waitStatus從0改變?yōu)?3，在這個過程中，說明其他線程調(diào)用doReleaseShared釋放了同步資源(h = head) == null：新頭節(jié)點(diǎn)為空，一般情況下不滿足，會向下繼續(xù)判斷h.waitStatus < 0：新頭節(jié)點(diǎn)的等待狀態(tài)可能取值為0或-3或-1如果后繼節(jié)點(diǎn)剛加入隊(duì)列，還沒有運(yùn)行到
shouldParkAfterFailedAcquire方法，修改其前驅(qū)節(jié)點(diǎn)的等待狀態(tài)時，此時可能為0如果節(jié)點(diǎn)被喚醒成為了新的頭節(jié)點(diǎn)，并且此時后繼節(jié)點(diǎn)才剛被加入同步隊(duì)列，又有其他線程釋放鎖調(diào)用了
doReleaseShared，會把頭節(jié)點(diǎn)的狀態(tài)從0改為-3隊(duì)列中的節(jié)點(diǎn)已經(jīng)調(diào)用了
shouldParkAfterFailedAcquire，會把waitStatus從0或-3 改為-1
如果滿足上面的任何一種狀態(tài)，并且它的后繼節(jié)點(diǎn)是SHARED狀態(tài)的，則執(zhí)行doReleaseShared方法釋放后繼節(jié)點(diǎn)：
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
doReleaseShared方法不僅在這里的共享狀態(tài)傳播的情況下被調(diào)用，還會在后面介紹的共享式釋放同步狀態(tài)中被調(diào)用。在方法中，當(dāng)頭節(jié)點(diǎn)不為空且不等于尾節(jié)點(diǎn)（意味著沒有后繼節(jié)點(diǎn)需要等待喚醒）時：
先將頭節(jié)點(diǎn)從
SIGNAL狀態(tài)更新為0，然后調(diào)用unparkSuccessor方法喚醒頭節(jié)點(diǎn)的后繼節(jié)點(diǎn)將頭節(jié)點(diǎn)的狀態(tài)從0更新為
PROPAGATE，表明狀態(tài)需要向后繼節(jié)點(diǎn)傳播如果頭節(jié)點(diǎn)在更新狀態(tài)的時候沒有發(fā)生改變，則退出循環(huán)
通過上面的流程，就實(shí)現(xiàn)了從頭節(jié)點(diǎn)嘗試向后喚醒節(jié)點(diǎn)，實(shí)現(xiàn)了共享狀態(tài)的向后傳播。
releaseShared
最后，再來看一下對應(yīng)的共享式釋放同步狀態(tài)方法：
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
releaseShared方法會釋放指定量的資源，如果調(diào)用子類重寫的tryReleaseShared方法返回值為true，表示釋放成功，那么還是執(zhí)行上面介紹過的doReleaseShared方法喚醒同步隊(duì)列中的等待線程。
3、自定義同步組件
在前面的介紹中說過，在使用AQS時，需要定義一個子類繼承AbstractQueuedSynchronizer抽象類，并實(shí)現(xiàn)它的抽象方法來管理同步狀態(tài)。接下來我們就來手寫一個獨(dú)占式的鎖，按照文檔中的推薦，我們將子類定義為自定義同步工具類的靜態(tài)內(nèi)部類：
public class MyLock {
private static class AqsHelper extends AbstractQueuedSynchronizer {
@Override
protected boolean tryAcquire(int arg) {
int state = getState();
if (state == 0) {
if (compareAndSetState(0, arg)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
} else if (getExclusiveOwnerThread() == Thread.currentThread()) {
setState(getState() + arg);
return true;
}
return false;
}
@Override
protected boolean tryRelease(int arg) {
int state = getState() - arg;
if (state == 0) {
setExclusiveOwnerThread(null);
setState(state);
return true;
}
setState(state);
return false;
}
@Override
protected boolean isHeldExclusively() {
return getState() == 1;
}
}
private final AqsHelper aqsHelper = new AqsHelper();
public void lock() {
aqsHelper.acquire(1);
}
public boolean tryLock() {
return aqsHelper.tryAcquire(1);
}
public void unlock() {
aqsHelper.release(1);
}
public boolean isLocked() {
return aqsHelper.isHeldExclusively();
}
}
在AQS的子類中，首先重寫了tryAcquire方法，在方法中利用CAS來修改state的狀態(tài)值，并在修改成功時設(shè)置當(dāng)前線程獨(dú)占資源。并且通過比較嘗試獲取鎖的線程與持有鎖的線程是否相同的方式，來實(shí)現(xiàn)了鎖的可重入性。在重寫的tryRelease方法中，進(jìn)行資源的釋放，如果存在重入的情況，會一直到所有重入鎖釋放完才會真正的釋放鎖，并放棄占有狀態(tài)。
可以注意到在自定義的鎖工具類中，我們定義了lock和tryLock兩個方法，分別調(diào)用了acquire和tryAcquire方法，它們的區(qū)別是lock會等待鎖資源，直到成功時才會返回，而tryLock嘗試獲取鎖時，會立即返回成功或失敗的狀態(tài)。
接下來，我們通過下面的測試代碼，驗(yàn)證自定義的鎖的有效性：
public class Test {
private MyLock lock=new MyLock();
private int i=0;
public void sayHi(){
try {
lock.lock();
System.out.println("i am "+i++);
}finally {
lock.unlock();
}
}
public static void main(String[] args) {
Test test=new Test();
Thread[] th=new Thread[20];
for (int i = 0; i < 20; i++) {
new Thread(()->{
test.sayHi();
}).start();
}
}
}
運(yùn)行上面的測試代碼，結(jié)果如下，可以看見通過加鎖保證了對變量i的同步訪問控制：
接下來通過下面的例子測試鎖的可重入性：
public class Test2 {
private MyLock lock=new MyLock();
public void function1(){
lock.lock();
System.out.println("execute function1");
function2();
lock.unlock();
}
public void function2(){
lock.lock();
System.out.println("execute function2");
lock.unlock();
}
public static void main(String[] args) {
Test2 test2=new Test2();
new Thread(()->{
test2.function1();
}).start();
}
}
執(zhí)行上面的代碼，可以看到在function1未釋放鎖的情況下，function2對鎖進(jìn)行了重入并執(zhí)行了后續(xù)的代碼：
總結(jié)
通過上面的學(xué)習(xí)，我們了解了AQS的兩大核心同步隊(duì)列和同步狀態(tài)，并對AQS對資源的管理以及隊(duì)列狀態(tài)的變化有了一定的研究。其實(shí)歸根結(jié)底，AQS只是提供給我們來開發(fā)同步組件的一個底層框架，在它的層面上，并不關(guān)心子類在繼承它時要實(shí)現(xiàn)什么功能，AQS只是提供了一套維護(hù)同步狀態(tài)的功能，至于要完成什么樣的一個工具類，這完全是由我們自己去定義的。
02
好文推薦
你為什么不敢重構(gòu)代碼？
好家伙！30% 國外程序員每天“摸魚”四五個小時，國內(nèi)似乎更嚴(yán)重…
因故意引入漏洞，整所大學(xué)被禁止為Linux內(nèi)核做貢獻(xiàn)，回應(yīng)來了！
一鍵三連「分享」、「點(diǎn)贊」和「在看」
技術(shù)干貨與你天天見~