問題

（1）重入鎖是什麼？

（2）ReentrantLock如何實現重入鎖？

（3）ReentrantLock為什麼預設是非公平模式？

（4）ReentrantLock除了可重入還有哪些特性？

簡介

Reentrant = Re + entrant，Re是重複、又、再的意思，entrant是enter的名詞或者形容詞形式，翻譯為進入者或者可進入的，所以Reentrant翻譯為可重複進入的、可再次進入的，因此ReentrantLock翻譯為重入鎖或者再入鎖。

重入鎖，是指一個執行緒獲取鎖之後再嘗試獲取鎖時會自動獲取鎖。

在Java中，除了ReentrantLock以外，synchronized也是重入鎖。

那麼，ReentrantLock的可重入性是怎麼實現的呢？

繼承體系

ReentrantLock實現了Lock介面，Lock接口裡面定義了java中鎖應該實現的幾個方法：

// 獲取鎖
void lock();
// 獲取鎖（可中斷）
void lockInterruptibly() throws InterruptedException;
// 嘗試獲取鎖，如果沒獲取到鎖，就返回false
boolean tryLock();
// 嘗試獲取鎖，如果沒獲取到鎖，就等待一段時間，這段時間內還沒獲取到鎖就返回false
boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
// 釋放鎖
void unlock();
// 條件鎖
Condition newCondition();
 

Lock介面中主要定義了 獲取鎖、嘗試獲取鎖、釋放鎖、條件鎖等幾個方法。

原始碼分析

主要內部類

ReentrantLock中主要定義了三個內部類：Sync、NonfairSync、FairSync。

abstract static class Sync extends AbstractQueuedSynchronizer {}

static final class NonfairSync extends Sync {}

static final class FairSync extends Sync {}

（1）抽象類Sync實現了AQS的部分方法；

（2）NonfairSync實現了Sync，主要用於非公平鎖的獲取；

（3）FairSync實現了Sync，主要用於公平鎖的獲取。

在這裡我們先不急著看每個類具體的程式碼，等下面學習具體的功能點的時候再把所有方法串起來。

主要屬性

private final Sync sync;

主要屬性就一個sync，它在構造方法中初始化，決定使用公平鎖還是非公平鎖的方式獲取鎖。

主要構造方法

// 預設構造方法
public ReentrantLock() {
    sync = new NonfairSync();
}
// 自己可選擇使用公平鎖還是非公平鎖
public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}

（1）預設構造方法使用的是非公平鎖；

（2）第二個構造方法可以自己決定使用公平鎖還是非公平鎖；

上面我們分析了ReentrantLock的主要結構，下面我們跟著幾個主要方法來看原始碼。

lock()方法

彤哥貼心地在每個方法的註釋都加上方法的來源。

公平鎖

這裡我們假設ReentrantLock的例項是通過以下方式獲得的：

ReentrantLock reentrantLock = new ReentrantLock(true);

下面的是加鎖的主要邏輯：

