同步器AbstractQueuedSynchronizer淺析

Java · 發表 2019-02-26 10:25:08

摘要： Java中的鎖主要有：synchronized鎖和JUC(java.util.concurrent)locks包中的鎖。synchronized鎖是JVM的內建鎖，底層通過"monitorenter"和"monitorexit"位元組碼指令實現。JUC中的鎖支援...

Java中的鎖主要有：synchronized鎖和JUC(java.util.concurrent)locks包中的鎖。synchronized鎖是JVM的內建鎖，底層通過"monitorenter"和"monitorexit"位元組碼指令實現。JUC中的鎖支援公平鎖（synchronized鎖是非公平鎖），讀寫鎖，鎖請求中斷，鎖請求超時等。今天要說的AbstractQueuedSynchronizer（AQS）是JUC鎖的基礎。JUC中的ReentrantLock，ReentrantReadWriteLock，Semaphore，CountDownLatch等都用到了AQS作為同步器。可以說AQS是JUC(java.util.concurrent)的基礎。

同步佇列

AQS本質上是一個FIFO的佇列，它的等待佇列是“CLH”（Craig, Landin, and Hagersten）佇列的變種。CLH佇列通常用作自旋鎖（spinlocks）。

AQS使用一個int的state來記錄當前鎖的狀態：

private volatile int state; // 狀態
protected final int getState() {
return state;
}
protected final void setState(int newState) {
state = newState;
}
protected final boolean compareAndSetState(int expect, int update) { // 通過CAS設定狀態
// See below for intrinsics setup to support this
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
複製程式碼

AQS支援兩種鎖模式：獨佔鎖和共享鎖。如果當前AQS獨佔鎖被獲取後，在獨佔鎖執行緒未釋放之前，其他的獨佔鎖和共享鎖請求都將被阻塞；如果共享鎖被獲取，獨佔鎖請求將被阻塞，而其他的共享鎖請求可以成功。在讀寫鎖ReentrantReadWriteLock中，AQS的高16位表示讀鎖（共享鎖）狀態，低16位表示寫鎖（獨佔鎖）狀態。

Node類

static final class Node {

// 標識當前節點在等待共享鎖
static final Node SHARED = new Node();
// 標識當前節點在等待獨佔鎖
static final Node EXCLUSIVE = null;

// 當前節點等待被中斷或者超時
static final int CANCELLED =1;
// 當前節點等待取消或者釋放鎖之後需要unpark它的後繼節點
static final int SIGNAL= -1;
// 當前節點在condition queue中等待
static final int CONDITION = -2;
// 只有頭節點才能設定改狀態，當請求處於共享狀態下時，當前執行緒被喚醒之後可能還需要喚醒其他執行緒。後續節點需要傳播該喚醒動作
static final int PROPAGATE = -3;
// 當前節點的等待狀態
volatile int waitStatus;

// 前驅等待節點
volatile Node prev;
// 後置等待節點
volatile Node next;

// 當前節點關聯的執行緒
volatile Thread thread;

// 指向condition queue等待的節點，或者指向SHARE節點，表明當前處於共享模式
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;
}
}
複製程式碼

Node中定義了CANCELLED、SIGNAL、CONDITION和PROPAGATE四種狀態：

CANCELLED（1）：代表當前節點等待超時或者被中斷。
SIGNAL（-1）：當前節點取消或者釋放鎖之後通知它後繼節點需要被喚醒。
CONDITION（-2）：當前節點在某個條件佇列上等待。
PROPAGATE（-3）：只有頭節點才會設定為改狀態，表明當前處於共享模式中，節點被喚醒之後需要傳播喚醒動作，繼續喚醒其他的節點。

API

AQS中已經實現的與加鎖和解鎖有關的方法如下：

方法	作用
acquire(int)	獲取獨佔鎖，不可中斷，可能執行緒會進入佇列中等待
acquireInterruptibly(int)	獲取獨佔鎖，可以中斷
tryAcquireNanos(int, long)	在指定時間之內嘗試獲取獨佔鎖
release(int)	釋放獨佔鎖
acquireShared(int)	獲取共享鎖，不可中斷
acquireSharedInterruptibly(int)	獲取共享鎖，可以中斷
tryAcquireSharedNanos(int, long)	在指定時間之內嘗試獲取共享鎖，可以中斷
releaseShared(int)	釋放共享鎖

獨佔鎖的獲取和釋放

獨佔鎖的獲取

acquire用於獲取獨佔鎖，請求不可中斷，首先會通過tryAcquire方法獲取鎖，如果獲取失敗，則進入等待佇列中。tryAcquire方法在AQS中預設丟擲一個異常，需要子類去實現具體的樂觀獲取獨佔鎖的方式：

protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
複製程式碼

如果tryAcquire獲取失敗，則通過addWaiter方法生成一個關聯當前執行緒的waiter節點放入佇列中，再通過acquireQueued方法獲取鎖。

