深入 ReentrantLock 內部：公平鎖與非公平鎖之奧秘
1 前言
在Java的JUC包中，提供了一個強大的鎖工具ReentrantLock，在多線程編程時，我們會時常用到。而其中的公平鎖與非公平鎖更是有著獨特的魅力和重要的作用。理解公平鎖與非公平鎖的區別，對于優化程序性能、確保資源的合理分配至關重要。
下面，我們將深入探討ReentrantLock的公平鎖與非公平鎖，帶你揭開它們的神秘面紗，掌握多線程編程的關鍵技巧。那么接下來，讓我們一起開啟這場探索之旅吧！
2 公平 VS 非公平鎖
首先我們先來了解下什么是公平鎖和非公平鎖。
公平鎖：指多個線程按照申請鎖的順序來獲取鎖。在公平鎖機制下，線程獲取鎖的順序是遵循先來后到的原則，就像在排隊一樣。
非公平鎖：指多個線程獲取鎖的順序是不確定的。當一個線程釋放鎖后，新請求鎖的線程有可能立即獲取到鎖，而不管在它之前是否還有其他等待的線程。
3 ReentrantLock公平鎖和非公平鎖
3.1 繼承關系圖譜
圖片
通過繼承關系圖譜，我們可以看到ReentrantLock類實現了Serializable接口和Lock接口，另外其內部定義了3個內部類Sync、NonfairSync、FairSync。Sycn是一個抽象類實現了AbstractQueuedSynchronizer（下文簡稱AQS），NonfairSync、FairSync為Sync的實現子類，通過類的命名其實我們就可以知道NonfairSync為非公平鎖的實現類，FairSync為公平鎖的實現類，而Sycn為抽象出來的公共抽象類。
3.2 創建公平鎖與非公平鎖
ReentrantLock中提供了兩個構造函數，一個是默認的構造函數，另一個是有參構造函數，通過布爾值參數控制創建鎖對象的類型?？梢钥吹绞褂媚J構造函數默認創建的是非公平鎖，使用有參構造函數參數為true時，創建的為公平鎖，參數為false時，創建的為非公平鎖。
/**
* 無參構造器
* 說明：從構造器內部實現可知默認構造的鎖為非公平鎖
*/
public ReentrantLock() {
sync = new NonfairSync();
}
/**
* 有參構造器
* 說明：fair參數設定構造的對象是公平鎖還是非公平鎖
* true：公平鎖
* false：非公平鎖
*/
public ReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}3.3 使用示例
3.3.1 非公平鎖
@Test
public void testUnfairLock() throws InterruptedException {
// 無參構造函數，默認創建非公平鎖模式
ReentrantLock lock = new ReentrantLock();
for (int i = 0; i < 6; i++) {
final int threadNum = i;
new Thread(() -> {
//獲取鎖
lock.lock();
try {
System.out.println("線程" + threadNum + "獲取鎖");
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// finally中解鎖
lock.unlock();
System.out.println("線程" + threadNum +"釋放鎖");
}
}).start();
Thread.sleep(999);
}
Thread.sleep(100000);
}運行結果：
線程0獲取鎖
線程0釋放鎖
線程1獲取鎖
線程1釋放鎖
線程3獲取鎖
線程3釋放鎖
線程2獲取鎖
線程2釋放鎖
線程5獲取鎖
線程5釋放鎖
線程4獲取鎖
線程4釋放鎖3.3.2 公平鎖
@Test
public void testfairLock() throws InterruptedException {
// 有參構造函數，true表示公平鎖，false表示非公平鎖
ReentrantLock lock = new ReentrantLock(true);
for (int i = 0; i < 6; i++) {
final int threadNum = i;
new Thread(() -> {
//獲取鎖
lock.lock();
try {
System.out.println("線程" + threadNum + "獲取鎖");
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// finally中解鎖
lock.unlock();
System.out.println("線程" + threadNum +"釋放鎖");
}
}).start();
Thread.sleep(10);
}
Thread.sleep(100000);
}運行結果：
線程0獲取鎖
線程0釋放鎖
線程1獲取鎖
線程1釋放鎖
線程2獲取鎖
線程2釋放鎖
線程3獲取鎖
線程3釋放鎖
線程4獲取鎖
線程4釋放鎖
線程5獲取鎖
線程5釋放鎖3.4 實現原理分析
接下來，我們從ReentrantLock提供的兩個核心API加鎖方法lock()和解鎖方法unlock()為入口，繼續深入探索其內部公平鎖和非公平鎖的實現原理。
3.4.1 加鎖流程剖析
1）ReentrantLock.lock()方法為ReentrantLock提供的加鎖方法。公平鎖和非公平鎖都可以通過該方法來獲取鎖，區別在于其內部的sync引用的實例對象不同，公平鎖時，sync引用的為FairSync對象，非公平鎖時，sync引用的為NonfairSync對象。
public void lock() {
sync.lock();
}2）那FairSync和NonfairSync中lock()方法的具體實現有哪些不同呢？
通過下面的代碼對比我們可以看到FairSync.lock()方法實現是直接調用了AQS提供的acquire()方法。而NonfairSync.lock()方法實現是先通過CAS的方式先嘗試獲取了一次鎖，如果嘗試成功則直接將當前線程設置為占用鎖線程，而獲取失敗時同樣調用了AQS提供的acquire()方法。從這里可以看到非公平鎖獲取鎖時，如果當前鎖未被其他任何線程占用時，當前線程是有一次機會直接獲取到鎖的，而從公平鎖的方法實現中我們還無法看到公平鎖是如何實現，那我們繼續深入看下AQS提供的acquire()方法的實現。
/**
* FairSync.lock()方法實現
**/
final void lock() {
//調用的AQS中提供的的實現獨占鎖方法
acquire(1);
}/**
* NonfairSync.lock()方法實現
**/
final void lock() {
//通過CAS的方式嘗試獲取鎖
if (compareAndSetState(0, 1))
//獲取鎖成功則將當前線程設置為占用鎖線程
setExclusiveOwnerThread(Thread.currentThread());
else
//未成功獲取到鎖，調用AQS中的acquire()方法，再次嘗試獲取鎖
acquire(1);
}3）AbstractQueuedSynchronizer.acquire()方法，該方法是AQS實現獨占鎖的核心方法，主要的邏輯都在if判斷條件中，這里面有3個重要的方法tryAcquire()，addWaiter()和acquireQueued()。這三個方法中分別封裝了加鎖流程中的主要處理邏輯。
方法中首先調用了tryAcquire()方法進行嘗試獲取鎖。如果嘗試獲取失敗則會調用addWaiter()方法將獲取鎖失敗的線程加入到等待隊列中，然后將addWaiter()方法返回的結果作為參數，繼續調用acquireQueued()方法，此時當前線程會不斷嘗試獲取鎖，當獲取鎖成功后則會調用selfInterrupt()方法喚醒線程繼續執行。
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}我們繼續層層剖析！分別看下tryAcquire()，addWaiter()和acquireQueued()的源碼實現。
4）AbstractQueuedSynchronizer.tryAcquire()方法，該方法默認拋出了UnsupportedOperationException異常，自身未提供具體實現，此方法為AQS提供的鉤子模版方法，由子類同步組件通過擴展該方法實現嘗試獲取鎖邏輯。FairSync和NonfairSync分別重寫了該方法并提供了不同的實現。
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}5）FairSync和NonfairSync中tryAcquire()方法重寫實現。
通過下圖中的源碼對比，我們可以明顯的看出公平鎖與非公平鎖主要區別就在于公平鎖在獲取同步狀態時多了一個限制條件：hasQueuedPredecessors()，而其他的代碼流程是基本一致的，那我們再進入hasQueuedPredecessors()方法看一下。
/**
* FairSync.lock()方法實現
**/
protected final boolean tryAcquire(int acquires) {
//獲取當前線程對象
final Thread current = Thread.currentThread();
//獲取當前鎖的狀態
int c = getState();
//狀態為0時表示鎖未被占用
if (c == 0) {
//首先調用hasQueuedPredecessors()方法，檢查隊列中是否存在等待執行的線程，如果隊列中有待執行的線程，會優先讓隊列中的線程執行，這是公平鎖實現的核心
if (!hasQueuedPredecessors() &&
//如果hasQueuedPredecessors()這個方法返回false，則表示隊列中沒有等待執行的線程，那么會繼續調用compareAndSetState(0, acquires)方法，通過cas嘗試獲取鎖
compareAndSetState(0, acquires)) {
//如果獲取鎖成功，設置當前線程對象為占用鎖的線程對象
setExclusiveOwnerThread(current);
//返回獲取鎖成功
return true;
}
//如果 current == getExclusiveOwnerThread() 相等，說明當前線程與占用鎖的線程是是同一個線程，則也會被認為獲取鎖成功，即：重入鎖
} else if (current == getExclusiveOwnerThread()) {
//疊加重入次數
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
//更新鎖重入次數
setState(nextc);
//返回獲取鎖成功
return true;
}
//返回獲取鎖失敗
return false;
}/**
* NonfairSync.lock()方法
**/
protected final boolean tryAcquire(int acquies) {
//繼續調用父抽象類Sync類中的nonfairTryAcquire方法
return nonfairTryAcquire(acquires);
}
/**
* Sync.nonfairTryAcquire()方法
**/
final boolean nonfairTryAcquire(int acquires) {
//獲取當前線程對象實例
final Thread current = Thread.currentThread();
//獲取state變量的值，即當前鎖被重入的次數
int c = getState();
//state值為0，說明當前鎖未被任何線程持有
if (c == 0) {
//以cas方式獲取鎖
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;
}6）FairSync.hasQueuedPredecessors()方法，可以看到該方法主要做一件事情：主要是判斷檢查隊列是否存在等待執行的線程，并且頭部等待線程非當前線程。如果是則返回true，否則返回false。該方法也是公平鎖實現的核心。當隊列中已存在其他等待中的線程時，則會獲取鎖失敗，會調用AbstractQueuedSynchronizer.addWaiter()方法將當前線程放入等待隊列的尾部來排隊獲取鎖。
/**
* 判斷檢查隊列頭部是否存在等待執行的線程，并且等待線程非當前線程
*
* @return
*/
public final boolean hasQueuedPredecessors() {
Node t = tail;
Node h = head;
Node s;
/**
* h != t，如果頭結點等于尾節點，說明隊列中無數據，則說明隊列中沒有等待處理的節點
* (s = h.next) == null，頭節點的下一個節點為空 返回true
* s.thread != Thread.currentThread() 頭結點的下一個節點(即將執行的節點)所擁有的線程不是當前線程，返回true，說明隊列中有即將執行的節點。
*/
return h != t &&
((s = h.next) == null || s.thread != Thread.currentThread());
}為了方便大家理解，下面羅列了此方法返回true和返回false的場景圖解：
圖片
7）AbstractQueuedSynchronizer.addWaiter()方法，該方法主要是將獲取鎖失敗的線程加入到等待隊列的尾部，也就是進行排隊，如果隊列已初始化完成則直接將線程加入到隊列尾部，如果隊列尚未初始化，則會調用AbstractQueuedSynchronizer.enq()方法來完成隊列的初始化再將當前線程加入到隊列尾部。
/**
* 將獲取鎖失敗的線程加入到等待隊列中
*
* return 返回新加入的節點對象
*/
private Node addWaiter(Node mode) {
//創建新的節點，設置節點線程為當前線程，模式為獨占模式
Node node = new Node(Thread.currentThread(), mode);
//pred引用尾節點
Node pred = tail;
//判定是否有尾節點
if (pred != null) {
//存在尾節點將當前節點的前驅指針指向尾節點
node.prev = pred;
//通過cas將當前節點設置為尾幾點，當期望尾節點為pred時，則將當前node節點更新為尾節點
if (compareAndSetTail(pred, node)) {
//將原尾節點的后繼指針指向當前節點，這里是雙向鏈表 node.prev = pred; pred.next = node; 構成雙向鏈表
pred.next = node;
//設置成功返回當前節點
return node;
}
}
//如果沒有尾節點說明隊列還未初始化，那么將進行初始化，并將當前節點添加值隊列尾部
enq(node);
return node;
}流程圖解：
圖片
8）AbstractQueuedSynchronizer.enq()方法，初始化隊列，并將當前節點追加到隊列尾部，如果已經初始化完成則直接追加。
/**
* 初始化隊列，并將當前節點追加到隊列尾部，如果已經初始化完成則直接追加
*
* return 節點對象
*/
private Node enq(final Node node) {
//死循環，直到插入隊列成功跳出
for (;;) {
Node t = tail;
//判斷尾節點是否為空，如果為空則說明當前隊列未進行初始化，則需進行初始化操作
if (t == null) {
//新建一個空節點設置為頭節點
if (compareAndSetHead(new Node()))
//尾節點指向頭節點，此時尾節點與頭結點為同一個節點
tail = head;
} else {
//如果不為空則說明已經初始化完成，直接將當前節點插入尾部，構成雙向鏈表 node.prev = t;t.next = node;
node.prev = t;
//設置當前節點為尾節點
if (compareAndSetTail(t, node)) {
//設置原尾節點的下一個節點為當前節點
t.next = node;
return t;
}
}
}
}流程圖解：
9）AbstractQueuedSynchronizer.acquireQueued()方法，將線程加入到隊列尾部后，加入線程會不斷嘗試獲取鎖，直到獲取成功或者不再需要獲?。ㄖ袛啵?。
該方法的實現分成兩部分：
9.1）如果當前節點已經成為頭結點，嘗試獲取鎖（tryAcquire）成功，然后返回。
9.2）否則檢查當前節點是否應該被park（等待），將該線程park并且檢查當前線程是否被可以被中斷。
/**
* 不斷嘗試（自旋）進行獲取鎖，直到獲取成功或者不再需要獲取（中斷）
*
* return
*/
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)) {
//如果獲取鎖成功將當前節點設置為頭結點
setHead(node);
//將原頭節點的后繼指針設為null，去除強引用關系，幫助GC回收
p.next = null;
//標記獲取鎖成功 failed = false
failed = false;
//返回當前線程的中斷標記，是否需要喚醒當前線程
return interrupted;
}
//檢查當前節點是否應該被阻塞等待park
if (shouldParkAfterFailedAcquire(p, node) &&
//設置當前線程進入阻塞狀態
parkAndCheckInterrupt())
//標記當前線程的中斷狀態為中斷掛起狀態，線程再次執行需要被喚醒。
interrupted = true;
}
} finally {
if (failed)
//只有在出異常的情況下才會執行到這里，需要將當前節點取消掉
cancelAcquire(node);
}
}3.4.2 解鎖流程剖析
1）ReentrantLock.unlock()方法為ReentrantLock提供的解鎖方法。從實現可以看到該方法繼續調用了release()方法，而NonFairLock、FairLock和Sync類中均未重寫release()方法，所以此處是直接調用了AQS提供的release()方法來進行的解鎖操作。
public void unlock() {
sync.release(1);
}2）AbstractQueuedSynchronizer.release()方法，此方法主要做了兩個事情，首先是通過調用tryRelease()方法嘗試釋放鎖，如果釋放失敗直接返回失敗，如果鎖釋放成功則會去喚醒下個節點線程的執行。下面，我們繼續先看下tryRelease()方法的實現。
/**
* 鎖釋放 并喚醒下一個節點
*
* @param arg
* @return
*/
public final boolean release(int arg) {
// 1.嘗試釋放鎖
if (tryRelease(arg)) {
Node h = head;
//2.頭結點不為null并且等待狀態不是初始化狀態，因為處于初始化狀態的節點可能仍未初始化完成
if (h != null && h.waitStatus != 0)
//3.喚醒頭結點的下一個節點
unparkSuccessor(h);
return true;
}
//嘗試獲取鎖失敗
return false;
}3）AbstractQueuedSynchronizer.tryRelease()方法，可以看到此方法同AbstractQueuedSynchronizer.tryAcquiry()方法一樣，是由子類決定具體實現的，我們回看下ReentrantLock中定義的內部類，可以看到Sync類中重寫了該方法，而NonFairLock和FairLock方法中并未再次重寫該方法。所以在調用AQS中的tryRelease()方法時其實是調用的Sync類中的tryRelease()方法。
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}4）Sync.tryRelease()方法，該方法做的事其實跟簡單主要是將已執行完成的線程持有的鎖資源釋放掉。
/**
* 已執行完成的線程，釋放資源
*
* @param releases
* @return 返回釋放鎖后的已重入次數
*/
protected final boolean tryRelease(int releases) {
//獲取當前資源狀態值，并重新計算已重入次數
int c = getState() - releases;
//如果當前線程不是獲得資源的線程，將拋出異常
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
//資源是否完全釋放，因為涉及到可重入鎖
boolean free = false;
if (c == 0) {
//等于0的情況下表示資源完全釋放
free = true;
//清除鎖的持有線程標記
setExclusiveOwnerThread(null);
}
//重新設置已重入次數
setState(c);
return free;
}5）Sync.unparkSuccessor()方法，鎖釋放成功后會調用該方法，來喚醒當前節點的后續節點線程的執行。
/**
* 喚醒當前節點的后續節點
*
* @param node the node
*/
private void unparkSuccessor(Node node) {
//獲取頭結點的等待狀態
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
//獲取當前節點的下一個節點
Node s = node.next;
//如果當前節點的下一個節點是null或者狀態大于0，說明當前節點的下一個節點不是有效節點，那么則需要找到下一個有效的等待節點
if (s == null || s.waitStatus > 0) {
s = null;
//從尾節點開始向前找，找到最前面的狀態小于0的節點
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
if (s != null)
//喚醒讓當前節點的下一個節點線程，繼續執行
LockSupport.unpark(s.thread);
}4 總結
從實現來看，公平鎖的實現利用了FIFO隊列的特性，先加入同步隊列等待的線程會比后加入的線程更靠近隊列的頭部，那么它將比后者更早的被喚醒，它也就能更早的得到鎖。從這個意義上，對于在同步隊列中等待的線程而言，它們獲得鎖的順序和加入同步隊列的順序一致，這顯然是一種公平模式。然而，線程并非只有在加入隊列后才有機會獲得鎖，哪怕同步隊列中已有線程在等待，非公平鎖的不公平之處就在于此。回看下非公平鎖的加鎖流程，線程在進入同步隊列等待之前有兩次搶占鎖的機會。
- 第一次是非重入式的獲取鎖，只有在當前鎖未被任何線程占有(包括自身)時才能成功。
圖片
- 第二次是在進入同步隊列前，包含所有情況的獲取鎖的方式。
圖片
只有這兩次獲取鎖都失敗后，線程才會構造結點并加入到同步隊列等待，而線程釋放鎖時是先釋放鎖(修改state值)，然后才喚醒后繼結點的線程的。試想下這種情況，線程A已經釋放鎖，但還沒來得及喚醒后繼線程C，而這時另一個線程B剛好嘗試獲取鎖，此時鎖恰好不被任何線程持有，它將成功獲取鎖而不用加入隊列等待。線程C被喚醒嘗試獲取鎖，而此時鎖已經被線程B搶占，故而其獲取失敗并繼續在隊列中等待。如果以線程第一次嘗試獲取鎖到最后成功獲取鎖的次序來看，非公平鎖確實很不公平。因為在隊列中等待很久的線程相比于還未進入隊列等待的線程并沒有優先權，甚至競爭也處于劣勢，在隊列中的線程要等待其他線程喚醒，在獲取鎖之前還要檢查前驅結點是否為頭結點。在鎖競爭激烈的情況下，在隊列中等待的線程可能遲遲競爭不到鎖。這也就非公平在高并發情況下會出現的饑餓問題。
5 思考
5.1 為什么非公平鎖性能好
非公平鎖對鎖的競爭是搶占式的（隊列中線程除外），線程在進入等待隊列前可以進行兩次嘗試，這大大增加了獲取鎖的機會。這種好處體現在兩個方面：
1）線程不必加入等待隊列就可以獲得鎖，不僅免去了構造結點并加入隊列的繁瑣操作，同時也節省了線程阻塞喚醒的開銷，線程阻塞和喚醒涉及到線程上下文的切換和操作系統的系統調用，是非常耗時的。在高并發情況下，如果線程持有鎖的時間非常短，短到線程入隊阻塞的過程超過線程持有并釋放鎖的時間開銷，那么這種搶占式特性對并發性能的提升會更加明顯。
2）減少CAS競爭，如果線程必須要加入阻塞隊列才能獲取鎖，那入隊時CAS競爭將變得異常激烈，CAS操作雖然不會導致失敗線程掛起，但不斷失敗重試導致的對CPU的浪費也不能忽視。
5.2 公平鎖與非公平鎖的選擇
公平鎖的優點是等待鎖的線程不會餓死。缺點是整體吞吐效率相對非公平鎖要低，等待隊列中除第一個線程以外的所有線程都會阻塞，CPU喚醒阻塞線程的開銷比非公平鎖大。所以適用場景適用于對資源訪問順序有嚴格要求的場景。例如，在一些資源分配系統中，要求按照請求的先后順序來分配資源，以避免饑餓現象（某個線程一直無法獲取鎖）的發生。
非公平鎖的優點是可以減少喚起線程的開銷，整體的吞吐效率高，因為線程有幾率不阻塞直接獲得鎖，CPU不必喚醒所有線程。缺點是處于等待隊列中的線程可能會餓死，或者等很久才會獲得鎖。所以非公平鎖適用于如果對線程獲取鎖的順序沒有嚴格要求的場景，例如在一些高并發的緩存系統或者日志系統中，可以使用非公平鎖來提高系統的整體性能。
關于作者孔德志 采貨俠Java開發工程師
