深入學習理解(9):java:AbstractQueuedSynchronizer詳解
導讀： 前一陣子在寫輕量級RPC框架的時候，由於系統中所需要用非同步RPC模型，由於系統所要求效能比較苛刻，所以基本所有耗時的操作都會採用非同步呼叫的方式：比如非同步讀寫DB,IO,更可能redis的操作都需要非同步（主程說了，我咋辦，做唄）。
正文
什麼是AbstractQueuedSynchronizer？
首先我們來簡單的認識一下什麼叫做AbstractQueuedSynchronizer。 AbstractQueuedSynchronizer簡稱叫做AQS. 是jdk提供的一種FIFO佇列。可以用於構建鎖或者其他相關同步裝置的基礎框架。內部採用了一個int變數來表示狀態，通過狀態的變化實現大部分同步需求的基礎。
AQS使用的方法是繼承，子類通過繼承同步器並需要實現它的方法來管理其狀態，管理的方式就是通過類似acquire和release的方式來操縱狀態。然而多執行緒環境中對狀態的操縱必須確保原子性，因此子類對於狀態的把握，需要使用這個同步器提供的以下三個方法對狀態進行操作：
java.util.concurrent.locks.AbstractQueuedSynchronizer.getState() java.util.concurrent.locks.AbstractQueuedSynchronizer.setState(int) java.util.concurrent.locks.AbstractQueuedSynchronizer.compareAndSetState(int, int)
先睹為快：如何使用AQS?
通過上面的簡單分析，我們知道，通過繼承的方式來使用AQS。下面我寫了個類來實現這個。這個正好是實現rpc非同步呼叫監聽的核心基礎。原始碼如下：
static class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = 1L; private final int done = 1; private final int pending = 0; @Override protected boolean tryAcquire(int acquires) { return getState() == done; } /** * CAS操作，保證原子性 * * @param releases * @return */ @Override protected boolean tryRelease(int releases) { if (getState() == pending) { if (compareAndSetState(pending, done)) { return true; } } return false; } public boolean isDone() { getState(); return getState() == done; } }
方法protected boolean tryAcquire(int arg) 表示
理論說明：排它的獲取這個狀態。這個方法的實現需要查詢當前狀態是否允許獲取，然後再進行獲取（使用compareAndSetState來做）狀態。
此處使用意義：當狀態等於制定狀態是代表rpc遠端呼叫已經得到返回。說明此任務已經完成。
方法：protected boolean tryRelease(int arg) 表示： 理論說明:釋放狀態。
此處使用意義：原子修改狀態值。當任務獲得鎖任務執行完成的時候，需要釋放這個鎖。
深入原理解析
如何在實戰專案中使用AQS,我們上面已經簡單的描述了，下面我們通過分析原始碼部分來解析AQS的工作原理。
jdk關於AQS的描述是長篇大論。其實我們只要明白這就可以了。原文註釋如下：
/**
* Provides a framework for implementing blocking locks and related
* synchronizers (semaphores, events, etc) that rely on
* first-in-first-out (FIFO) wait queues. This class is designed to
* be a useful basis for most kinds of synchronizers that rely on a
* single atomic {@code int} value to represent state. Subclasses
* must define the protected methods that change this state, and which
* define what that state means in terms of this object being acquired
* or released. Given these, the other methods in this class carry
* out all queuing and blocking mechanics. Subclasses can maintain
* other state fields, but only the atomically updated {@code int}
* value manipulated using methods {@link #getState}, {@link
* #setState} and {@link #compareAndSetState} is tracked with respect
* to synchronization.
大概的意思就是
為實現阻塞鎖和相關的鎖提供了一個框架依賴於同步器（訊號量、事件等）先入先出（FIFO）等待佇列。這類的目的是對於大多數依賴於它的同步器來說，這是一個有用的基礎。單個原子int值來表示狀
態。子類必須定義寫這個狀態的受保護的方法，以及定義這個狀態在被獲取的物件上的含義或釋放。考慮到這些，這個類中的其他方法排除所有的排隊和阻塞機制。子類可以維護其他狀態欄位，但僅是原子性更新的@code int使用方法 getstate setstate和compareandsetstate被跟蹤同步。
然我們明白AQS其實提供的是一種鎖的功能。在jdk中，AQS的功能可以分為兩類：獨佔功能和共享功能，它的所有子類中，要麼實現並使用了它獨佔功能的API，要麼使用了共享鎖的功能，而不會同時使用兩套API，即便是它最有名的子類ReentrantReadWriteLock，也是通過兩個內部類：讀鎖和寫鎖，分別實現的兩套API來實現的，為什麼這麼做，後面我們再分析，到目前為止，我們只需要明白AQS在功能上有獨佔控制和共享控制兩種功能即可。
內部Node節點
Node 節點是代表獲取lock的執行緒, 存在於 Condition Queue, Sync Queue 裡面， 而其主要就是 nextWaiter (標記共享還是獨佔),waitStatus 標記node的狀態。
waitStatus的狀態變化:
執行緒剛入 Sync Queue 裡面, 發現獨佔鎖被其他人獲取, 則將其前繼節點標記為 SIGNAL, 然後再嘗試獲取一下鎖(呼叫 tryAcquire 方法) 若呼叫 tryAcquire 方法獲取失敗, 則判斷一下是否前繼節點被標記為 SIGNAL, 若是的話 直接 block(block前會確保前繼節點被標記為SIGNAL, 因為前繼節點在進行釋放鎖時根據是否標記為 SIGNAL 來決定喚醒後繼節點與否 <- 這是獨佔的情況下) 前繼節點使用完lock, 進行釋放, 因為自己被標記為 SIGNAL, 所以喚醒其後繼節點 waitStatus 變化過程:
獨佔模式下: 0(初始) -> signal(被後繼節點標記為release需要喚醒後繼節點) -> 0 (等釋放好lock, 會恢復到0) 獨佔模式 + 使用 Condition情況下: 0(初始) -> signal(被後繼節點標記為release需要喚醒後繼節點) -> 0 (等釋放好lock, 會恢復到0)其上可能涉及 中斷與超時, 只是多了一個 CANCELLED, 當節點變成 CANCELLED, 後就等著被清除。
共享模式下: 0(初始) -> PROPAGATE(獲取 lock 或release lock 時) (獲取 lock 時會呼叫 setHeadAndPropagate 來進行 傳遞式的喚醒後繼節點, 直到碰到 獨佔模式的節點) 共享模式 + 獨佔模式下: 0(初始) -> signal(被後繼節點標記為release需要喚醒後繼節點) -> 0 (等釋放好lock, 會恢復到0)
獨佔鎖
在jdk中提供獨佔鎖的功能，例如獨佔控制功能的子類ReentrantLock，對於ReentrantLock，使用過的同學應該都知道，通常是這麼用它的：
reentrantLock.lock()
//do something
finally{
reentrantLock.unlock()
}
ReentrantLock會保證 do something在同一時間只有一個執行緒在執行這段程式碼，或者說，同一時刻只有一個執行緒的lock方法會返回。其餘執行緒會被掛起，直到獲取鎖。從這裡可以看出，其實ReentrantLock實現的就是一個獨佔鎖的功能：有且只有一個執行緒獲取到鎖，其餘執行緒全部掛起，直到該擁有鎖的執行緒釋放鎖，被掛起的執行緒被喚醒重新開始競爭鎖。沒錯，ReentrantLock使用的就是AQS的獨佔API實現的。
獨佔方式獲取lock主要流程
呼叫 tryAcquire 嘗試性的獲取鎖(一般都是由子類實現), 成功的話直接返回
tryAcquire 呼叫獲取失敗, 將當前的執行緒封裝成 Node 加入到 Sync Queue 裡面(呼叫addWaiter), 等待獲取 signal 訊號
呼叫 acquireQueued 進行自旋的方式獲取鎖(有可能會 repeatedly blocking and unblocking)
根據acquireQueued的返回值判斷在獲取lock的過程中是否被中斷, 若被中斷, 則自己再中斷一下(selfInterrupt), 若是響應中斷的則直接丟擲異常
獨佔方式獲取lock主要分成3類 acquire 不響應中斷的獲取lock, 這裡的不響應中斷指的是執行緒被中斷後會被喚醒, 並且繼續獲取lock,在方法返回時, 根據剛才的獲取過程是否被中斷來決定是否要自己中斷一下(方法 selfInterrupt)
doAcquireInterruptibly 響應中斷的獲取 lock, 這裡的響應中斷, 指線上程獲取 lock 過程中若被中斷, 則直接丟擲異常
doAcquireNanos 響應中斷及超時的獲取 lock, 當執行緒被中斷, 或獲取超時, 則直接丟擲異常, 獲取失敗
獨佔的獲取lock 方法 acquire
public final void acquire(int arg){
if(!tryAcquire(arg)&&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) {
selfInterrupt();
}
}
呼叫 tryAcquire 嘗試性的獲取鎖(一般都是又子類實現), 成功的話直接返回
tryAcquire 呼叫獲取失敗, 將當前的執行緒封裝成 Node 加入到 Sync Queue 裡面(呼叫addWaiter), 等待獲取 signal 訊號
呼叫 acquireQueued 進行自旋的方式獲取鎖(有可能會 repeatedly blocking and unblocking)
根據acquireQueued的返回值判斷在獲取lock的過程中是否被中斷, 若被中斷, 則自己再中斷一下(selfInterrupt)。
迴圈獲取lock 方法 acquireQueued
final boolean acquireQueued(final Node node, int arg){
boolean failed = true;
try {
boolean interrupted = false;
for(;;){
final Node p = node.predecessor(); // 1. 獲取當前節點的前繼節點 (當一個n在 Sync Queue 裡面, 並且沒有獲取 lock 的 node 的前繼節點不可能是 null)
if(p == head && tryAcquire(arg)){ // 2. 判斷前繼節點是否是head節點(前繼節點是head, 存在兩種情況 (1) 前繼節點現在佔用 lock (2)前繼節點是個空節點, 已經釋放 lock, node 現在有機會獲取 lock); 則再次呼叫 tryAcquire嘗試獲取一下
setHead(node); // 3. 獲取 lock 成功, 直接設定 新head(原來的head可能就直接被回收)
p.next = null; // help GC // help gc
failed = false;
return interrupted; // 4. 返回在整個獲取的過程中是否被中斷過 ; 但這又有什麼用呢? 若整個過程中被中斷過, 則最後我在 自我中斷一下 (selfInterrupt), 因為外面的函式可能需要知道整個過程是否被中斷過
}
if(shouldParkAfterFailedAcquire(p, node) && // 5. 呼叫 shouldParkAfterFailedAcquire 判斷是否需要中斷(這裡可能會一開始 返回 false, 但在此進去後直接返回 true(主要和前繼節點的狀態是否是 signal))
parkAndCheckInterrupt()){ // 6. 現在lock還是被其他執行緒佔用 那就睡一會, 返回值判斷是否這次執行緒的喚醒是被中斷喚醒
interrupted = true;
}
}
}finally {
if(failed){ // 7. 在整個獲取中出錯
cancelAcquire(node); // 8. 清除 node 節點(清除的過程是先給 node 打上 CANCELLED標誌, 然後再刪除)
}
}
當前節點的前繼節點是head節點時，先 tryAcquire獲取一下鎖, 成功的話設定新 head, 返回 第一步不成功, 檢測是否需要sleep, 需要的話就sleep, 等待前繼節點在釋放lock時喚醒或通過中斷來喚醒 整個過程可能需要blocking nonblocking 幾次
中斷獲取lock 方法 doAcquireInterruptibly
private void doAcquireInterruptibly(int arg) throws InterruptedException{
final Node node = addWaiter(Node.EXCLUSIVE); // 1. 將當前的執行緒封裝成 Node 加入到 Sync Queue 裡面
boolean failed = true;
try {
for(;;){
final Node p = node.predecessor(); // 2. 獲取當前節點的前繼節點 (當一個n在 Sync Queue 裡面, 並且沒有獲取 lock 的 node 的前繼節點不可能是 null)
if(p == head && tryAcquire(arg)){ // 3. 判斷前繼節點是否是head節點(前繼節點是head, 存在兩種情況 (1) 前繼節點現在佔用 lock (2)前繼節點是個空節點, 已經釋放 lock, node 現在有機會獲取 lock); 則再次呼叫 tryAcquire嘗試獲取一下
setHead(node);
p.next = null; // help GC
failed = false;
return;
}
if(shouldParkAfterFailedAcquire(p, node) && // 4. 呼叫 shouldParkAfterFailedAcquire 判斷是否需要中斷(這裡可能會一開始 返回 false, 但在此進去後直接返回 true(主要和前繼節點的狀態是否是 signal))
parkAndCheckInterrupt()){ // 5. 現在lock還是被其他執行緒佔用 那就睡一會, 返回值判斷是否這次執行緒的喚醒是被中斷喚醒
throw new InterruptedException(); // 6. 執行緒此時喚醒是通過執行緒中斷, 則直接拋異常
}
}
}finally {
if(failed){ // 7. 在整個獲取中出錯(比如執行緒中斷)
cancelAcquire(node); // 8. 清除 node 節點(清除的過程是先給 node 打上 CANCELLED標誌, 然後再刪除)
}
}
}
public final void acquireInterruptibly(int arg) throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (!tryAcquire(arg))
doAcquireInterruptibly(arg);
}
超時&中斷獲取lock 方法
tryAcquireNanos(int arg, long nanosTimeout):獨佔且支援超時模式獲取： 帶有超時時間，如果經過超時時間則會退出。
釋放lock方法
- 呼叫子類的 tryRelease 方法釋放獲取的資源 判斷是否完全釋放lock(這裡有 lock 重複獲取的情況)
- 判斷是否有後繼節點需要喚醒, 需要的話呼叫unparkSuccessor進行喚醒
共享鎖
共享方式獲取lock流程 呼叫 tryAcquireShared 嘗試性的獲取鎖(一般都是由子類實現), 成功的話直接返回 tryAcquireShared 呼叫獲取失敗, 將當前的執行緒封裝成 Node 加入到 Sync Queue 裡面(呼叫addWaiter), 等待獲取 signal 訊號
在 Sync Queue 裡面進行自旋的方式獲取鎖(有可能會 repeatedly blocking and unblocking 當獲取失敗, 則判斷是否可以 block(block的前提是前繼節點被打上 SIGNAL 標示) 共享與獨佔獲取lock的區別主要在於 在共享方式下獲取 lock 成功會判斷是否需要繼續喚醒下面的繼續獲取共享lock的節點(及方法 doReleaseShared)
共享方式獲取lock主要分成3類 1:acquireShared 不響應中斷的獲取lock, 這裡的不響應中斷指的是執行緒被中斷後會被喚醒, 並且繼續獲取lock,在方法返回時, 根據剛才的獲取過程是否被中斷來決定是否要自己中斷一下(方法 selfInterrupt)
2:doAcquireSharedInterruptibly 響應中斷的獲取 lock, 這裡的響應中斷, 指線上程獲取 lock 過程中若被中斷, 則直接丟擲異常
3:doAcquireSharedNanos 響應中斷及超時的獲取 lock, 當執行緒被中斷, 或獲取超時, 則直接丟擲異常, 獲取失敗
獲取共享lock 方法 acquireShared
public final void acquireShared(int arg){
if(tryAcquireShared(arg) < 0){ // 1. 呼叫子類, 獲取共享 lock 返回 < 0, 表示失敗
doAcquireShared(arg); // 2. 呼叫 doAcquireShared 當前 執行緒加入 Sync Queue 裡面, 等待獲取 lock
}
}
獲取共享lock 方法 doAcquireShared
private void doAcquireShared(int arg){
final Node node = addWaiter(Node.SHARED); // 1. 將當前的執行緒封裝成 Node 加入到 Sync Queue 裡面
boolean failed = true;
try {
boolean interrupted = false;
for(;;){
final Node p = node.predecessor(); // 2. 獲取當前節點的前繼節點 (當一個n在 Sync Queue 裡面, 並且沒有獲取 lock 的 node 的前繼節點不可能是 null)
if(p == head){
int r = tryAcquireShared(arg); // 3. 判斷前繼節點是否是head節點(前繼節點是head, 存在兩種情況 (1) 前繼節點現在佔用 lock (2)前繼節點是個空節點, 已經釋放 lock, node 現在有機會獲取 lock); 則再次呼叫 tryAcquireShared 嘗試獲取一下
if(r >= 0){
setHeadAndPropagate(node, r); // 4. 獲取 lock 成功, 設定新的 head, 並喚醒後繼獲取 readLock 的節點
p.next = null; // help GC
if(interrupted){ // 5. 在獲取 lock 時, 被中斷過, 則自己再自我中斷一下(外面的函式可能需要這個引數)
selfInterrupt();
}
failed = false;
return;
}
}
if(shouldParkAfterFailedAcquire(p, node) && // 6. 呼叫 shouldParkAfterFailedAcquire 判斷是否需要中斷(這裡可能會一開始 返回 false, 但在此進去後直接返回 true(主要和前繼節點的狀態是否是 signal))
parkAndCheckInterrupt()){ // 7. 現在lock還是被其他執行緒佔用 那就睡一會, 返回值判斷是否這次執行緒的喚醒是被中斷喚醒
interrupted = true;
}
}
}finally {
if(failed){ // 8. 在整個獲取中出錯(比如執行緒中斷/超時)
cancelAcquire(node); // 9. 清除 node 節點(清除的過程是先給 node 打上 CANCELLED標誌, 然後再刪除)
}
}
}
獲取共享lock 方法 doAcquireSharedInterruptibly
private void doAcquireSharedInterruptibly(int arg) throws InterruptedException{
final Node node = addWaiter(Node.SHARED); // 1. 將當前的執行緒封裝成 Node 加入到 Sync Queue 裡面
boolean failed = true;
try {
for(;;){
final Node p = node.predecessor(); // 2. 獲取當前節點的前繼節點 (當一個n在 Sync Queue 裡面, 並且沒有獲取 lock 的 node 的前繼節點不可能是 null)
if(p == head){
int r = tryAcquireShared(arg); // 3. 判斷前繼節點是否是head節點(前繼節點是head, 存在兩種情況 (1) 前繼節點現在佔用 lock (2)前繼節點是個空節點, 已經釋放 lock, node 現在有機會獲取 lock); 則再次呼叫 tryAcquireShared 嘗試獲取一下
if(r >= 0){
setHeadAndPropagate(node, r); // 4. 獲取 lock 成功, 設定新的 head, 並喚醒後繼獲取 readLock 的節點
p.next = null; // help GC
failed = false;
return;
}
}
if(shouldParkAfterFailedAcquire(p, node) && // 5. 呼叫 shouldParkAfterFailedAcquire 判斷是否需要中斷(這裡可能會一開始 返回 false, 但在此進去後直接返回 true(主要和前繼節點的狀態是否是 signal))
parkAndCheckInterrupt()){ // 6. 現在lock還是被其他執行緒佔用 那就睡一會, 返回值判斷是否這次執行緒的喚醒是被中斷喚醒
throw new InterruptedException(); // 7. 若此次喚醒是 通過執行緒中斷, 則直接丟擲異常
}
}
}finally {
if(failed){ // 8. 在整個獲取中出錯(比如執行緒中斷/超時)
cancelAcquire(node); // 9. 清除 node 節點(清除的過程是先給 node 打上 CANCELLED標誌, 然後再刪除)
}
}
}
獲取共享lock 方法 doAcquireSharedNanos
private boolean doAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException{
if (nanosTimeout <= 0L){
return false;
}
final long deadline = System.nanoTime() + nanosTimeout; // 0. 計算超時的時間
final Node node = addWaiter(Node.SHARED); // 1. 將當前的執行緒封裝成 Node 加入到 Sync Queue 裡面
boolean failed = true;
try {
for(;;){
final Node p = node.predecessor(); // 2. 獲取當前節點的前繼節點 (當一個n在 Sync Queue 裡面, 並且沒有獲取 lock 的 node 的前繼節點不可能是 null)
if(p == head){
int r = tryAcquireShared(arg); // 3. 判斷前繼節點是否是head節點(前繼節點是head, 存在兩種情況 (1) 前繼節點現在佔用 lock (2)前繼節點是個空節點, 已經釋放 lock, node 現在有機會獲取 lock); 則再次呼叫 tryAcquireShared 嘗試獲取一下
if(r >= 0){
setHeadAndPropagate(node, r); // 4. 獲取 lock 成功, 設定新的 head, 並喚醒後繼獲取 readLock 的節點
p.next = null; // help GC
failed = false;
return true;
}
}
nanosTimeout = deadline - System.nanoTime(); // 5. 計算還剩餘的 timeout , 若小於0 則直接return
if(nanosTimeout <= 0L){
return false;
}
if(shouldParkAfterFailedAcquire(p, node) && // 6. 呼叫 shouldParkAfterFailedAcquire 判斷是否需要中斷(這裡可能會一開始 返回 false, 但在此進去後直接返回 true(主要和前繼節點的狀態是否是 signal))
nanosTimeout > spinForTimeoutThreshold){// 7. 在timeout 小於 spinForTimeoutThreshold 時 spin 的效率, 比 LockSupport 更高
LockSupport.parkNanos(this, nanosTimeout);
}
if(Thread.interrupted()){ // 7. 若此次喚醒是 通過執行緒中斷, 則直接丟擲異常
throw new InterruptedException();
}
}
}finally {
if (failed){ // 8. 在整個獲取中出錯(比如執行緒中斷/超時)
cancelAcquire(node); // 10. 清除 node 節點(清除的過程是先給 node 打上 CANCELLED標誌, 然後再刪除)
}
}
}
釋放共享lock
當 Sync Queue中存在連續多個獲取 共享lock的節點時, 會出現併發的喚醒後繼節點(因為共享模式下獲取lock後會喚醒近鄰的後繼節點來獲取lock)。首先呼叫子類的 tryReleaseShared來進行釋放 lock，然後判斷是否需要喚醒後繼節點來獲取 lock
private void doReleaseShared(){
for(;;){
Node h = head; // 1. 獲取 head 節點, 準備 release
if(h != null && h != tail){ // 2. Sync Queue 裡面不為 空
int ws = h.waitStatus;
if(ws == Node.SIGNAL){ // 3. h節點後面可能是 獨佔的節點, 也可能是 共享的, 並且請求了喚醒(就是給前繼節點打標記 SIGNAL)
if(!compareAndSetWaitStatus(h, Node.SIGNAL, 0)){ // 4. h 恢復 waitStatus 值置0 (為啥這裡要用 CAS 呢, 因為這裡的呼叫可能是在 節點剛剛獲取 lock, 而其他執行緒又對其進行中斷, 所用cas就出現失敗)
continue; // loop to recheck cases
}
unparkSuccessor(h); // 5. 喚醒後繼節點
}
else if(ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE)){ //6. h後面沒有節點需要喚醒, 則標識為 PROPAGATE 表示需要繼續傳遞喚醒(主要是區別 獨佔節點最終狀態0 (獨佔的節點在沒有後繼節點, 並且release lock 時最終 waitStatus 儲存為 0))
continue; // loop on failed CAS // 7. 同樣這裡可能存在競爭
}
}
if(h == head){ // 8. head 節點沒變化, 直接 return(從這裡也看出, 一個共享模式的 節點在其喚醒後繼節點時, 只喚醒一個, 但是它會在獲取 lock 時喚醒, 釋放 lock 時也進行, 所以或導致競爭的操作)
break; // head 變化了, 說明其他節點獲取 lock 了, 自己的任務完成, 直接退出
}
}
}
總結
本文主要講過了抽象的佇列式的同步器AQS的主要方法和實現原理。分別介紹了ASQ的實戰使用方式以及相關原理：Node、Condition Queue、 Sync Queue、獨佔獲取釋放lock、共享獲取釋放lock的具體原始碼實現。AQS定義了一套多執行緒訪問共享資源的同步器框架，許多同步類實現都依賴於它。