怎么理解ReentrantLock

這篇文章主要介紹“怎么理解ReentrantLock”，在日常操作中，相信很多人在怎么理解ReentrantLock問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”怎么理解ReentrantLock”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

 ReentrantLock 是可重入鎖

啥是可重入鎖呢?比如：線程 1 通過調用 lock() 方法獲取鎖之后，再調用 lock 時，就不會再進行阻塞獲取鎖，而是直接增加重試次數。

還記得 synchronized 嗎?它有 monitorenter 和 monitorexit  兩種指令來保證鎖，而它們的作用可以理解為每個鎖對象擁有一個鎖計數器，也就是如果再次調用 lock() 方法，計數器會進行加 1 操作

所以， synchronized 和 ReentrantLock 都是可重入鎖

ReentrantLock 與 synchronized 區別

既然 synchronized 和 ReentrantLock 都是可重入鎖，那 ReentrantLock 與 synchronized  有什么區別呢?

synchronized 是 Java 語言層面提供的語法，所以不需要考慮異常;ReentrantLock 是 Java  代碼實現的鎖，所以必須先要獲取鎖，然后再正確釋放鎖

synchronized 在獲取鎖時必須一直等待沒有額外的嘗試機制;ReentrantLock 可以嘗試獲取鎖(這一點等下分析源碼時會看到)

ReentrantLock 支持獲取鎖時的公平和非公平選擇

不 BB 了，直接上源碼

lock & NonfairSync & FairSync 詳解

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

其中， sync 是 ReentrantLock 的靜態內部類，它繼承 AQS 來實現重入鎖的邏輯， Sync 有兩個具體實現類: NonfairSync  和 FairSync

NonfairSync

先來看一下 NonfairSync :

