本文分析的ReentrantLock所對應的Java版本為JDK8。

　　在閱讀本文前，讀者應該知道什么是CAS、自旋。

本文大綱

　　1.ReentrantLock公平鎖簡介
　　2.AQS
　　3.lock方法
　　4.unlock方法

1. ReentrantLock公平鎖簡介

　　ReentrantLock是JUC（java.util.concurrent）包中Lock接口的一個實現(xiàn)類，它是基于AbstractQueuedSynchronizer（下文簡稱AQS）來實現(xiàn)鎖的功能。ReentrantLock的內(nèi)部類Sync繼承了AbstractQueuedSynchronizer，Sync又有FairSync和NonFairSync兩個子類。FairSync實現(xiàn)了公平鎖相關(guān)的操作，NonFairSync實現(xiàn)了非公平鎖相關(guān)的操作。它們之間的關(guān)系如下：

?

　　公平鎖的公平之處主要體現(xiàn)在，對于一個新來的線程，如果鎖沒有被占用，它會判斷等待隊列中是否還有其它的等待線程，如果有的話，就加入等待隊列隊尾，否則就去搶占鎖。

　　下面這段代碼展示了公平鎖的使用方法：

private final Lock lock = new ReentrantLock(true); // 參數(shù)true代表創(chuàng)建公平鎖public void method() {lock.lock(); // block until condition holdstry {// ... method body} finally {lock.unlock();} }

2. AQS

　　下面簡單介紹一下AQS中的Node內(nèi)部類和幾個重要的成員變量。

2.1 Node

　　AQS中，維護了一個Node內(nèi)部類，用于包裝我們的線程。我們需要關(guān)注Node中的如下屬性：

• 　　pre：當前節(jié)點的前驅(qū)節(jié)點。
• 　　next：當前節(jié)點的后繼節(jié)點。
• 　　thread：thread表示被包裝的線程。
• 　　waitStatus：waitStatus是一個int整型，可以被賦予如下幾種值：

static final int CANCELLED = 1; // 線程被取消 static final int SIGNAL = -1; // 后繼節(jié)點中的線程需要被喚醒 static final int CONDITION = -2; // 暫不關(guān)注 static final int PROPAGATE = -3; // 暫不關(guān)注

　　另外，當一個新的Node被創(chuàng)建時，waitStatus是0。

2.2 head

　　head指向隊列中的隊首元素，可以理解為當前持有鎖的線程。

2.3 tail

　　tail指向隊列中的隊尾元素。

2.4 state

　　state表示在ReentrantLock中可以理解為鎖的狀態(tài)，0表示當前鎖沒有被占用，大于0的數(shù)表示鎖被當前線程重入的次數(shù)。例如，當state等于2時，表示當前線程在這把鎖上進入了兩次。

2.5 exclusiveOwnerThread

　　表示當前占用鎖的線程。

2.6 等待隊列

　　下圖簡單展示了AQS中的等待隊列：

3. lock方法

　　有了上面的AQS的基礎(chǔ)知識后，我們就可以展開對ReentrantLock公平鎖的分析了，先從lock方法入手。

　　ReentrantLock中的lock方法很簡單，只是調(diào)用了Sync類（本文研究公平鎖，所以應該是FairSync類）中的lock方法：

public void lock() {sync.lock(); }

　　我們跟到FairSync的lock方法，代碼也很簡單，調(diào)用了AQS中的acquire方法：

final void lock() {acquire(1); }

　　acquire方法：