// ReentrantLock.lock()
public void lock() {
    // 呼叫的sync屬性的lock()方法
    // 這裡的sync是公平鎖，所以是FairSync的例項
    sync.lock();
}
// ReentrantLock.FairSync.lock()
final void lock() {
    // 呼叫AQS的acquire()方法獲取鎖
    // 注意，這裡傳的值為1
    acquire(1);
}
// AbstractQueuedSynchronizer.acquire()
public final void acquire(int arg) {
    // 嘗試獲取鎖
    // 如果失敗了，就排隊
    if (!tryAcquire(arg) &&
        // 注意addWaiter()這裡傳入的節點模式為獨佔模式
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}
// ReentrantLock.FairSync.tryAcquire()
protected final boolean tryAcquire(int acquires) {
    // 當前執行緒
    final Thread current = Thread.currentThread();
    // 檢視當前狀態變數的值
    int c = getState();
    // 如果狀態變數的值為0，說明暫時還沒有人佔有鎖
    if (c == 0) {
        // 如果沒有其它執行緒在排隊，那麼當前執行緒嘗試更新state的值為1
        // 如果成功了，則說明當前執行緒獲取了鎖
        if (!hasQueuedPredecessors() &&
            compareAndSetState(0, acquires)) {
            // 當前執行緒獲取了鎖，把自己設定到exclusiveOwnerThread變數中
            // exclusiveOwnerThread是AQS的父類AbstractOwnableSynchronizer中提供的變數
            setExclusiveOwnerThread(current);
            // 返回true說明成功獲取了鎖
            return true;
        }
    }
    // 如果當前執行緒本身就佔有著鎖，現在又嘗試獲取鎖
    // 那麼，直接讓它獲取鎖並返回true
    else if (current == getExclusiveOwnerThread()) {
        // 狀態變數state的值加1
        int nextc = c + acquires;
        // 如果溢位了，則報錯
        if (nextc < 0)
            throw new Error("Maximum lock count exceeded");
        // 設定到state中
        // 這裡不需要CAS更新state
        // 因為當前執行緒佔有著鎖，其它執行緒只會CAS把state從0更新成1，是不會成功的
        // 所以不存在競爭，自然不需要使用CAS來更新
        setState(nextc);
        // 當執行緒獲取鎖成功
        return true;
    }
    // 當前執行緒嘗試獲取鎖失敗
    return false;
}
// AbstractQueuedSynchronizer.addWaiter()
// 呼叫這個方法，說明上面嘗試獲取鎖失敗了
private Node addWaiter(Node mode) {
    // 新建一個節點
    Node node = new Node(Thread.currentThread(), mode);
    // 這裡先嚐試把新節點加到尾節點後面
    // 如果成功了就返回新節點
    // 如果沒成功再呼叫enq()方法不斷嘗試
    Node pred = tail;
    // 如果尾節點不為空
    if (pred != null) {
        // 設定新節點的前置節點為現在的尾節點
        node.prev = pred;
        // CAS更新尾節點為新節點
        if (compareAndSetTail(pred, node)) {
            // 如果成功了，把舊尾節點的下一個節點指向新節點
            pred.next = node;
            // 並返回新節點
            return node;
        }
    }
    // 如果上面嘗試入隊新節點沒成功，呼叫enq()處理
    enq(node);
    return node;
}
// AbstractQueuedSynchronizer.enq()
private Node enq(final Node node) {
    // 自旋，不斷嘗試
    for (;;) {
        Node t = tail;
        // 如果尾節點為空，說明還未初始化
        if (t == null) { // Must initialize
            // 初始化頭節點和尾節點
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            // 如果尾節點不為空
            // 設定新節點的前一個節點為現在的尾節點
            node.prev = t;
            // CAS更新尾節點為新節點
            if (compareAndSetTail(t, node)) {
                // 成功了，則設定舊尾節點的下一個節點為新節點
                t.next = node;
                // 並返回舊尾節點
                return t;
            }
        }
    }
}
// AbstractQueuedSynchronizer.acquireQueued()
// 呼叫上面的addWaiter()方法使得新節點已經成功入隊了
// 這個方法是嘗試讓當前節點來獲取鎖的
final boolean acquireQueued(final Node node, int arg) {
    // 失敗標記
    boolean failed = true;
    try {
        // 中斷標記
        boolean interrupted = false;
        // 自旋
        for (;;) {
            // 當前節點的前一個節點
            final Node p = node.predecessor();
            // 如果當前節點的前一個節點為head節點，則說明輪到自己獲取鎖了
            // 呼叫ReentrantLock.FairSync.tryAcquire()方法再次嘗試獲取鎖
            if (p == head && tryAcquire(arg)) {
                // 嘗試獲取鎖成功
                // 這裡同時只會有一個執行緒在執行，所以不需要用CAS更新
                // 把當前節點設定為新的頭節點
                setHead(node);
                // 並把上一個節點從連結串列中刪除
                p.next = null; // help GC
                // 未失敗
                failed = false;
                return interrupted;
            }
            // 是否需要阻塞
            if (shouldParkAfterFailedAcquire(p, node) &&
                // 真正阻塞的方法
                parkAndCheckInterrupt())
                // 如果中斷了
                interrupted = true;
        }
    } finally {
        // 如果失敗了
        if (failed)
            // 取消獲取鎖
            cancelAcquire(node);
    }
}
// AbstractQueuedSynchronizer.shouldParkAfterFailedAcquire()
// 這個方法是在上面的for()迴圈裡面呼叫的
// 第一次呼叫會把前一個節點的等待狀態設定為SIGNAL，並返回false
// 第二次呼叫才會返回true
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    // 上一個節點的等待狀態
    // 注意Node的waitStatus欄位我們在上面建立Node的時候並沒有指定
    // 也就是說使用的是預設值0
    // 這裡把各種等待狀態再貼出來
    //static final int CANCELLED =  1;
    //static final int SIGNAL    = -1;
    //static final int CONDITION = -2;
    //static final int PROPAGATE = -3;
    int ws = pred.waitStatus;
    // 如果等待狀態為SIGNAL(等待喚醒)，直接返回true
    if (ws == Node.SIGNAL)
        return true;
    // 如果前一個節點的狀態大於0，也就是已取消狀態
    if (ws > 0) {
        // 把前面所有取消狀態的節點都從連結串列中刪除
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        // 如果前一個節點的狀態小於等於0，則把其狀態設定為等待喚醒
        // 這裡可以簡單地理解為把初始狀態0設定為SIGNAL
        // CONDITION是條件鎖的時候使用的
        // PROPAGATE是共享鎖使用的
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}
// AbstractQueuedSynchronizer.parkAndCheckInterrupt()
private final boolean parkAndCheckInterrupt() {
    // 阻塞當前執行緒
    // 底層呼叫的是Unsafe的park()方法
    LockSupport.park(this);
    // 返回是否已中斷
    return Thread.interrupted();
}

看過之前彤哥寫的【死磕 java同步系列之自己動手寫一個鎖Lock】的同學看今天這個加鎖過程應該思路會比較清晰。

下面我們看一下主要方法的呼叫關係，可以跟著我的 → 層級在腦海中大概過一遍每個方法的主要程式碼：

ReentrantLock#lock()
->ReentrantLock.FairSync#lock() // 公平模式獲取鎖
  ->AbstractQueuedSynchronizer#acquire() // AQS的獲取鎖方法
    ->ReentrantLock.FairSync#tryAcquire() // 嘗試獲取鎖
    ->AbstractQueuedSynchronizer#addWaiter()  // 新增到佇列
      ->AbstractQueuedSynchronizer#enq()  // 入隊
    ->AbstractQueuedSynchronizer#acquireQueued() // 裡面有個for()迴圈，喚醒後再次嘗試獲取鎖
      ->AbstractQueuedSynchronizer#shouldParkAfterFailedAcquire() // 檢查是否要阻塞
      ->AbstractQueuedSynchronizer#parkAndCheckInterrupt()  // 真正阻塞的地方

獲取鎖的主要過程大致如下：

（1）嘗試獲取鎖，如果獲取到了就直接返回了；

（2）嘗試獲取鎖失敗，再呼叫addWaiter()構建新節點並把新節點入隊；

（3）然後呼叫acquireQueued()再次嘗試獲取鎖，如果成功了，直接返回；

（4）如果再次失敗，再呼叫shouldParkAfterFailedAcquire()將節點的等待狀態置為等待喚醒（SIGNAL）；

（5）呼叫parkAndCheckInterrupt()阻塞當前執行緒；

（6）如果被喚醒了，會繼續在acquireQueued()的for()迴圈再次嘗試獲取鎖，如果成功了就返回；

（7）如果不成功，再次阻塞，重複（3）（4）（5）直到成功獲取到鎖。

以上就是整個公平鎖獲取鎖的過程，下面我們看看非公平鎖是怎麼獲取鎖的。

非公平鎖

// ReentrantLock.lock()
public void lock() {
    sync.lock();
}
// ReentrantLock.NonfairSync.lock()
// 這個方法在公平鎖模式下是直接呼叫的acquire(1);
final void lock() {
    // 直接嘗試CAS更新狀態變數
    if (compareAndSetState(0, 1))
        // 如果更新成功，說明獲取到鎖，把當前執行緒設為獨佔執行緒
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}
// ReentrantLock.NonfairSync.tryAcquire()
protected final boolean tryAcquire(int acquires) {
    // 呼叫父類的方法
    return nonfairTryAcquire(acquires);
}
// ReentrantLock.Sync.nonfairTryAcquire()
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        // 如果狀態變數的值為0，再次嘗試CAS更新狀態變數的值
        // 相對於公平鎖模式少了!hasQueuedPredecessors()條件
        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;
}