// 將當前執行緒放入佇列中等待
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)) { // 通過CAS將節點放入佇列的尾部
pred.next = node;
return node;
}
}
enq(node);
return node;
}

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)) { // head不關聯任何執行緒，是一個dummy節點，如果當前節點的前置節點是head，則通過tryAcquire方法嘗試獲取鎖
setHead(node); // 獲取鎖成功，設定當前節點為head，在setHead中會將node的thread和prev指標置為null。因為head節點不關聯任何執行緒
p.next = null; // help GC
failed = false;
return interrupted;
}
// shouldParkAfterFailedAcquire判斷當前節點獲取鎖失敗後是否需要掛起當前執行緒（park），如果需要掛起，則通過parkAndCheckInterrupt方法掛起執行緒（LockSupport.park），然後清除執行緒的中斷狀態(Thread.interrupted)。
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true; // 返回執行緒已經被中斷
}
} finally {
if (failed) // 最後如果失敗，則取消鎖請求
cancelAcquire(node);
}
}
複製程式碼

acquireQueued方法中是一個迴圈，如果判斷當前節點是佇列中的第一個節點並且通過tryAcquire獲取到鎖之後通過setHead將當前節點設定為head，在setHead方法中將head的thread和prev指標置為null，因為head不會關聯任何執行緒。如果獲取鎖失敗，則通過shouldParkAfterFailedAcquire判斷當前節點獲取鎖失敗後是否需要掛起當前執行緒，如果需要掛起當前執行緒，則通過parkAndCheckInterrupt方法來掛起當前執行緒，等待它的predecessor節點喚醒：

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL) // 當前節點的predecessor節點的waitStatus為SIGNAL，當predecessor釋放鎖時會喚醒當前執行緒
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
if (ws > 0) { // 找到waitStatus不是CANCELLED的節點
/*
* 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); // 設定predecessor的waitStatus為SIGNAL，代表當前節點需要predecessor的signal訊號，但是當前執行緒還未掛起，執行緒需要在掛起之前需要重試
}
return false;
}

複製程式碼

下面是獲取獨佔鎖的流程圖：

獨佔鎖的釋放

public final boolean release(int arg) {
if (tryRelease(arg)) { // tryRelease由子類複寫
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h); // 喚醒後繼節點
return true;
}
return false;
}

private void unparkSuccessor(Node node) {
/*
* If status is negative (i.e., possibly needing signal) try
* to clear in anticipation of signalling.It is OK if this
* fails or if status is changed by waiting thread.
*/
int ws = node.waitStatus;
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.
*/
// 如果當前節點的next節點為空或者next節點等待被取消則從從tail開始尋找可被喚醒的節點（waitStatus <= 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); // 喚醒執行緒
}
複製程式碼

獨佔鎖的釋放過程比較簡單：

通過tryRelease釋放佔有的資源，子類需要複寫該方法，修改AQS中的state。
喚醒後繼等待的節點：找到waitStatus不是CANCELLED的節點，然後通過LockSupport.unpark方法喚醒該執行緒。

釋放獨佔鎖的流程圖如下所示：

共享鎖的獲取和釋放

共享鎖的獲取

public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0) // 首先tryAcquireShared，如果成功則直接return，否則doAcquireShared
doAcquireShared(arg);
}

private void doAcquireShared(int arg) {
final Node node = addWaiter(Node.SHARED); // 將當前執行緒包裝成Node放入佇列中
boolean failed = true;
try {
boolean interrupted = false;
for (;;) { // 迴圈重試
final Node p = node.predecessor(); // 當前節點的前置節點
if (p == head) { // 如果當前節點的predecessor節點為head，則tryAcquireShared
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r); // 設定當前節點為頭節點，並判斷後繼節點是否需要喚醒
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
// 前面說過這段程式碼，shouldParkAfterFailedAcquire判斷當前節點獲取鎖失敗後是否需要掛起當前執行緒（park），如果需要掛起，則通過parkAndCheckInterrupt方法掛起執行緒（LockSupport.park），然後清除執行緒的中斷狀態(Thread.interrupted)。
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
複製程式碼

首先通過tryAcquireShared方法來獲取共享鎖，如果獲取成功，則直接返回。否則將當前執行緒放入等待佇列中，迴圈重試獲取鎖：如果當前節點的前置節點為head，則通過tryAcquireShared來獲取鎖，如果獲取成功，則設定頭節點，並判斷後續節點是否需要喚醒；如果獲取失敗，則判斷當前節點是否需要（shouldParkAfterFailedAcquire）掛起(LockSupport.lock)，如果需要掛起執行緒，則通過LockSupport.park方法將當前執行緒掛起。下面是獲取共享鎖的流程圖:

釋放共享鎖

public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) { // 如果tryReleaseShared成功則直接返回，否則doReleaseShared。tryReleaseShared需要子類複寫該方法
doReleaseShared();
return true;
}
return false;
}

private void doReleaseShared() {
/*
* Ensure that a release propagates, even if there are other
* in-progress acquires/releases.This proceeds in the usual
* way of trying to unparkSuccessor of head if it needs
* signal. But if it does not, status is set to PROPAGATE to
* ensure that upon release, propagation continues.
* Additionally, we must loop in case a new node is added
* while we are doing this. Also, unlike other uses of
* unparkSuccessor, we need to know if CAS to reset status
* fails, if so rechecking.
*/
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) { // waitStatus為SIGNAL的狀態，需要喚醒後續節點
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) // CAS充實將當前節點狀態設為0
continue;// loop to recheck cases
unparkSuccessor(h); // 喚醒後繼節點
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) // 狀態為0，沒有後續節點需要喚醒，CAS設定狀態為PROPAGATE
continue;// loop on failed CAS
}
if (h == head)// loop if head changed
break;
}
}
複製程式碼

