同步器AbstractQueuedSynchronizer淺析
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中沒有具體的實現(只拋異常),需要子類去覆寫。