什么是AQS？
AQS即AbstractQueuedSynchronizer,是一個用於構建鎖和同步器的框架。它能降低構建鎖和同步器的工作量，還可以避免處理多個位置上發生的競爭問題。在基於AQS構建的同步器中，只可能在一個時刻發生阻塞，從而降低上下文切換的開銷，並提高吞吐量。
AQS支持獨占鎖（exclusive）和共享鎖(share)兩種模式。
- 獨占鎖：只能被一個線程獲取到(Reentrantlock)
- 共享鎖：可以被多個線程同時獲取(CountDownLatch,ReadWriteLock).
無論是獨占鎖還是共享鎖，本質上都是對AQS內部的一個變量state的獲取。state是一個原子的int變量，用來表示鎖狀態、資源數等。
AQS內部的數據結構與原理
AQS內部實現了兩個隊列，一個同步隊列，一個條件隊列。
同步隊列的作用是：當線程獲取資源失敗之后，就進入同步隊列的尾部保持自旋等待，不斷判斷自己是否是鏈表的頭節點，如果是頭節點，就不斷參試獲取資源，獲取成功后則退出同步隊列。
條件隊列是為Lock實現的一個基礎同步器，並且一個線程可能會有多個條件隊列，只有在使用了Condition才會存在條件隊列。
同步隊列和條件隊列都是由一個個Node組成的。AQS內部有一個靜態內部類Node。
static final class Node {
static final Node EXCLUSIVE = null;
//當前節點由於超時或中斷被取消
static final int CANCELLED = 1;
//表示當前節點的前節點被阻塞
static final int SIGNAL = -1;
//當前節點在等待condition
static final int CONDITION = -2;
//狀態需要向后傳播
static final int PROPAGATE = -3;
volatile int waitStatus;
volatile Node prev;
volatile Node next;
volatile Thread thread;
Node nextWaiter;
final boolean isShared() {
return nextWaiter == SHARED;
}
final Node predecessor() throws NullPointerException {
Node p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}
Node() { // Used to establish initial head or SHARED marker
}
Node(Thread thread, Node mode) { // Used by addWaiter
this.nextWaiter = mode;
this.thread = thread;
}
Node(Thread thread, int waitStatus) { // Used by Condition
this.waitStatus = waitStatus;
this.thread = thread;
}
}
重要方法的源碼解析
//獨占模式下獲取資源
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
acquire(int arg)首先調用tryAcquire(arg)嘗試直接獲取資源，如果獲取成功，因為與運算的短路性質，就不再執行后面的判斷，直接返回。tryAcquire(int arg)的具體實現由子類負責。如果沒有直接獲取到資源，就將當前線程加入等待隊列的尾部，並標記為獨占模式，使線程在等待隊列中自旋等待獲取資源，直到獲取資源成功才返回。如果線程在等待的過程中被中斷過，就返回true，否則返回false。
如果acquireQueued(addWaiter(Node.EXCLUSIVE), arg)執行過程中被中斷過，那么if語句的條件就全部成立，就會執行 selfInterrupt();方法。因為在等待隊列中自旋狀態的線程是不會響應中斷的，它會把中斷記錄下來，如果在自旋時發生過中斷，就返回true。然后就會執行selfInterrupt()方法，而這個方法就是簡單的中斷當前線程Thread.currentThread().interrupt();其作用就是補上在自旋時沒有響應的中斷。
可以看出在整個方法中，最重要的就是 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)。
我們首先看Node addWaiter(Node mode)方法，顧名思義，這個方法的作用就是添加一個等待者，根據之前對AQS中數據結構的分析，可以知道，添加等待者就是將該節點加入等待隊列.
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
enq(node);
return node;
}
//執行入隊
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;//打斷循環
}
}
}
}
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();//拿到node的上一個節點
//前置節點為head，說明可以嘗試獲取資源。排隊成功后，嘗試拿鎖
if (p == head && tryAcquire(arg)) {
setHead(node);//獲取成功，更新head節點
p.next = null; // help GC
failed = false;
return interrupted;
}
//嘗試拿鎖失敗后，根據條件進行park
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
//獲取資源失敗后，檢測並更新等待狀態
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
do {
//如果前節點取消了，那就往前找到一個等待狀態的接待你，並排在它的后面
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/*
* waitStatus must be 0 or PROPAGATE. Indicate that we
* need a signal, but don't park yet. Caller will need to
* retry to make sure it cannot acquire before parking.
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
//阻塞當前線程，返回中斷狀態
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
具體的boolean tryAcquire(int acquires)實現有所不同。
公平鎖的實現如下：
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
非公平鎖的實現如下：
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
park
在AQS的實現中有一個出現了一個park的概念。park即LockSupport.park().它的作用是阻塞當前線程，並且可以調用LockSupport.unpark(Thread)去停止阻塞。它們的實質都是通過UnSafe類使用了CPU的原語。在AQS中使用park的主要作用是，讓排隊的線程阻塞掉（停止其自旋，自旋會消耗CPU資源），並在需要的時候，可以方便的喚醒阻塞掉的線程。