doReleaseShared方法中，判斷當前節點的waitStatus，如果為SIGNAL，說明需要喚醒後續節點，喚醒之前先CAS重試把節點狀態修改為0，然後unparkSuccessor喚醒後續節點。如果當前節點的waitStatus為0，則說明後續沒有節點需要喚醒，CAS重試將當前節點waitStatus狀態修改為PROPAGATE。下圖是釋放共享鎖的流程圖：

超時和中斷

AQS支援加鎖超時和中斷機制，這也是JUC鎖相對synchronized鎖的優勢。AQS支援獨佔鎖和共享鎖的超時和中斷， public final boolean tryAcquireNanos(int arg, long nanosTimeout) 用於在超時時間之內獲取獨佔鎖， public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) 用於在超時時間之內獲取共享鎖。

public final boolean tryAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted()) // 支援響應中斷請求，首先判斷當前執行緒是否被中斷，如果中斷了，則丟擲異常，結束
throw new InterruptedException();
// tryAcquire需要子類實現，主要的加鎖邏輯在doAcquireNanos方法中
return tryAcquire(arg) ||
doAcquireNanos(arg, nanosTimeout);
}
複製程式碼

tryAcquireNanos可以響應中斷請求，首先檢查執行緒是否被中斷，如果被中斷，則直接丟擲異常，加鎖失敗。否則先通過tryAcquire嘗試獲取鎖，如果成功則直接返回。如果失敗，則通過doAcquireNanos來加鎖。

private boolean doAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (nanosTimeout <= 0L)
return false;
final long deadline = System.nanoTime() + nanosTimeout; // 計算本次請求的deadline
final Node node = addWaiter(Node.EXCLUSIVE); // 將當前執行緒加入等待佇列中
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) { // 如果當前節點的predecessor為head，則tryAcquire嘗試獲取鎖，如果成功則設定當前節點為頭節點，返回true
setHead(node); // 設定當前節點為head，設定head節點的thread和prev指標為null
p.next = null; // help GC
failed = false;
return true;
}
nanosTimeout = deadline - System.nanoTime(); // 截止deadline之前本次請求還剩餘的時間
if (nanosTimeout <= 0L) // nanosTimeout小於等於0，說明本次deadline已經到了，返回false，加鎖失敗
return false;
// 本次加鎖失敗，通過shouldParkAfterFailedAcquire判斷是否需要掛起當前執行緒，如果需要掛起向前執行緒，並且nanosTimeout大於spinForTimeoutThreshold，則掛起當前執行緒nanosTime。
// 這裡的spinForTimeoutThreshold相當於一個掛起執行緒的最小時間閾值，如果小於等於這個時間，則直接重試就可以了，而不是掛起執行緒，因為掛起和喚醒執行緒是有效能開銷的
if (shouldParkAfterFailedAcquire(p, node) &&
nanosTimeout > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanosTimeout); // 掛起執行緒
if (Thread.interrupted()) // 響應中斷請求
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
複製程式碼

根據使用者設定的超時時間，計算本次加鎖的deadline，在迴圈體中判斷當前時間是否已經超過deadline，如果超過返回false，加鎖失敗。迴圈體中首先判斷當前節點的predecessor是不是head，如果是head，則嘗試加鎖，如果加鎖成功則設定當前節點為head。如果加鎖失敗，計算本次離deadline還剩多長時間，如果已經到了deadline則返回false，加鎖失敗。如果還未到deadline，如果shouldParkAfterFailedAcquire為true並且nanosTimeout大於spinForTimeoutThreshold則掛起當前執行緒。否則先響應中斷請求再迴圈重試。流程圖如下：

總結

AQS是JUC鎖的基礎，ReentrantLock，ReentrantReadWriteLock，Semaphore，CountDownLatch都用到了AQS作為其同步器。
AQS本質是上一個FIFO佇列，它使用一個int state來表示當前同步狀態，提供了setState，getState和compareAndSetWaitStatus來獲取和設定狀態。
AQS中已經實現了acquire，acquireInterruptibly，tryAcquireNanos，release，acquireShared，acquireSharedInterruptibly，tryAcquireSharedNanos和releaseShared等方法。tryAcquire，tryRelease，tryAcquireShared，tryReleaseShared這些方法在AQS中沒有具體的實現（只拋異常），需要子類去覆寫。