static final class NonfairSync extends Sync {     private static final long serialVersionUID = 7316153563782823691L;      /**     * Performs lock.  Try immediate barge, backing up to normal     * acquire on failure.     */     // 重寫 Sync 的 lock 方法     final void lock() {      // 先不管其他,上來就先 CAS 操作,嘗試搶占一下鎖         if (compareAndSetState(0, 1))          // 如果搶占成功,就獲得了鎖             setExclusiveOwnerThread(Thread.currentThread());         else          // 沒有搶占成功,調用 acquire() 方法,走里面的邏輯             acquire(1);     }  // 重寫了 AQS 的 tryAcquire 方法     protected final boolean tryAcquire(int acquires) {         return nonfairTryAcquire(acquires);     } }

FairSync

接下來看一下 FairSync :

static final class FairSync extends Sync {     private static final long serialVersionUID = -3000897897090466540L;   // 重寫 Sync 的 lock 方法     final void lock() {         acquire(1);     }      /**     * Fair version of tryAcquire.  Don't grant access unless     * recursive call or no waiters or is first.     */     // 重寫了 Sync 的 tryAcquire 方法     protected final boolean tryAcquire(int acquires) {      // 獲取當前執行的線程         final Thread current = Thread.currentThread();         // 獲取 state 的值         int c = getState();         // 在無鎖狀態下         if (c == 0) {          // 沒有前驅節點且替換 state 的值成功時             if (!hasQueuedPredecessors() &&                 compareAndSetState(0, acquires)) {                 // 保存當前獲得鎖的線程,下次再來時,就不需要嘗試競爭鎖,直接重入即可                 setExclusiveOwnerThread(current);                 return true;             }         }         else if (current == getExclusiveOwnerThread()) {          // 如果是同一個線程來獲得鎖,直接增加重入次數即可             int nextc = c + acquires;             // nextc 小于 0 ,拋異常             if (nextc < 0)                 throw new Error("Maximum lock count exceeded");             setState(nextc);             // 獲取鎖成功             return true;         }         // 獲取鎖失敗         return false;     } }

總結 NonfairSync 與 FairSync

到這里，應該就比較清楚了， Sync 有兩個具體的實現類，分別是:

• NonfairSync :可以搶占鎖，調用 NonfairSync 時，不管當前隊列上有沒有其他線程在等待，上來我就先 CAS  操作一番，成功了就獲得了鎖，沒有成功就走 acquire 的邏輯;在釋放鎖資源時，走的是 Sync.nonfairTryAcquire 方法

• FairSync :所有線程按照 FIFO 來獲取鎖，在 lock 方法中，沒有 CAS 嘗試，直接就是 acquire  的邏輯;在釋放資源時，走的是自己的 tryAcquire 邏輯

接下來咱們看看 NonfairSync 和 FairSync 是如何獲取鎖的

ReentrantLock 獲取鎖

NonfairSync.lock()

在 NonfairSync 中，獲取鎖的方法是:

final void lock() {  // 不管別的,上來就先 CAS 操作,嘗試搶占一下鎖     if (compareAndSetState(0, 1))      // 如果搶占成功,就獲得了鎖         setExclusiveOwnerThread(Thread.currentThread());     else      // 沒有搶占成功,調用 acquire() 方法,走里面的邏輯         acquire(1); }

if 里面沒啥說的，咱們來看看 acquire() 方法

AQS.acquire()

acquire 是 AQS 的核心方法：

public final void acquire(int arg) {     if (!tryAcquire(arg) &&         acquireQueued(addWaiter(Node.EXCLUSIVE), arg))         selfInterrupt(); }

在這里，會先 tryAcquire 去嘗試獲取鎖，如果獲取成功，那就返回 true ，如果失敗就通過 addWaiter 方法，將當前線程封裝成 Node  插入到等待隊列中

先來看 tryAcquire 方法:

NonfairSync.tryAcquire(arg)

在 AQS 中 tryAcquire 方法沒有具體實現，只是拋出了異常:

protected boolean tryAcquire(int arg) {     throw new UnsupportedOperationException(); }

NonfairSync 中的 tryAcquire() 方法，才是我們想要看的:

final boolean nonfairTryAcquire(int acquires) {  // 獲取當前執行的線程     final Thread current = Thread.currentThread();     // 獲取 state 的值     int c = getState();     // 當 state 為 0 是,說明此時為無鎖狀態     if (c == 0) {      // CAS 替換 state 的值,如果 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; }

有沒有一種似曾相識的趕腳?在 FairSync 那里，分析過 90% 的代碼(好像說分析過 99% 的代碼也不過分)，只是 FairSync  多了一個判斷就是，是否有前驅節點

tryAcquire 分析完畢了，接下來看 addWaiter 方法

AQS.addWaiter

如果 tryAcquire() 方法獲取鎖成功，那就直接執行線程的任務就可以了，執行完畢釋放鎖

如果獲取鎖失敗，就會調用 addWaiter 方法，將當前線程插入到等待隊列中，插入的邏輯大概是這樣的:

• 將當前線程封裝成 Node 節點

• 當前鏈表中 tail 節點(也就是下面的 pred )是否為空，如果不為空，則 CAS 操作將當前線程的 node 添加到 AQS 隊列

• 如果為空，或者 CAS 操作失敗，則調用 enq 方法，再次自旋插入

咱們看具體的代碼實現:

private Node addWaiter(Node mode) {  // 生成該線程所對應的 Node 節點     Node node = new Node(Thread.currentThread(), mode);     // 將 Node 插入隊列中     Node pred = tail;     // 如果 pred 不為空     if (pred != null) {         node.prev = pred;         // 使用 CAS 操作,如果成功就返回         if (compareAndSetTail(pred, node)) {             pred.next = node;             return node;         }     }     // 如果 pred == null 或者 CAS 操作失敗,則調用 enq 方法再次自旋插入     enq(node);     return node; }  // 自旋 CAS 插入等待隊列 private Node enq(final Node node) {     for (;;) {         Node t = tail;         if (t == null) { // Must initialize          // 必須初始化,使用 CAS 操作進行初始化             if (compareAndSetHead(new Node()))              // 初始化狀態時,頭尾節點指向同一節點                 tail = head;         } else {             node.prev = t;             // 如果剛開始就是初始化好的,直接 CAS 操作,將 Node 插入到隊尾即可             if (compareAndSetTail(t, node)) {                 t.next = node;                 return t;             }         }     } }

AQS.acquireQueued

通過 addWaiter 將當前線程加入到隊列中之后，會走 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)  方法

acquireQueued 方法實現的主要邏輯是:

• 獲取當前節點的前驅節點 p

• 如果節點 p 為 head 節點，說明當前節點為第二個節點，那么它就可以嘗試獲取鎖，調用 tryAcquire 方法嘗試進行獲取

• 調用 tryAcquire 方法獲取鎖成功之后，就將 head 指向自己，原來的節點 p 就需要從隊列中刪除

• 如果獲取鎖失敗，則調用 shouldParkAfterFailedAcquire 或者 parkAndCheckInterrupt  方法來決定后面操作

最后，通過 cancelAcquire 方法取消獲得鎖 看具體的代碼實現:

final boolean acquireQueued(final Node node, int arg) {     boolean failed = true;     try {         boolean interrupted = false;         for (;;) {             final Node p = node.predecessor();             // 如果 Node 的前驅節點 p 是 head,說明 Node 是第二個節點,那么它就可以嘗試獲取鎖             if (p == head && tryAcquire(arg)) {              // 如果鎖獲取成功,則將 head 指向自己                 setHead(node);                 // 鎖獲取成功之后,將 next 指向 null ,即將節點 p 從隊列中移除                 p.next = null; // help GC                 failed = false;                 return interrupted;             }             // 節點進入等待隊列后,調用 shouldParkAfterFailedAcquire 或者 parkAndCheckInterrupt 方法             // 進入阻塞狀態,即只有頭結點的線程處于活躍狀態             if (shouldParkAfterFailedAcquire(p, node) &&                 parkAndCheckInterrupt())                 interrupted = true;         }     } finally {         if (failed)             cancelAcquire(node);     } }

shouldParkAfterFailedAcquire

線程獲取鎖失敗之后，會通過調用 shouldParkAfterFailedAcquire 方法，來決定這個線程要不要掛起

shouldParkAfterFailedAcquire 方法實現的主要邏輯:

• 首先判斷 pred 的狀態是否為 SIGNAL ，如果是，則直接掛起即可

• 如果 pred 的狀態大于 0 ，說明該節點被取消了，那么直接從隊列中移除即可

• 如果 pred 的狀態不是 SIGNAL 也不大于 0 ，進行 CAS 操作修改節點狀態為 SIGNAL ，返回 false ，也就是不需要掛起

看一下代碼是如何實現的:

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {  // 獲取 pred 的狀態     int ws = pred.waitStatus;     // 如果狀態為 SIGNAL ,那么直接返回 true ,掛起線程即可     if (ws == Node.SIGNAL)         return true;     // 如果狀態大于 0 ,說明線程被取消     if (ws > 0) {      // 從鏈表中移除被 cancel 的線程,使用循環來保證移除成功         do {             node.prev = pred = pred.prev;         } while (pred.waitStatus > 0);         pred.next = node;     } else {      // CAS 操作修改 pred 節點狀態為 SIGNAL         compareAndSetWaitStatus(pred, ws, Node.SIGNAL);     }     // 不需要掛起線程     return false; }

到這里，關于 NonfairSync 的獲取鎖就結束了

接下來咱們看看 FairSync 的獲取鎖和它有什么不同

FairSync.lock()

在 FairSync.lock() 方法中是這樣的:

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

因為 FairSync 是公平鎖，所以不存在 CAS 操作去競爭，直接就是調用 acquire 方法

接下來的邏輯就和上面一樣了，這里我就不重復了

咱們瞅瞅 ReentrantLock 是怎么釋放鎖的

ReentrantLock 釋放鎖

在 ReentrantLock 釋放鎖時，調用的是 sync.release() 方法:

public void unlock() {     sync.release(1); }

點進去發現調用的是 AQS 的 release 方法

AQS.release()

AQS 的 release 方法比較好理解，就直接看源碼了:

public final boolean release(int arg) {  // 如果釋放鎖成功     if (tryRelease(arg)) {      // 獲取 AQS 隊列的頭結點         Node h = head;         // 如果頭結點不為空,且狀態 != 0         if (h != null && h.waitStatus != 0)          // 調用 unparkSuccessor 方法喚醒后續節點             unparkSuccessor(h);         return true;     }     return false; }

ReentrantLock.tryRelease()

在 AQS 中的 tryRelease 方法，只是拋出了異常而已，說明具體實現是由子類 ReentrantLock 來實現的

就直接看 ReentrantLock 中的 tryRelease 方法了

在 ReentrantLock 中實現 tryRelease 方法主要邏輯是:

• 首先，如果是同一個線程獲取的同一個鎖，那么它有可能被重入多次，所以需要獲取到要釋放線程的重入次數即 getState()  然后判斷，該線程是否為獲取到鎖的線程，只有獲取到鎖的線程，才有釋放鎖一說

• 進行 unlock 釋放鎖，即:將 state 的值減到 0 ，才算是釋放掉了鎖，此時才能將 owner 置為 null 同時返回 true

看一下具體實現:

protected final boolean tryRelease(int releases) {     int c = getState() - releases;     // 判斷當前線程是否為獲取到鎖的線程,如果不是則拋出異常     // 只有獲取到鎖的線程才釋放鎖     if (Thread.currentThread() != getExclusiveOwnerThread())         throw new IllegalMonitorStateException();     boolean free = false;     // 次數為 0 ,說釋放鎖完畢     if (c == 0) {         free = true;         // 釋放之后,當前線程置為 null         setExclusiveOwnerThread(null);     }     // 更新重入次數     setState(c);     return free; }

AQS.unparkSuccessor

釋放鎖成功之后，接下來要做的就是喚醒后面的進程，這個方法是在 AQS 中實現的

主要邏輯是:

• 獲取當前節點狀態，如果小于 0 ，則置為 0

• 獲取當前節點的下一個節點，如果不為空，直接喚醒

• 如果為空，或者節點狀態大于 0 ，則尋找下一個狀態小于 0 的節點

代碼的具體實現

private void unparkSuccessor(Node node) {  // 獲取當前節點的狀態     int ws = node.waitStatus;     // 如果節點狀態小于 0 ,則進行 CAS 操作設置為 0     if (ws < 0)         compareAndSetWaitStatus(node, ws, 0);     // 獲取當前節點的下一個節點 s     Node s = node.next;     // 如果 s 為空,則從尾部節點開始,或者s.waitStatus 大于 0 ,說明節點被取消     // 從尾節點開始,尋找到距離 head 節點最近的一個 waitStatus <= 0 的節點     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)      // next 節點不為空,直接喚醒即可         LockSupport.unpark(s.thread); }

為什么要從尾節點開始尋找距離 head 節點最近的一個 waitStatus <= 0 的節點呢?

這是因為在 enq() 構建節點的方法中，最后是 t.next = node (忘了就再往上翻翻看)，設置原來的 tail 的 next  節點指向新的節點

如果在 CAS 操作之后， t.next = node 操作之前，有其他線程調用 unlock 方法從 head 開始向后遍歷，因為此時 t.next =  node 還沒有執行結束，意味著鏈表的關系還沒有建立好，這樣就會導致遍歷的時候到 t 節點這里發生中斷，因為此時 tail 還沒有指向新的尾節點

如果從后向前遍歷的話，就不會存在這樣的問題

接下來下一個線程就被喚醒了，然后程序會把它當成新的節點開始執行

而原來執行結束的線程，則會將它從隊列中移除，然后開始循環循環

到此，關于“怎么理解ReentrantLock”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續學習更多相關知識，請繼續關注億速云網站，小編會繼續努力為大家帶來更多實用的文章！