public final void acquire(int arg) {if (!tryAcquire(arg) && // 調(diào)用tryAcquire嘗試去獲取鎖，如果獲取成功，則方法結(jié)束acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // 如果獲取鎖失敗，執(zhí)行acquireQueued方法，將把當前線程排入隊尾 selfInterrupt(); }

　　tryAcquire方法：

protected final boolean tryAcquire(int acquires) {final Thread current = Thread.currentThread();int c = getState(); // 獲取鎖的狀態(tài)if (c == 0) { // 如果狀態(tài)是0，表示鎖沒有被占用if (!hasQueuedPredecessors() && // 判斷是隊列中是否有排隊中的線程compareAndSetState(0, acquires)) { // 隊列中沒有排隊的線程，則嘗試用CAS去獲取一下鎖setExclusiveOwnerThread(current); // 獲取鎖成功，則將當前占有鎖的線程設(shè)置為當前線程return true;}}// 鎖被占用、隊列中有排隊的線程或者當前線程在獲取鎖的時候失敗將執(zhí)行下面的代碼else if (current == getExclusiveOwnerThread()) { // 當前線程是否是占有鎖的線程int nextc = c + acquires; // 是的話，表示當前線程是重入這把鎖，將鎖的狀態(tài)進行加1if (nextc < 0)throw new Error("Maximum lock count exceeded"); // 鎖的重入次數(shù)超過int能夠表示最大的值，拋出異常setState(nextc); // 設(shè)置鎖的狀態(tài)return true;}return false; // 沒有獲取到鎖 }

　　hasQueuedPredecessors方法：

public final boolean hasQueuedPredecessors() {Node t = tail;Node h = head;Node s;return h != t && // 隊列中的隊首和隊尾元素不相同((s = h.next) == null || s.thread != Thread.currentThread()); // 隊列中的第二個元素不為null，且第二個元素中的線程不是當前線程。這里如果返回true，說明隊列中至少存在tail、head兩個節(jié)點，就會執(zhí)行acquireQueued將當前線程加入隊尾 }

　　如果tryAcquire沒有獲取到鎖，將執(zhí)行：

acquireQueued(addWaiter(Node.EXCLUSIVE), arg)

　　我們先分析addWaiter方法：

private Node addWaiter(Node mode) {Node node = new Node(Thread.currentThread(), mode); // 將當前線程包裝成Node，mode參數(shù)值為null，表示獨占模式// Try the fast path of enq; backup to full enq on failureNode pred = tail;if (pred != null) {node.prev = pred; // 如果隊列中的尾節(jié)點不為空，將當前node的前驅(qū)節(jié)點設(shè)置為之前隊列中的tailif (compareAndSetTail(pred, node)) { // 用CAS把當前node設(shè)置為隊尾元素pred.next = node; // 成功的話，則將之前隊尾元素的后繼節(jié)點設(shè)置為當前節(jié)點。如果這里不清楚的話，請結(jié)合前面講等待隊列的那張圖進行理解。return node;}}enq(node); // 隊尾節(jié)點為空，或者用CAS設(shè)置隊尾元素失敗，則用自旋的方式入隊return node; }

　　enq方法：

private Node enq(final Node node) {for (;;) {Node t = tail;if (t == null) {if (compareAndSetHead(new Node())) // 隊尾元素為空，創(chuàng)建一個空的Node，并設(shè)置為隊首tail = head; // 設(shè)置隊首和隊尾為同一個空Node，進入下一次循環(huán)} else {node.prev = t; // 如果隊列中的尾節(jié)點不為空，將當前node的前驅(qū)節(jié)點設(shè)置為之前隊列中的tailif (compareAndSetTail(t, node)) { // 用CAS把當前node設(shè)置為隊尾元素t.next = node; // 成功的話，則將之前隊尾元素的后繼節(jié)點設(shè)置為當前節(jié)點return t; }}} }

?　　下面這張圖反應了上面enq方法的處理流程：

　　經(jīng)過上面的方法，當前node已經(jīng)加入等待隊列的隊尾，接下來將執(zhí)行acquireQueued方法：

final boolean acquireQueued(final Node node, int arg) {boolean failed = true;try {boolean interrupted = false;for (;;) {final Node p = node.predecessor(); // 獲取node的前驅(qū)節(jié)點if (p == head && tryAcquire(arg)) { // 如果node的前驅(qū)是head，它將去嘗試獲取鎖（tryAcquire方法在前面已經(jīng)分析過）setHead(node); // 獲取成功，則將node設(shè)置為headp.next = null; // 將之前的head的后繼節(jié)點置空failed = false;return interrupted;}if (shouldParkAfterFailedAcquire(p, node) && // 當前node的前驅(qū)不是head，將為當前node找到一個能夠?qū)⑵鋯拘训那膀?qū)節(jié)點；或者當前node的前驅(qū)是head，但是獲取鎖失敗parkAndCheckInterrupt()) // 將當前線程掛起interrupted = true;}} finally {if (failed)cancelAcquire(node);} }

　　shouldParkAfterFailedAcquire方法的作用就是找到一個能夠喚醒當前node的節(jié)點：

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {int ws = pred.waitStatus; // 開始時是0if (ws == Node.SIGNAL)/** This node has already set status asking a release* to signal it, so it can safely park.*/return true; // 前驅(qū)節(jié)點的狀態(tài)是-1，會喚醒后繼節(jié)點，可以將線程掛起if (ws > 0) {/** Predecessor was cancelled. Skip over predecessors and* indicate retry.*/do {node.prev = pred = pred.prev; // 前驅(qū)節(jié)點中的線程被取消，那就需要一直循環(huán)直到找到一個沒有被設(shè)置為取消狀態(tài)的前驅(qū)節(jié)點} while (pred.waitStatus > 0);pred.next = node; // 從后向前找，將第一個非取消狀態(tài)的節(jié)點，設(shè)置這個節(jié)點的后繼節(jié)點設(shè)置為當前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); // waitStatus是0或者-3的時候，這時waitStatus都將被設(shè)置為-1// 即后繼節(jié)點需要前驅(qū)節(jié)點喚醒 }return false; // 上層代碼再進行一次循環(huán)，下次進入此方法時，將進入第一個if條件 }

　　找到了合適的前驅(qū)節(jié)點，parkAndCheckInterrupt方法當前線程掛起：

private final boolean parkAndCheckInterrupt() { // 將線程掛起，等待前驅(qū)節(jié)點的喚醒LockSupport.park(this);return Thread.interrupted(); }

?4. unlock方法

　　ReentrantLock的unlock方法調(diào)用AQS中的release方法：

public void unlock() {sync.release(1); // 調(diào)用AQS的release方法 }

　　release方法：

public final boolean release(int arg) {if (tryRelease(arg)) { // 嘗試去釋放鎖Node h = head;if (h != null && h.waitStatus != 0)unparkSuccessor(h); // 釋放鎖成功，head不為空，并且head的waitStatus不為0的情況下，將喚醒后繼節(jié)點return true;}return false; }

　　tryRelease方法：

protected final boolean tryRelease(int releases) {int c = getState() - releases; // 將鎖的狀態(tài)減1if (Thread.currentThread() != getExclusiveOwnerThread())throw new IllegalMonitorStateException(); // 準備釋放鎖的線程不是持有鎖的線程，拋出異常boolean free = false;if (c == 0) {free = true; // 鎖的狀態(tài)是0，說明不存在重入的情況了，可以直接釋放了setExclusiveOwnerThread(null);}setState(c);return free; }

　　鎖釋放成功，將喚醒后繼節(jié)點，unparkSuccessor方法：

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; // 注意，這個node是head節(jié)點if (ws < 0)compareAndSetWaitStatus(node, ws, 0); // 當前node的狀態(tài)是小于0，將其狀態(tài)設(shè)置為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; // head節(jié)點的后繼節(jié)點if (s == null || s.waitStatus > 0) {s = null; // 執(zhí)行到這表示head的后繼節(jié)點是1，處于取消的狀態(tài)for (Node t = tail; t != null && t != node; t = t.prev)if (t.waitStatus <= 0)s = t; // 從等待隊列的隊尾向前找，找到倒序的最后一個處于非取消狀態(tài)的節(jié)點 }if (s != null)LockSupport.unpark(s.thread); // 喚醒head后面的處于非取消狀態(tài)的第一個（正序）節(jié)點 }

　　到此，全文結(jié)束，大家看代碼的時候結(jié)合圖來理解會容易很多。

轉(zhuǎn)載于:https://www.cnblogs.com/pedlar/p/10731516.html

總結(jié)

以上是生活随笔為你收集整理的ReentrantLock之公平锁源码分析的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。