你來講講AQS是什麼吧？都是怎麼用的？
阿新 • • 發佈：2020-09-30
## 前言
在Java面試的時候，多執行緒相關的知識是躲不掉的，肯定會被問。我就被問到了AQS的知識，就直接了當的問，AQS知道是什麼吧，來講講它是怎麼實現的，以及哪些地方用到了它。當時自己確實沒有講好，所以這次來總結一下這個知識點。
## 什麼是AQS
AQS全稱是`AbstractQueuedSynchronizer`，形如其名，抽象佇列同步器。
AQS定義了兩種資源共享模式：
* **獨佔式，每次只能有一個執行緒持有鎖，例如`ReentrantLock`實現的就是獨佔式的鎖資源。**
* **共享式，允許多個執行緒同時獲取鎖，併發訪問共享資源，`ReentrantWriteLock`和`CountDownLatch`等就是實現的這種模式。**
它維護了一個`volatile`的`state`變數和一個FIFO（先進先出）的佇列。
其中`state`變數代表的是競爭資源標識，而佇列代表的是競爭資源失敗的執行緒排隊時存放的容器。
```java
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {
...
/**
* The synchronization state.
*/
private volatile int state;
/**
* Wait queue node class.
**/
static final class Node {
...
}
...
}
```
AQS中提供了操作state的方法：
- getState();
- setState();
- compareSetState();
```java
protected final int getState() {
return state;
}
protected final void setState(int newState) {
state = newState;
}
protected final boolean compareAndSetState(int expect, int update) {
// See below for intrinsics setup to support this
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
```
因為`AbstractQueuedSynchronizer`是一個抽象類，他採用**模板方法**的設計模式，規定了**獨佔**和**共享**模式需要實現的方法，並且將一些通用的功能已經進行了實現，所以不同模式的使用方式，只需要自己定義好實現共享資源的獲取與釋放即可，至於具體執行緒在等待佇列中的維護（獲取資源入佇列、喚醒出佇列等），AQS已經實現好了。
所以根據共享資源的模式一般實現的方法有如下幾個：
- `isHeldExclusively()；`// 是否為獨佔模式；但是隻有使用到了Condition的，才需要去實現它。例如：ReentrantLock。
- `boolean tryAcquire(int arg);` // 獨佔模式；嘗試獲取資源，成功返回true，失敗返回false。
- `boolean tryRelease(int arg) ;` // 獨佔模式；嘗試釋放資源，成功返回true，失敗返回false。
- `int tryAcquireShared(int arg);` // 共享模式；嘗試獲取資源，負數表示失敗；0表示成功，但是沒有剩餘可用資源了；正數表示成功，且有剩餘可用資源。
- `boolean tryReleaseShared(int arg) ;` // 共享歐式；嘗試釋放資源，若釋放資源後允許喚醒後續等待節點返回true，否則返回false。
上面的這幾個方法在`AbstractQueuedSynchronizer`這個抽象類中，都沒有被定義為abstract的，說明這些方法都是可以按需實現的，共享模式下可以只實現共享模式的方法（例如`CountDownLatch`），獨佔模式下可以只實現獨佔模式的方法（例如`ReentrantLock`）,也支援兩種都實現，兩種模式都使用（例如`ReentrantReadWriteLock`）。
## AQS原始碼分析
我們先簡單介紹AQS的兩種模式的實現類的代表`ReentrantLock`（**獨佔模式**）和`CountDownLatch`（**共享模式**），是如何來共享資源的一個過程，然後再詳細通過AQS的原始碼來分析整個實現過程。
- `ReentrantLock`在初始化的時候state=0，表示資源未被鎖定。當A執行緒執行`lock()`方法時，會呼叫`tryAcquire()`方法，將AQS中佇列的模式設定為獨佔，並將獨佔執行緒設定為執行緒A，以及將`state+1`。
這樣線上程A沒有釋放鎖前，其他執行緒來競爭鎖，呼叫`tryAcquire()`方法時都會失敗，然後競爭鎖失敗的執行緒就會進入到佇列中。當執行緒A呼叫執行`unlock()`方法將`state=0`後，其他執行緒才有機會獲取鎖（注意`ReentrantLock`是可重入的，同一執行緒多次獲取鎖時`state`值會進行壘加的，在釋放鎖時也要釋放相應的次數才算完全釋放了鎖）。
- `CountDownLatch`會將任務分成N個子執行緒去執行，state的初始值也是N（`state`與子執行緒數量一致）。N個子執行緒是並行執行的，每個子執行緒執行完成後`countDown()`一次，`state`會通過CAS方式減1。直到所有子執行緒執行完成後（`state=0`）,會通過`unpark()`方法喚醒主執行緒，然後主執行緒就會從`await()`方法返回，繼續後續操作。
### 獨佔模式分析
在AbstractQueuedSynchronizer的類裡面有一個靜態內部類Node。它代表的是佇列中的每一個節點。
其中Node節點有如下幾個屬性：
```java
// 節點的狀態
volatile int waitStatus;
// 當前節點的前一個節點
volatile Node prev;
// 當前節點的後一個節點
volatile Node next;
// 當前節點中所包含的執行緒物件
volatile Thread thread;
// 等待佇列中的下一個節點
Node nextWaiter;
```
每個屬性代表的什麼，已經寫在程式碼註釋中了。其中Node類中還有幾個常量，代表了幾個節點的狀態（`waitStatus`）值。
```java
/** waitStatus value to indicate thread has cancelled */
static final int CANCELLED = 1;
/** waitStatus value to indicate successor's thread needs unparking */
static final int SIGNAL = -1;
/** waitStatus value to indicate thread is waiting on condition */
static final int CONDITION = -2;
/**
* waitStatus value to indicate the next acquireShared should
* unconditionally propagate
*/
static final int PROPAGATE = -3;
```
首先節點的狀態值waitStatus預設是0，然後下面幾個常量有自己具體的含義。
`CANCELLED = 1；` 代表的是當前節點從同步佇列中取消，當timeout或被中斷（響應中斷的情況下），會觸發變更為此狀態，進入該狀態後的結點將不會再變化。
`SIGNAL = -1；` 代表後繼節點處於等待狀態。後繼結點入隊時，會將前繼結點的狀態更新為SIGNAL。
`CONDITION = -2;` 節點在等待佇列中，節點執行緒等待在Condition上，當其他執行緒對Condition呼叫了 `signal()`方法後，該節點將會從等待佇列中轉移到同步佇列中，加入到對同步狀態的獲取中。
`PROPAGATE = -3;` 表示在共享模式下，前繼節點在釋放資源後會喚醒後繼節點，並將這種共享模式傳播下去。
通過上面幾個固定的常量值，我們可以看出節點狀態中通常負數值通常表示節點處於有效的等待狀態，而正數值代表節點已經被取消了。
所以AQS原始碼中有很多地方都用`waitStatus>0`或`waitStatus<0`這種方式來判斷佇列中節點的是否正常。
**獨佔模式下，只能有一個執行緒佔有鎖資源，其他競爭資源的執行緒，在競爭失敗後都會進入到等待佇列中，等待佔有鎖資源的執行緒釋放鎖，然後再重新被喚醒競爭資源。**
### ReentrantLock加鎖過程
`ReentrantLock`預設是非公平鎖，就是說，執行緒在競爭鎖的時候並不是按照先來後到的順序來獲取鎖的，但是`ReentrantLock`也是支援公平鎖的，在建立的時候傳入一個引數值即可。
下面我們以`ReentrantLock`預設情況下的加鎖來分析AQS的原始碼。
`ReentrantLock`並沒有直接繼承AQS類，而是通過內部類來繼承AQS類的，這樣自己的實現功能，自己用。
我們在用`ReentrantLock`加鎖的時候都是呼叫的用`lock()`方法，那麼我們來看看預設非公平鎖下，`lock()`方法的原始碼：
```java
/**
* Sync object for non-fair locks
*/
static final class NonfairSync extends Sync {
private static final long serialVersionUID = 7316153563782823691L;
/**
* Performs lock. Try immediate barge, backing up to normal
* acquire on failure.
*/
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
```
通過原始碼可以看到，`lock()`方法，首先是通過CAS的方式搶佔鎖，如果搶佔成功則將state的值設定為1。然後將物件獨佔執行緒設定為當前執行緒。
```java
protected final void setExclusiveOwnerThread(Thread thread) {
exclusiveOwnerThread = thread;
}
```
如果搶佔鎖失敗，就會呼叫`acquire()`方法，這個`acquire()`方法的實現就是在AQS類中了，說明具體搶佔鎖失敗後的邏輯，AQS已經規定好了模板。
```java
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
```
上面已經介紹了，獨佔模式是需要實現`tryAcquire()`方法的，這裡首先就是通過`tryAcquire()`方法搶佔鎖，如果成功返回true，失敗返回false。`tryAcquire()`方法的具體實現，是在`ReentrantLock`裡面的，AQS類中預設是直接丟擲異常的。
* 首先獲取state值，如果state值為0，說明無鎖，那麼通過CAS嘗試加鎖，成功後，將獨佔執行緒設定為當前執行緒。
* 如果state值不為0，並且當前的獨佔執行緒和當前執行緒為同一執行緒，那麼state重入次數加1。
* 如果state值不為0，並且當前執行緒不是獨佔執行緒，直接返回false。
```java
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();// 獲取state值
if (c == 0) {
// 如果state值為0，說明無鎖，那麼就通過cas方式，嘗試加鎖，成功後將獨佔執行緒設定為當前執行緒
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) { // 如果是同一個執行緒再次來獲取鎖，那麼就將state的值進行加1處理（可重入鎖的，重入次數）。
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
```
我們繼續來看`acquire()`方法，在執行完`tryAcquire()`方法後，如果加鎖失敗那麼就會執行`addWaiter()`方法和`acquireQueued()`，這兩個方法的作用是將競爭鎖失敗的執行緒放入到等待佇列中。
`addWaiter()`方法的原始碼如下：
```java
private Node addWaiter(Node mode) {
// 用引數指定的模式將當前執行緒封裝成佇列中的節點（EXCLUSIVE【獨佔】，SHARED【共享】）
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
// tail是佇列的尾部節點，初始時佇列為空，尾部節點為null，直接呼叫enq將節點插入佇列
if (pred != null) {
// 將當前執行緒節點的前級節點指向佇列的尾部節點。
node.prev = pred;
// 通過CAS方式將節點插入到佇列中
if (compareAndSetTail(pred, node)) {
// 插入成功後，將原先的尾部節點的後級節點指向新的尾部節點
pred.next = node;
return node;
}
}
// 如果尾部節點為空或通過CAS插入佇列失敗則通過enq方法插入節點
enq(node);
return node;
}
```
`addWaiter()`中主要做了三件事：
* 將當前執行緒封裝成Node。
* 判斷佇列中尾部節點是否為空，若不為空，則將當前執行緒的Node節點通過CAS插入到尾部。
* 如果尾部節點為空或CAS插入失敗則通過`enq()`方法插入到佇列中。
#### 那麼enq()方法是又是怎麼插入節點的呢？
enq()方法原始碼如下：
```java
private Node enq(final Node node) {
// 看到死迴圈，就明白是通過自旋咯
for (;;) {
// 當tail節點為空時直接將當前節點設定成尾部節點，並插入到佇列中，以及設定它為head節點。
Node t = tail;
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
// 若是因為在addWaiter()方法中插入失敗或第二次進入迴圈，那麼將當前執行緒的前級節點指向尾部節點，並通過CAS方式將尾部節點指向當前執行緒的節點。
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
```
其實enq()方法主要就是通過自旋將資料插入到佇列中的操作：
* 當佇列為空時，將當前節點設定為頭節點和尾節點。
* 進入二次迴圈後，將node新增到尾部。
這樣addWaiter()方法就構造了一個佇列，並將當前執行緒新增到了佇列中了。
我們再回到`acquire()`方法中。
現在就剩下`acquireQueued()`方法沒看了，這個方法中的操作挺多的。
```java
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
// 獲取前級節點，如果未null，則丟擲異常
final Node p = node.predecessor();
// 如果前級節點為head，並且執行搶佔鎖成功。
if (p == head && tryAcquire(arg)) {
// 搶佔鎖成功，當前節點成功新的head節點
setHead(node);
// 然後將原先的head節點指向null，方便垃圾回收進行回收
p.next = null; // help GC
failed = false;
return interrupted;
}
// 如果當前節點不為head,或者搶佔鎖失敗。就根據節點的狀態決定是否需要掛起執行緒。
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed) // 如果獲取鎖異常，則出取消獲取鎖操作。
cancelAcquire(node);
}
}
```
```java
final Node predecessor() throws NullPointerException {
Node p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}
```
* 首先獲取節點的前級節點。
* 如果當前節點的前級節點是head那麼就可以去搶佔鎖了。
* 搶佔成功後就將新節點設定為head，原先的head置為空。
* 如果搶佔鎖失敗，則根據`waitStatus`值決定是否掛起執行緒。
* 最後，通過`cancelAcquire()`取消獲取鎖操作。
下面看一下`shouldParkAfterFailedAcquire()`和`parkAndCheckInterrupt()`這兩個方法是如何掛起執行緒的。
```java
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;// 獲取前級節點
if (ws == Node.SIGNAL)// 如果前級節點的waitStatus值為SIGNAL(-1),說明當前節點也已經在等待喚醒了，直接返回true。
return true;
// 如果前級節點的waitStatus值大於0說明前級節點已經取消了。
if (ws > 0) {
// 如果前級節點已經是CANCEL狀態了，那麼會繼續向前找，直到找到的節點不是CANCEL（waitStatue>0）狀態的節點，然後將其設定為當前節點的前級節點。
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
// 如果前級節點為0或者其他不為-1的小於0的值，則將當前節點的前級節點設定為 SIGNAL(-1)
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
```
`parkAndCheckInterrupt()`方法的作用就是掛起執行緒，如果`shouldParkAfterFailedAcquire()`方法成功，會執行`parkAndCheckInterrupt()`方法，它通過`LockSupport的park()`方法，將當前執行緒掛起（WAITING），它需要`unpark()`方法喚醒它，通過這樣一種FIFO機制的等待，來實現Lock操作。
```java
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
```
`LockSupport`是JDK從1.6開始提供的一個執行緒同步源語工具類，在這裡主要用到了它的兩個方法，掛起執行緒和喚醒執行緒：
```java
public static void park() {
UNSAFE.park(false, 0L);
}
public static void unpark(Thread thread) {
if (thread != null)
UNSAFE.unpark(thread);
}
```
`LockSupport`的掛起和喚醒執行緒都是不可重入的，它由一個許可標誌，當呼叫park()時就會將許可設定為0，掛起執行緒，如果再呼叫一次`park()`，會阻塞執行緒。當呼叫`unpark()`時才會將許可標誌設定成1。
### ReentrantLock釋放鎖過程
`ReentrantLock`釋放鎖的過程主要有兩個階段：
- 釋放鎖。
- 喚醒掛起的執行緒。
`unlock()`方法的原始碼如下。
```java
public void unlock() {
sync.release(1);
}
```
釋放鎖的方法是寫在父類，`AbstractQueuedSynchronizer`類中的。
原始碼如下：
```java
// 釋放獨佔模式下的鎖資源
public final boolean release(int arg) {
if (tryRelease(arg)) { // 嘗試釋放資源
Node h = head;
//釋放成功後，判斷頭節點的狀態是否為無鎖狀態，如果不為無鎖狀態就將頭節點中的執行緒喚醒。
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false; // 釋放資源失敗，直接返回false
}
```
我們首先釋放資源來看`tryRelease()`方法的原始碼，看看釋放資源是怎樣的過程。
```java
protected final boolean tryRelease(int releases) {
// 從state中減去傳入引數的相應值（一般為1）
int c = getState() - releases;
// 當釋放資源的執行緒與獨佔鎖現有執行緒不一致時，非法執行緒釋放，直接丟擲異常。
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
// 這裡是處理重入鎖的機制，因為可重入機制，所以每次都重入state值都加1，
//所以在釋放的時候也要相應的減1，直到state的值為0才算完全的釋放鎖資源。
if (c == 0) {
free = true;
// 完全釋放資源後，將獨佔執行緒設定為null,這樣後面的競爭執行緒才有可能搶佔。
setExclusiveOwnerThread(null);
}
// 重新賦值state
setState(c);
return free;
}
```
`tryRelease()`方法在釋放鎖資源時，可以單純的理解為是修改獨佔模式的狀態值和置空佔有執行緒的操作。將`state`的值減掉相應的引數值（一般是1），如果計算結果為0，就將他的獨佔執行緒設定為`null`，其他執行緒才有機會搶佔成功。
在加鎖時，同一執行緒加一次鎖，`state`狀態值就會加1，在解鎖的時候沒解鎖一次就會減1，同一個鎖可重入，只有`lock`次數與`unlock`次數相同才會釋放資源，將獨佔執行緒設定為null。
釋放了資源後，我們再看喚醒掛起執行緒時的過程。這個過程就在`unparkSuccessor()`方法中。
```java
private void unparkSuccessor(Node node) {
/* 獲取當前節點的等待狀態，一般是頭節點，佔有鎖的節點是在頭節點上。 */
int ws = node.waitStatus;
// 將當前節點的執行緒的狀態值設為0，成為無鎖狀態。
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
/*
* Thread to unpark is held in successor, which is normally
* just the next node. But if cancelled or apparently null,
* traverse backwards from tail to find the actual
* non-cancelled successor.
*/
Node s = node.next;// 獲取下一個需要喚醒的節點執行緒。
if (s == null || s.waitStatus >