AQS 源碼解析：原理與實踐
在當今多線程編程日益普遍的時代，Java 并發包以其豐富的工具和強大的功能為開發者提供了堅實的支持。而在這其中，抽象隊列同步器（AbstractQueuedSynchronizer，簡稱 AQS）無疑是并發包的核心基石，它默默地支撐著眾多高級并發工具的實現，像 ReentrantLock、CountDownLatch、Semaphore 等。 這些工具在我們的日常開發中頻繁出現，極大地提升了我們處理并發場景的效率。
然而，很多開發者在使用它們時，往往只是停留在表面，對其底層的實現原理知之甚少。深入了解 AQS 的源碼，就如同拿到了一把打開 Java 并發編程底層世界的鑰匙，能夠讓我們更加透徹地理解這些并發工具的工作機制，在面對復雜的并發問題時，能夠更加游刃有余地進行分析和解決。
本文將帶領大家一同踏上探索 AQS 源碼的旅程，從 AQS 的基本概念入手，逐步深入到源碼的每一個關鍵環節，揭開它神秘的面紗，讓大家領略 Java 并發框架底層設計的精妙之處 。
一、AQS簡介
在JUC并發包下，有可重入鎖(ReentrantLock)、倒計時門閂(CountDownLatch)等并發流程工具，其底層對于線程資源爭搶以及獲取執行權都是基于AQS實現的。AQS的思想也很巧妙，將多線程抽象為一個個節點放到一個CLH雙向隊列中，爭搶到的節點可以修改一個為state的狀態值，這個state我們這里暫且可以理解為爭搶到資源的標識，只有當前獲取執行權的線程將state設置到某個值的時候才能進行臨界資源操作，此時其余線程就會感知到這一點進入CLH隊列中等待鎖釋放后的爭搶：
可以說AQS在Doug Lea大神的操刀下使用起來非常方便，他把多線程當作一個個節點，實現了線程因爭搶不到失敗而進入等待隊列，以及從等待隊列中喚醒線程等細節都實現好了。我們只需要按需實現自己嘗試獲取鎖和釋放鎖的邏輯即可。
二、基于AQS手寫一個可重入鎖
1. 落地思路
我們希望編寫一把可重入鎖，他能做到同一個線程可以操作這把鎖，例如線程1連續上鎖5次，釋放的時候也得連續釋放5次，只有完全釋放干凈之后，其他爭搶的線程才能操作這把鎖。 目前我們初定的邏輯是，在可重入鎖的實現一個內部類，這個內部集成AQS重寫嘗試取鎖和嘗試釋放鎖的邏輯。這里可能會有人有疑問，為什么我們只要實現嘗試獲取鎖和嘗試釋放鎖的邏輯呢？
因為AQS的已經為我們內置的如下的抽象邏輯，我們只需按需實現部分的規則盡可能基于AQS寫出一款強大的并發工具：
定義了一個CLH雙向鏈表隊列，存放線程節點，對應的我們可以在AQS的源碼中看到關于鏈表節點的定義：
static final class Node {
//前驅節點
volatile Node prev;
//后繼節點
volatile Node next;
//當前線程
volatile Thread thread;
//......
}內置一個嘗試獲取鎖的邏輯，這個鎖用一個int標識即state表示，內容大致為:通過CAS嘗試取鎖，成功就執行則返回成功標識。取不到就返回失敗：
protected final boolean compareAndSetState(int expect, int update) {
// See below for intrinsics setup to support this
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}實現一個釋放鎖的邏輯，嘗試釋放當前鎖，成功了喚醒后繼節點并返回true，反之返回false。
public final boolean release(int arg) {
//嘗試釋放鎖
if (tryRelease(arg)) {
Node h = head;
//成功后喚醒后繼節點
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}可以看出AQS實現時通過模板方法封裝的統一的抽象邏輯，將上鎖、釋放鎖和喚醒CLH隊列節點的邏輯都做了統一的封裝，我們只要基于這些方法編寫對應并發工具的上鎖和釋放鎖的業務邏輯即可。
所以我們實現可重入鎖的設計思路為，確保第一個將state設置為1的節點進行無限次重入并累加state的值，而其他節點則阻塞等待直到state因為鎖釋放變為0時再進行爭搶：
2. 基于AbstractQueuedSynchronizer實現邏輯抽象
按照上文描述，我們將AbstractQueuedSynchronizer 組合進來構成一個內部類，然后基于這個類實現嘗試取鎖和釋放的抽象邏輯，可以看到這段邏輯本質上都是通過AbstractQueuedSynchronizer 內置的方法編寫出來的，我們只需按照自己的邏輯按照可重入的鎖實現設置state狀態即可：
private class AQSSync extends AbstractQueuedSynchronizer {
/**
* 嘗試取鎖
* @param arg
* @return
*/
@Override
protected boolean tryAcquire(int arg) {
//獲取當前狀態值
int state = getState();
//獲取當前線程
Thread currentThread = Thread.currentThread();
//若為0說明沒有線程拿到這個資源，當前線程可以基于CAS改變狀態值，若CAS修改成功則說明這個線程拿到資源了
if (0 == state) {
if (compareAndSetState(0, arg)) {
//設置當前資源擁有者為當前線程
setExclusiveOwnerThread(currentThread);
return true;
}
} else if (getExclusiveOwnerThread() == currentThread) {//不走上述邏輯，走到這里則說明這個資源當前線程之前搶到了，這里又搶了一次，我們再疊加狀態值即可
int newState = arg + state;
if (newState < 0) {
throw new Error("Maximum lock count exceeded");
}
setState(newState);
return true;
}
return false;
}
/**
* 嘗試釋放鎖
* @param arg
* @return
*/
@Override
protected boolean tryRelease(int arg) {
//如果進行釋放的不是當前線程則拋異常
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean flag = false;
int newState = getState() - arg;
//如果state扣減后為0說明當前線程完全釋放資源了，其他線程可以開搶了
if (0 == newState) {
//設置資源擁有者為空
setExclusiveOwnerThread(null);
flag = true;
}
setState(newState);
return flag;
}
final Condition newCondition(){
return new ConditionObject();
}
}3. 基于內置類對外暴露可重入鎖
最終代碼如下如下可以看到這就模板方法的好處，我們通過lock確定行為，基于AQS作為具體實現細節。那些取鎖和釋放鎖的邏輯用AQS自帶的即可。而嘗試取鎖、釋放鎖的邏輯用我們自己的實現的即可。
/**
* 自定義可重入鎖
*/
public class ReentrantAQSLock implements Lock {
private AQSSync sync = new AQSSync();
@Override
public void lock() {
//調用基于AQS實現好的邏輯即可
sync.acquire(1);
}
@Override
public void lockInterruptibly() throws InterruptedException {
//調用基于AQS實現好的邏輯即可
sync.acquireInterruptibly(1);
}
@Override
public boolean tryLock() {
//調用我們實現的嘗試取鎖邏輯
return sync.tryAcquire(1);
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return sync.tryAcquireNanos(1, unit.toNanos(time));
}
@Override
public void unlock() {
//調用我們實現的嘗試釋放鎖邏輯
sync.release(1);
}
@Override
public Condition newCondition() {
return sync.newCondition();
}
private class AQSSync extends AbstractQueuedSynchronizer {
/**
* 嘗試取鎖
*
* @param arg
* @return
*/
@Override
protected boolean tryAcquire(int arg) {
//獲取當前狀態值
int state = getState();
//獲取當前線程
Thread currentThread = Thread.currentThread();
//若為0說明沒有線程拿到這個資源，當前線程可以基于CAS改變狀態值，若CAS修改成功則說明這個線程拿到資源了
if (0 == state) {
if (compareAndSetState(0, arg)) {
//設置當前資源擁有者為當前線程
setExclusiveOwnerThread(currentThread);
return true;
}
} else if (getExclusiveOwnerThread() == currentThread) {//不走上述邏輯，走到這里則說明這個資源當前線程之前搶到了，這里又搶了一次，我們再疊加狀態值即可
int newState = arg + state;
if (newState < 0) {
throw new Error("Maximum lock count exceeded");
}
setState(newState);
return true;
}
return false;
}
/**
* 嘗試釋放鎖
*
* @param arg
* @return
*/
@Override
protected boolean tryRelease(int arg) {
//如果進行釋放的不是當前線程則拋異常
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean flag = false;
int newState = getState() - arg;
//如果state扣減后為0說明當前線程完全釋放資源了，其他線程可以開搶了
if (0 == newState) {
//設置資源擁有者為空
setExclusiveOwnerThread(null);
flag = true;
}
setState(newState);
return flag;
}
final Condition newCondition() {
return new ConditionObject();
}
}
}4. 測試用例
我們希望使用可重入鎖，實現一個線程獲取鎖兩次后，對一個數字自增的邏輯，為了實現并發場景筆者寫了下面這段代碼
/**
* 可重入鎖測試
*/
public class ReentrantAQSLockTest {
private static Logger logger = LoggerFactory.getLogger(ReentrantAQSLockTest.class);
private int count;
private ReentrantAQSLock lock = new ReentrantAQSLock();
public void incrementCount(){
try{
logger.info("嘗試取鎖");
lock.lock();
logger.info("第1次取鎖成功");
lock.lock();
logger.info("第2次取鎖成功");
++count;
}finally {
logger.info("嘗試鎖釋放");
lock.unlock();
logger.info("第1次釋放鎖成功");
lock.unlock();
logger.info("第2次釋放鎖成功");
}
}
public static void main(String[] args) {
//500 個線程
ExecutorService threadPool = Executors.newFixedThreadPool(500);
ReentrantAQSLockTest reentrantAQSLock=new ReentrantAQSLockTest();
CountDownLatch countDownLatch=new CountDownLatch(1);
for (int i = 0; i < 500; i++) {
threadPool.submit(()->{
try {
//500 個線程全部等待
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
reentrantAQSLock.incrementCount();
});
}
//扣減倒計時門閂，500個線程同時嘗試取鎖，自增count
countDownLatch.countDown();
threadPool.shutdown();
while (!threadPool.isTerminated()){
}
logger.info("最終修改結果： "+reentrantAQSLock.count);
}
}輸出結果為500，說明邏輯沒有問題。
[main] INFO com.guide.thread.base.ReentrantAQSLockTest - 最終修改結果： 500三、詳解AQS底層數據結構
通過上面手寫了一個可重入鎖，我們大致對可重入鎖有個大致的了解，所以我們就在這里更進一步的了解一下AQS。
我們之前說過，AQS就是一個同步器，把線程當作一個個節點放在一個雙向隊列里，而這里的資源其實就是state。以上文我們手寫的可重入鎖，state為0就代表沒有線程獲取這個資源，所有節點都可以基于CAS爭搶。而不為1以及獲取到state的線程不為自己，則說明資源被其他人拿了，這些線程都會被添加到雙向隊列中等待喚醒后進行資源爭搶：
注意:這個隊列的隊首元素是AQS默認創建的一個Node節點，并沒有存放實際線程，所以在后續資源爭搶中是不參與的，這一點我們可以在上鎖失敗后第一個進入等待隊列的線程所執行的方法enq印證。 可以看到該方法在隊列全空的情況下通過CAS設置一個頭節點然后才將node設置進去：
private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) { //隊列全空情況下通過CAS創建頭節點
if (compareAndSetHead(new Node()))
tail = head;
} else {
//然后在第二輪循環將當前節點設置到等待隊列中
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}四、AQS內部隊列模式
實際上AQS有兩種隊列，一種是同步隊列，就是上文所描述的。 另一種則是同步條件隊列，原理和同步隊列差不多，只不過，取鎖是還得判斷自己是否符合條件，若符合才能爭搶資源。
1. 狀態位變量state
上文我們一直提到state，這里我們就可以展開探討。我們不妨打開AQS(AbstractQueuedSynchronizer)，可以看到這是一個volatile 變量，這就意味著對修改對每個node是保證可見性的。眾所周知volatile無法保證原子性，所以我們實現類中對state的操作都是基于CAS的。而且我們在源碼中也可以看到某些對于state的操作也是基于CAS實現。
private volatile int state;
protected final boolean compareAndSetState(int expect, int update) {
// 通過CAS修改State的值
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}2. 節點類Node
接下來就是節點類了，它用于記錄每個參與資源掙錢的線程的信息，其核心源碼如下，可以看到每個Node都會記錄自己前后節點以及自己是那條線程、以及等待狀態、是否是獨占模式等。
static final class Node {
//當前節點等待狀態
volatile int waitStatus;
//前驅節點
volatile Node prev;
//后繼節點
volatile Node next;
//當前節點記錄的線程
volatile Thread thread;
//下一個等待者
Node nextWaiter;
//設定為共享模式，意味著多線程可以使用同一個資源
static final Node SHARED = new Node();
//設置為獨占模式
static final Node EXCLUSIVE = null;上文提到了waitStatus，其實設計者也為它設置了幾種規定狀態
//表示當前線程已被取消
static final int CANCELLED = 1;
/** 表示后繼的節點等待喚醒*/
static final int SIGNAL = -1;
/**表示當前線程等待某個條件被喚醒*/
static final int CONDITION = -2;
/**
* 表示當前場景可以執行后續嘗試獲取共享鎖的邏輯
*/
static final int PROPAGATE = -3;五、獨占鎖模式源碼
1. 取鎖的邏輯
我們就基于我們手寫的可重入鎖了解一下獨占鎖實現細節，首先我們的代碼入口為lock()
public void incrementCount(){
try{
....
lock.lock();
......
}finally {
.....
}
}查看lock內部邏輯，不過是調用了我們的aqs的acquire方法：
@Override
public void lock() {
//調用基于AQS實現好的邏輯即可
sync.acquire(1);
}步入其內部，從角度語義我們可以知道，這里首先會進行嘗試取鎖，若不成功則放到等待隊列中，并且打斷當前線程。
public final void acquire(int arg) {
//嘗試取鎖，若失敗則放到等待隊列中
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
//......
}了解了核心邏輯后，我們就開始了解每個細節，筆者會從嘗試取鎖細節、放到等待隊列細節、設置為獨占鎖細節、打斷線程細節逐個討論。
首先是嘗試取鎖的邏輯，實際上這就是我們重寫的邏輯
/**
* 嘗試取鎖
*
* @param arg
* @return
*/
@Override
protected boolean tryAcquire(int arg) {
//獲取當前狀態值
int state = getState();
//獲取當前線程
Thread currentThread = Thread.currentThread();
//若為0說明沒有線程拿到這個資源，當前線程可以基于CAS改變狀態值，若CAS修改成功則說明這個線程拿到資源了
if (0 == state) {
if (compareAndSetState(0, arg)) {
//設置當前資源擁有者為當前線程
setExclusiveOwnerThread(currentThread);
return true;
}
} else if (getExclusiveOwnerThread() == currentThread) {//不走上述邏輯，走到這里則說明這個資源當前線程之前搶到了，這里又搶了一次，我們再疊加狀態值即可
int newState = arg + state;
if (newState < 0) {
throw new Error("Maximum lock count exceeded");
}
setState(newState);
return true;
}
return false;
}接下來就是取不到鎖，放到等待隊列的邏輯，代碼如下：
private Node addWaiter(Node mode) {
//為當前線程創建一個node節點
Node node = new Node(Thread.currentThread(), mode);
// 獲取隊尾節點，通過CAS講當前節點添加到隊尾
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
//若是第一次入隊，則會走到該邏輯進行入隊操作
enq(node);
return node;
}2. 釋放鎖的邏輯
入口為筆者實現的unlock，其內部同樣用到的AQS的release方法：
@Override
public void unlock() {
//調用我們實現的嘗試釋放鎖邏輯
sync.release(1);
}我們看看核心邏輯，嘗試釋放鎖，若成功則獲取頭節點，若頭節點不為空且不為0(0代表在等待隊列中等待取鎖)，則unparkSuccessor喚醒
public final boolean release(int arg) {
//嘗試解鎖即通過CAS修改state成功
if (tryRelease(arg)) {
Node h = head;
//從頭節點開始找到后繼節點將其喚醒
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}了解核心邏輯后，我們展開聊聊嘗試釋放，unparkSuccessor邏輯首先是嘗試釋放鎖，同樣是筆者上文重寫的代碼就不貼出來
private void unparkSuccessor(Node node) {
//......
//獲取頭節點的后繼節點，若為空或者大于0(上文提到取消狀態CANCELLED = 1)，說明這個節點被取消了，那么我們就需要從后向前找到可以被喚醒的節點
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);
}六、共享鎖模式
1. 取鎖
代碼如下所示，嘗試取鎖成功就進行doAcquireShared，我們不妨看看doAcquireShared做了什么
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}doAcquireShared邏輯則是判斷當前節點的前驅是否為head，若是則獲取資源，若成功則設置當前節點為隊首，并看看資源還有沒有剩下若有則通知其他線程獲取。
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) {//若成功則將節點放到head，并將狀態設置為propagate(表示可以取共享鎖)
setHeadAndPropagate(node, r);
//輔助gc回收這個p節點
p.next = null; // help GC
//如果這個線程需要打斷，則打斷
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
//如果因為獲取資源失敗，則判斷p是否是SIGNAL狀態，若是則直接打斷，并設置 interrupted = true告知后節點也要一起等待
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}2. 釋放鎖
同樣的入口在releaseShared，或嘗試釋放從成功，則調用doReleaseShared：
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}doReleaseShared邏輯很簡單，從隊首開始，將SIGNAL狀態的節點設置為0意為可喚醒后續節點獲取鎖，然后調用unparkSuccessor找到可以喚醒的線程將其調用LockSupport.unpark將其喚醒。
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
//將頭節點設置為0成功后嘗試喚醒后繼節點
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
//......
}
if (h == head) // loop if head changed
break;
}
}七、相關面試題
1. AQS是什么？
AQS全名AbstractQueuedSynchronizer即抽象隊列同步器，用于實現多線程之間資源管理和調度的一個抽象類，類似ReentrantLock，Semaphore，其他的諸如 ReentrantReadWriteLock，SynchronousQueue，FutureTask都是基于AQS實現的。
2. AQS原理分析
AQS將線程當作一個個節點Node，存到CLH隊列中。搶到資源的線程就會被設置為占有資源的線程，并且可以操作一個線程共有可見參數state，所有線程都會CAS查看這個狀態的值從而判斷資源是否被占有線程釋放進而決定是否爭搶。
3. AQS對資源的兩種共享方式
- 獨占式(Exclusive):獨占以為著某個時間段資源只能被一個線程持有。并且獨占式鎖爭搶規則還分為公平和非公平兩種，公平鎖則是占有鎖的線程釋放鎖后，根據隊列順序獲取資源，非公平鎖則是無視隊列順序所有線程集體爭搶資源。
- 共享式:單位時間內，多個線程可以獲取資源，例如: CountDownLatch、Semaphore、 CyclicBarrier、ReadWriteLock等。
八、小結
至此，我們對 AQS 的源碼解析之旅已接近尾聲。在這趟探索中，我們深入了解了 AQS 的核心架構與關鍵機制。AQS 作為 Java 并發框架的基礎，通過一個 FIFO 隊列來管理等待獲取資源的線程，以 state 變量表示同步狀態，精妙地實現了資源的同步控制。