相對於公平鎖，非公平鎖加鎖的過程主要有兩點不同：

（1）一開始就嘗試CAS更新狀態變數state的值，如果成功了就獲取到鎖了；

（2）在tryAcquire()的時候沒有檢查是否前面有排隊的執行緒，直接上去獲取鎖才不管別人有沒有排隊呢；

總的來說，相對於公平鎖，非公平鎖在一開始就多了兩次直接嘗試獲取鎖的過程。

lockInterruptibly()方法

支援執行緒中斷，它與lock()方法的主要區別在於lockInterruptibly()獲取鎖的時候如果執行緒中斷了，會丟擲一個異常，而lock()不會管執行緒是否中斷都會一直嘗試獲取鎖，獲取鎖之後把自己標記為已中斷，繼續執行自己的邏輯，後面也會正常釋放鎖。

題外話：

執行緒中斷，只是在執行緒上打一箇中斷標誌，並不會對執行中的執行緒有什麼影響，具體需要根據這個中斷標誌幹些什麼，使用者自己去決定。

比如，如果使用者在呼叫lock()獲取鎖後，發現執行緒中斷了，就直接返回了，而導致沒有釋放鎖，這也是允許的，但是會導致這個鎖一直得不到釋放，就出現了死鎖。

lock.lock();

