Java並發之ReentrantLock源碼解析（二）
在了解如何加鎖時候，我們再來了解如何解鎖。可重入互斥鎖ReentrantLock的解鎖方法unlock()並不區分是公平鎖還是非公平鎖，Sync類並沒有實現release(int arg)方法，這裡會實現調用其父類AbstractQueuedSynchronizer的release(int arg)方法。在release(int arg)方法中，會先調用其子類實現的tryRelease(int arg)方法，這裡我們看看ReentrantLock.Sync.tryRelease(int releases)的實現。
正常情況下，這段程式碼只能由佔有鎖的執行緒調用，所以這裡會先獲取鎖的引用計數，再減去釋放次數，即：c = getState() - releases，然後判斷嘗試釋放鎖的執行緒，是否是獨佔鎖的執行緒，如果不是則拋出IllegalMonitorStateException異常。如果獨佔執行緒在釋放鎖後鎖的引用計數為0，則設置獨佔執行緒為null，再設置state為0，代表鎖成為無主狀態。
如果確定鎖成為無主狀態後，release(int arg)會檢查隊列中是否有需要喚醒的執行緒，如果頭節點header為null，則代表除了釋放鎖的執行緒，沒有任何執行緒搶鎖，如果頭節點的等待狀態為0，代表頭節點目前沒有需要喚醒的後繼節點。如果頭節點不為null且等待狀態不為0，則代表隊列中可能存在需要喚醒的執行緒，會進而執行unparkSuccessor(Node node)，將頭節點作為參數傳入。
當我們將頭節點傳入unparkSuccessor(Node node)，如果判斷頭節點的等待狀態<0，則會將頭節點的等待狀態設置為0，如果頭節點的後繼節點不為null，或者後繼節點尚未被取消，會直接喚醒後繼節點，後繼節點會退出parkAndCheckInterrupt()方法，acquireQueued(final Node node, int arg)的循環中重新競爭鎖，如果競爭成功，則後繼節點成為頭節點，退出搶鎖邏輯開始訪問資源，如果競爭失敗，這裡最多循環兩次執行shouldParkAfterFailedAcquire(Node pred, Node node)，第一次先將後繼節點的前驅節點的等待狀態由0改為SIGNAL（1），表示前驅節點的後繼節點處於等待喚醒狀態，第二次循環如果還是搶鎖失敗，shouldParkAfterFailedAcquire(Node pred, Node node)判斷前驅節點的等待狀態為SIGNAL，返回true，後繼節點的執行緒陷入阻塞。需要注意的是，即便是公平鎖，也可能存在不公平的情況，公平鎖頭節點的後繼節點，也有可能存在搶鎖失敗的情況，比如之前說過，公平鎖在調用tryLock()時是不保證公平的。
這裡我們需要注意一下，後繼節點是可能存在被移除的情況，這種情況會在後續講解tryLock(long timeout, TimeUnit unit)的時候說明，如果一旦出現後繼節點被移除的情況（waitStatus > 0），在喚醒後繼節點時，會從尾節點向前驅節點遍歷，找到最靠近頭節點的有效節點。
public class ReentrantLock implements Lock, java.io.Serializable {
//...
private final Sync sync;
//...
abstract static class Sync extends AbstractQueuedSynchronizer {
//...
protected final boolean tryRelease(int releases) {
int c = getState() - releases;
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}
//...
}
//...
public void unlock() {
sync.release(1);
}
//...
}
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
//...
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
//...
//由子類實現嘗試解鎖方法。
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}
//...
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)
node.compareAndSetWaitStatus(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 > 0) {
s = null;
//如果後繼節點不為null但被移除，會從尾節點向前驅節點遍歷，找到最靠前的有效節點
for (Node p = tail; p != node && p != null; p = p.prev)
if (p.waitStatus <= 0)
s = p;
}
if (s != null)
LockSupport.unpark(s.thread);
}
//...
}
可重入互斥鎖的tryLock()方法不分公平鎖或非公平鎖，統一調用ReentrantLock.Sync.nonfairTryAcquire(int acquires)，在JUC長征篇之ReentrantLock源碼解析（一）已經介紹過nonfairTryAcquire(int acquires)方法了，這裡就不再贅述了。
public class ReentrantLock implements Lock, java.io.Serializable {
//...
private final Sync sync;
//...
abstract static class Sync extends AbstractQueuedSynchronizer {
//...
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;
}
//...
}
//...
public boolean tryLock() {
return sync.nonfairTryAcquire(1);
}
//...
}
調用ReentrantLock.tryLock(long timeout, TimeUnit unit)方法時，會調用Sync父類AQS的tryAcquireNanos(int arg, long nanosTimeout)方法，如果執行緒已被標記為中斷，則會進入<1>處的分支並拋出InterruptedException異常，否則先調用tryAcquire(int acquires)嘗試搶奪，如果搶鎖失敗，則會調用doAcquireNanos(int arg, long nanosTimeout)陷入計時阻塞，如果在阻塞期間內執行緒被中斷，則會拋出InterruptedException異常，如果到達有效期後執行緒還未獲得鎖，tryAcquireNanos(int arg, long nanosTimeout)將返回false。
當進入doAcquireNanos(int arg, long nanosTimeout)後，會先在<2>處計算獲取鎖的截止時間，之後和原先的acquireQueued()很像，先把當前執行緒封裝成一個Node對象並加入到等待隊列，再判斷當前節點的前驅節點是否是頭節點，是的話再嘗試搶鎖，如果搶鎖成功則退出。否則用截止時間減去系統當前時間，算出執行緒剩餘的搶鎖時間（nanosTimeout）。如果剩餘時間<=0的話，代表到達截止時間後執行緒依舊未佔用鎖，於是調用<3>處的代表將執行緒對應的Node節點從等待隊列中移除，返回false表示搶鎖失敗。如果在後面的循環中如果發現當前時間大於截止時間，而執行緒還未獲得鎖，代表搶鎖失敗。
如果剩餘時間大於0，則會調用shouldParkAfterFailedAcquire()，如果前驅節點的狀態狀態為0，則將前驅節點的等待狀態設置為-1表示其後繼節點等待喚醒，然後在下一次循環的時候，shouldParkAfterFailedAcquire()判斷前置節點的等待狀態為-1，就有機會阻塞當前執行緒。但相比acquireQueued()不同的是，這裡會判斷剩餘時間是否大於1000納秒，如果剩餘時間小於等於1000納秒的話，這裡就不會阻塞執行緒，而是用自旋的方式，直到搶鎖成功，或者鎖超時搶鎖失敗。如果自旋期間或者阻塞期間執行緒被中斷，則會在<6>處拋出InterruptedException異常。
public class ReentrantLock implements Lock, java.io.Serializable {
//...
private final Sync sync;
//...
public boolean tryLock(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireNanos(1, unit.toNanos(timeout));
}
//...
}
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
static final long SPIN_FOR_TIMEOUT_THRESHOLD = 1000L;
//...
public final boolean tryAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted())//<1>
throw new InterruptedException();
return tryAcquire(arg) ||
doAcquireNanos(arg, nanosTimeout);
}
//...
private boolean doAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (nanosTimeout <= 0L)
return false;
final long deadline = System.nanoTime() + nanosTimeout;//<2>
final Node node = addWaiter(Node.EXCLUSIVE);
try {
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
return true;
}
nanosTimeout = deadline - System.nanoTime();
if (nanosTimeout <= 0L) {
cancelAcquire(node);//<3>
return false;
}
if (shouldParkAfterFailedAcquire(p, node) &&//<4>
nanosTimeout > SPIN_FOR_TIMEOUT_THRESHOLD)//<5>
LockSupport.parkNanos(this, nanosTimeout);
if (Thread.interrupted())
throw new InterruptedException();//<6>
}
} catch (Throwable t) {
cancelAcquire(node);
throw t;
}
}
//...
}
下面，我們來看看AQS是如何移除一個節點的，當要移除一個節點時，會先置空它的執行緒引用，再在<1>處遍歷當前節點的前驅節點，直到找到有效節點（等待狀態<=0），找到最靠近當前節點的有效節點pred後，再找到有效節點的後繼節點predNext（pred.next），然後我們將當前節點的等待狀態置為CANCELLED（1），如果當前節點是尾節點，則用CAS的方式設置尾節點為有效節點pred，如果pred成功設置為為節點，這裡還會用CAS的方式設置pred的後繼為null，因為尾節點不應該有後繼節點，這裡用CAS設置尾節點的後繼節點是防止pred成為尾節點後，還未置空尾節點的後繼節點前，又有新的節點入隊成為新的尾節點，並且設置pred的後繼節點為最新的尾節點。如果<2>出的程式碼執行成功，代表當前沒有新的節點入隊，如果執行失敗，代表有新的節點入隊，pred已經不是尾節點，且pred的後繼已經被修改。
如果node不是尾節點，或者在執行compareAndSetTail(node, pred)的時候，有新節點入隊，tail引用指向的對象已經不是node本身,或者在執行compareAndSetTail()執行失敗，則會進入<3>處的分支。進入<3>處的分支後，會先判斷當前節點的前驅節點是不是頭節點，如果是頭節點的話，會進入<5>處的分支，喚醒當前節點的後繼節點來競爭鎖，如果當前節點的後繼節點為null或者被取消的話，則會從尾節點開始遍歷前驅節點，找到隊列最前的有效節點，喚醒有效節點競爭鎖。
如果前驅節點不是頭節點，且前驅節點的等待狀態為SIGNAL或者前驅節點的等待狀態為有效狀態（waitStatus<=0）且成功設置前驅節點的等待狀態為SIGNAL，再判斷前驅節點前驅節點的thread引用是否為null，如果不為null則代表前驅節點還不是頭節點或者尚未被取消，此時就可以進入<4>處的分支，這裡如果判斷當前節點的後繼節點不為null或者尚未被取消，則用CAS的方式設置前驅節點的後繼節點，為當前節點的後繼節點。
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
//...
private void cancelAcquire(Node node) {
// Ignore if node doesn't exist
if (node == null)
return;
node.thread = null;
// Skip cancelled predecessors
Node pred = node.prev;
while (pred.waitStatus > 0)//<1>
node.prev = pred = pred.prev;
// predNext is the apparent node to unsplice. CASes below will
// fail if not, in which case, we lost race vs another cancel
// or signal, so no further action is necessary, although with
// a possibility that a cancelled node may transiently remain
// reachable.
Node predNext = pred.next;
// Can use unconditional write instead of CAS here.
// After this atomic step, other Nodes can skip past us.
// Before, we are free of interference from other threads.
node.waitStatus = Node.CANCELLED;
// If we are the tail, remove ourselves.
if (node == tail && compareAndSetTail(node, pred)) {
pred.compareAndSetNext(predNext, null);//<2>
} else {//<3>
// If successor needs signal, try to set pred's next-link
// so it will get one. Otherwise wake it up to propagate.
int ws;
if (pred != head &&
((ws = pred.waitStatus) == Node.SIGNAL ||
(ws <= 0 && pred.compareAndSetWaitStatus(ws, Node.SIGNAL))) &&
pred.thread != null) {//<4>
Node next = node.next;
if (next != null && next.waitStatus <= 0)
pred.compareAndSetNext(predNext, next);
} else {//<5>
unparkSuccessor(node);
}
node.next = node; // help GC
}
}
//...
}
先前我們說過，會存在後繼節點指向被取消節點的情況，就是發生在cancelAcquire(Node node)方法里，下圖是一個鎖的等待隊列，N1是隊頭，N1所對應的執行緒T1正佔有鎖進行資源訪問，N2和N5調用lock()方法採用非計時阻塞請求鎖，除非N2和N5對應的執行緒獲取到鎖，否則將永遠阻塞；N3和N4調用tryLock(long timeout, TimeUnit unit)方法採用計時阻塞請求鎖，如果超時對應的執行緒還未獲取到鎖，N3和N4將會從隊列中移除，返回搶鎖失敗。
我們假定N3和N4已經超時，要從隊列中移除，看看並發場景下是如何出現有效節點的後繼引用指向無效節點，這裡筆者稍微簡化cancelAcquire(Node node)，我們只要專註可能出現有效節點指向無效節點的程式碼。
假定N3和N4兩個節點對應的執行緒是T3和T4，T3要從等待隊列中移除N3，先獲取N3的前驅節點pred(N3)為N2，N2是一個有效節點（waitStatus<=0），所以不需要在<1>處遍歷，接著在<2>處獲取N2的後繼節點predNext(N2)為N3，再將N3的等待狀態改為CANCELLED，此時T3掛起，T4開始執行。T4執行緒同樣獲取N4前驅節點pred(n4)為N3，然後發現N3的等待狀態>0，會一直往前遍歷到N2，所以N4的前驅節點pred(N4)會為N2，接著T4執行<2>處的程式碼，predNext(N2)也是N3，此時T4掛起，T3恢復執行。T3判斷N3不是尾節點，於是進入分值<4>，T3判斷N3的前驅節點N2不是頭節點，N2的狀態為為SIGNAL，且N2的thread欄位不為空，表明N2既沒有被取消，也不是頭節點，於是進入<5>處的分值，這裡獲取N3的後繼節點N4，由於N4對應的執行緒T4尚未執行到<3>處的程式碼，N4的等待狀態依舊為SIGNAL，所以T3會進入<6>處的分支，將N2的後繼節點指向N4。此時T4開始執行，將N4的等待狀態改為CANCELLED，T4進行<4>處的分支，N4的前驅N2不是頭節點，等待狀態為SIGNAL，且N2的執行緒引用不為怒null，繼而進入<5>處的分值，獲取到N4的後繼N5，N5不Wie空，且N5的等待狀態<=0，進而進入到<6>分支，最後要用CAS的方式嘗試將N2的後繼節點設置為N5，但這裡的設置一定會失敗，因為此時N2的後繼節點為N4，而T4原先獲取到N2的後繼節點為N3，出現了有效節點指向無效節點的情況。
private void cancelAcquire(Node node) {
//...
Node pred = node.prev;
while (pred.waitStatus > 0)
node.prev = pred = pred.prev;//<1>
Node predNext = pred.next;//<2>
node.waitStatus = Node.CANCELLED;//<3>
if (node == tail && compareAndSetTail(node, pred)) {
//...
} else {//<4>
// If successor needs signal, try to set pred's next-link
// so it will get one. Otherwise wake it up to propagate.
int ws;
if (pred != head &&
((ws = pred.waitStatus) == Node.SIGNAL ||
(ws <= 0 && pred.compareAndSetWaitStatus(ws, Node.SIGNAL))) &&
pred.thread != null) {//<5>
Node next = node.next;
if (next != null && next.waitStatus <= 0)//<6>
pred.compareAndSetNext(predNext, next);
} else {
//...
}
//...
}
}
最後等待隊列的布局如下所示，N2指向無效節點N4，而非N4的後繼節點。同時這裡也引出另一個問題，哪怕N4不是無效節點，在上面移除節點的程式碼中，只設置了N2的後繼引用指向N4，卻沒設置N4的前驅引用指向N2，所以這裡N4的前驅依舊指向N3。
那麼如果真的出現上述這樣的情況，ReentrantLock是如何來修復這個隊列呢？答案在釋放鎖的時候調用unparkSuccessor(Node node)。筆者先前在介紹這個方法時，就有意提到後繼節點可能指向無效節點，當N1對應的執行緒使用完鎖釋放之後，N2對應的執行緒T2接著使用鎖並釋放鎖，在N2釋放鎖的時候，發現N2的後繼節點已經成為失效節點（waitStatus > 0），這裡會從尾節點開始找到隊列中最前面的有效節點，然後將其喚醒，這裡也就是N5。
private void unparkSuccessor(Node node) {
//...
Node s = node.next;
if (s == null || s.waitStatus > 0) {
s = null;
for (Node p = tail; p != node && p != null; p = p.prev)
if (p.waitStatus <= 0)
s = p;
}
if (s != null)
LockSupport.unpark(s.thread);
}
N5在被喚醒後，會調用shouldParkAfterFailedAcquire(Node pred, Node node)發現原先的前驅節點N4的等待狀態處於被移除，會進入<1>處的分支，查找到最靠近自己的有效前驅節點，並將前驅節點的後繼節點指向自己。這裡也能回答我們先前的問題，為何在要移除一個節點時，只修改前驅節點的後繼引用為被移除節點的後繼，卻不將被移除節點的後繼節點的前驅引用，指向其前驅，因為在釋放鎖的時候，會喚醒頭節點的後繼，如果被喚醒的後繼發現自己的前驅已經被移除，會往前查找最靠近自己的有效前驅，這裡一般是頭節點，接著將頭節點的後繼引用指向自己，再往後就是我們熟悉的流程了，如果前驅節點是頭節點且搶鎖成功，則退出acquireQueued()方法進行資源的訪問，如果搶鎖失敗，則最多執行兩次shouldParkAfterFailedAcquire()然後陷入阻塞，等待下一次的喚醒。
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
//...
if (ws > 0) {//<1>
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
//...
}
return false;
}
我們已經了解了lcok()、tryLock()以及tryLock(long timeout, TimeUnit unit)，下面的lockInterruptibly()就變得尤為簡單，lockInterruptibly()顧名思義也是無限期陷入阻塞，直到獲得鎖或者被中斷，如果執行緒本身被中斷，在搶鎖時會直接拋出InterruptedException異常，否則就開始搶鎖，如果搶鎖成功則皆大歡喜，搶鎖失敗則執行doAcquireInterruptibly(int arg)，這個方法相信不需要筆者做過多的介紹，基本上很多步驟和方法在上面已經介紹過了。將執行緒封裝成Node節點併入隊，判斷Node節點的前驅是否是頭節點，是的話則試圖搶鎖，如果不是的話則最多循環兩次執行shouldParkAfterFailedAcquire()，然後陷入阻塞，直到前驅節點成為頭節點並釋放鎖將其喚醒，或者執行緒被中斷，拋出InterruptedException異常。
public class ReentrantLock implements Lock, java.io.Serializable {
//...
private final Sync sync;
//...
public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
//..
}
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
//...
public final void acquireInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (!tryAcquire(arg))
doAcquireInterruptibly(arg);
}
//...
private void doAcquireInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.EXCLUSIVE);
try {
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
return;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} catch (Throwable t) {
cancelAcquire(node);
throw t;
}
}
//...
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
//...
}
最後，筆者就簡單介紹下公平鎖（FairSync）的tryAcquire(int acquires)方法，如果我們查看ReentrantLock的源碼會發現，在執行lock()、tryLock(long timeout, TimeUnit unit)和lockInterruptibly()，這幾個方法最終都會調用到AQS對應的三個方法：acquire(int arg)、tryAcquireNanos(int arg, long nanosTimeout)、acquireInterruptibly(int arg)，這三個方法在搶鎖的時候會優先執行子類實現的tryAcquire(int arg)方法，也就是公平鎖（FairSync）或者非公平鎖（NonfairSync）實現的tryAcquire(int arg)方法，搶鎖失敗再執行AQS自身實現的入隊、阻塞方法。
下面，我們來看下公平鎖實現的tryAcquire(int acquires)方法，其實這個方法的實現也非常簡單，先判斷鎖目前是否處於無主狀態，是的話再判斷隊列中是否有等待執行緒，確認鎖是無主狀態且隊列中沒有等待執行緒，便開始嘗試搶鎖，搶鎖成功則直接返回。公平鎖相較於非公平鎖的搶鎖邏輯，也僅僅是多了一步而已，判斷鎖是否無主，是的話再判斷隊列中是否有等待執行緒，不像非公平鎖，只要判斷是無主執行緒便不再查看等待隊列，直接嘗試搶鎖。
static final class FairSync extends Sync {
private static final long serialVersionUID = -3000897897090466540L;
/**
* Fair version of tryAcquire. Don't grant access unless
* recursive call or no waiters or is first.
*/
@ReservedStackAccess
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
//如果state為0，且隊列中沒有等待執行緒，則嘗試搶鎖
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;
}
}