if (Thread.currentThread().interrupted()) {
    return ;
}

lock.unlock();

當然，這裡只是舉個例子，實際使用肯定是要把lock.lock()後面的程式碼都放在try...finally...裡面的以保證鎖始終會釋放，這裡主要是為了說明執行緒中斷只是一個標誌，至於要做什麼完全由使用者自己決定。

tryLock()方法

嘗試獲取一次鎖，成功了就返回true，沒成功就返回false，不會繼續嘗試。

// ReentrantLock.tryLock()
public boolean tryLock() {
    // 直接呼叫Sync的nonfairTryAcquire()方法
    return sync.nonfairTryAcquire(1);
}
// ReentrantLock.Sync.nonfairTryAcquire()
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;
}

tryLock()方法比較簡單，直接以非公平的模式去嘗試獲取一次鎖，獲取到了或者鎖本來就是當前執行緒佔有著就返回true，否則返回false。

tryLock(long time, TimeUnit unit)方法

嘗試獲取鎖，並等待一段時間，如果在這段時間內都沒有獲取到鎖，就返回false。

// ReentrantLock.tryLock()
public boolean tryLock(long timeout, TimeUnit unit)
        throws InterruptedException {
    // 呼叫AQS中的方法
    return sync.tryAcquireNanos(1, unit.toNanos(timeout));
}
// AbstractQueuedSynchronizer.tryAcquireNanos()
public final boolean tryAcquireNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    // 如果執行緒中斷了，丟擲異常
    if (Thread.interrupted())
        throw new InterruptedException();
    // 先嚐試獲取一次鎖
    return tryAcquire(arg) ||
        doAcquireNanos(arg, nanosTimeout);
}
// AbstractQueuedSynchronizer.doAcquireNanos()
private boolean doAcquireNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    // 如果時間已經到期了，直接返回false
    if (nanosTimeout <= 0L)
        return false;
    // 到期時間
    final long deadline = System.nanoTime() + nanosTimeout;
    final Node node = addWaiter(Node.EXCLUSIVE);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return true;
            }
            nanosTimeout = deadline - System.nanoTime();
            // 如果到期了，就直接返回false
            if (nanosTimeout <= 0L)
                return false;
            // spinForTimeoutThreshold = 1000L;
            // 只有到期時間大於1000納秒，才阻塞
            // 小於等於1000納秒，直接自旋解決就得了
            if (shouldParkAfterFailedAcquire(p, node) &&
                nanosTimeout > spinForTimeoutThreshold)
                // 阻塞一段時間
                LockSupport.parkNanos(this, nanosTimeout);
            if (Thread.interrupted())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

tryLock(long time, TimeUnit unit)方法在阻塞的時候加上阻塞時間，並且會隨時檢查是否到期，只要到期了沒獲取到鎖就返回false。

unlock()方法

釋放鎖。

// java.util.concurrent.locks.ReentrantLock.unlock()
public void unlock() {
    sync.release(1);
}
// java.util.concurrent.locks.AbstractQueuedSynchronizer.release
public final boolean release(int arg) {
    // 呼叫AQS實現類的tryRelease()方法釋放鎖
    if (tryRelease(arg)) {
        Node h = head;
        // 如果頭節點不為空，且等待狀態不是0，就喚醒下一個節點
        // 還記得waitStatus嗎？
        // 在每個節點阻塞之前會把其上一個節點的等待狀態設為SIGNAL（-1）
        // 所以，SIGNAL的準確理解應該是喚醒下一個等待的執行緒
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}
// java.util.concurrent.locks.ReentrantLock.Sync.tryRelease
protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    // 如果當前執行緒不是佔有著鎖的執行緒，丟擲異常
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    // 如果狀態變數的值為0了，說明完全釋放了鎖
    // 這也就是為什麼重入鎖呼叫了多少次lock()就要呼叫多少次unlock()的原因
    // 如果不這樣做，會導致鎖不會完全釋放，別的執行緒永遠無法獲取到鎖
    if (c == 0) {
        free = true;
        // 清空佔有執行緒
        setExclusiveOwnerThread(null);
    }
    // 設定狀態變數的值
    setState(c);
    return free;
}
private void unparkSuccessor(Node node) {
    // 注意，這裡的node是頭節點
    
    // 如果頭節點的等待狀態小於0，就把它設定為0
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    // 頭節點的下一個節點
    Node s = node.next;
    // 如果下一個節點為空，或者其等待狀態大於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)
        LockSupport.unpark(s.thread);
}

釋放鎖的過程大致為：

（1）將state的值減1；

（2）如果state減到了0，說明已經完全釋放鎖了，喚醒下一個等待著的節點；

未完待續，下一章我們繼續學習ReentrantLock中關於條件鎖的部分

彩蛋

為什麼ReentrantLock預設採用的是非公平模式？

答：因為非公平模式效率比較高。

為什麼非公平模式效率比較高？

答：因為非公平模式會在一開始就嘗試兩次獲取鎖，如果當時正好state的值為0，它就會成功獲取到鎖，少了排隊導致的阻塞/喚醒過程，並且減少了執行緒頻繁的切換帶來的效能損耗。

非公平模式有什麼弊端？

答：非公平模式有可能會導致一開始排隊的執行緒一直獲取不到鎖，導致執行緒餓死。

推薦閱讀

1. 死磕 java同步系列之AQS起篇

2. 死磕 java同步系列之自己動手寫一個鎖Lock

3. 死磕 java魔法類之Unsafe解析

4. 死磕 java同步系列之JMM（Java Memory Model）

5. 死磕 java同步系列之volatile解析

6. 死磕 java同步系列之synchronized解析

---

歡迎關注我的公眾號“彤哥讀原始碼”，檢視更多原始碼系列文章, 與彤哥一起暢遊原始碼的海洋。

相關推薦