點關注，不迷路！如果本文對你有幫助的話不要忘記點贊支持哦！

顯示鎖ReentrantLock的內部同步依賴于AQS（AbstractQueuedSynchronizer），因此，分析ReentrantLock必然涉及AQS，

本文假設讀者已熟練掌握AQS的基本原理（參考AQS的基本原理），通過分析ReentrantLock#lock()與ReentrantLock#unlock()的實作原理，用實體幫助讀者理解AQS的等待佇列模型，

JDK版本：oracle java 1.8.0_102

介面宣告

public interface Lock {
    void lock();

    void lockInterruptibly() throws InterruptedException;

    boolean tryLock();

    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;

    void unlock();

    Condition newCondition();
}

ReentrantLock對標內置鎖，實作了Lock介面，忽略Condition相關，主要提供lock、unlock兩種語意，和兩種語意的衍生品，

實作原理

繼承AQS

本文中的“繼承”指“擴展extend”，

AQS復習

AQS并提供了多個未實作的protected方法，留給作者覆寫以開發不同的同步器:

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    ...
    protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }
    protected boolean tryRelease(int arg) {
        throw new UnsupportedOperationException();
    }
    protected int tryAcquireShared(int arg) {
        throw new UnsupportedOperationException();
    }
    protected boolean tryReleaseShared(int arg) {
        throw new UnsupportedOperationException();
    }
    protected boolean tryReleaseShared(int arg) {
        throw new UnsupportedOperationException();
    }
    ...
}

而其他非私有方法則使用final修飾，禁止子類覆寫，

繼承

ReentrantLock支持公平、非公平兩種策略，并通過繼承AQS實作了對應兩種策略的同步器NonfairSync與FairSync，ReentrantLock默認使用非公平策略，即NonfairSync：

    public ReentrantLock() {
        sync = new NonfairSync();
    }
    ...
    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;

        final void lock() {
            ...
        }

        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
    }
    abstract static class Sync extends AbstractQueuedSynchronizer {
        ...
        abstract void lock();

        final boolean nonfairTryAcquire(int acquires) {
            ...
        }

        protected final boolean tryRelease(int releases) {
            ...
        }

        final ConditionObject newCondition() {
            return new ConditionObject();
        }
        ...
    }
    ...

先不追究細節，下面以默認的非公平策略為例，講解lock和unlock的實作，

lock

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

非公平策略下，sync指向一個NonfairSync實體，

    static final class NonfairSync extends Sync {
        ...
        final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }
        ...
    }

ReentrantLock用state表示“所有者執行緒已經重復獲取該鎖的次數”，當state等于0時，表示當前沒有執行緒持有該鎖，因此，將state CAS設定為1，并記錄排他的所有者執行緒ownerThread（ownerThread只會在0->1及1->0兩次狀態轉換中修改）；否則，state必然大于0，則嘗試再獲取一次鎖，ownerThread將在state大于0時，用于判斷重入性，

排他性：如果執行緒T1已經持有鎖L，則不允許除T1外的任何執行緒T持有該鎖L
重入性：如果執行緒T1已經持有鎖L，則允許執行緒T1多次獲取鎖L，更確切的說，獲取一次后，可多次進入鎖，

二者結合，描述了ReentrantLock的一個性質：允許ownerThread重入，不允許其他執行緒進入或重入，

acquire

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    ...
    public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }
    static void selfInterrupt() {
        Thread.currentThread().interrupt();
    }
    ...
}

改寫：

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    ...
    public final void acquire(int arg) {
        if (tryAcquire(arg)) {
            return;
        }
        Node newNode = addWaiter(Node.EXCLUSIVE);
        boolean interrupted = acquireQueued(newNode, arg);
        if (interrupted) {
            selfInterrupt();
        }
    }
    ...
}

首先，通過tryAcquire()嘗試獲取鎖，按照AQS的約定，tryAcquire()回傳true表示獲取成功，可直接回傳；否則獲取失敗，如果獲取失敗，則向等待佇列中添加一個獨占模式的節點，并通過acquireQueued()阻塞的等待該節點被呼叫（即當前執行緒被喚醒），如果是因為被中斷而喚醒的，則復現中斷信號，

tryAcquire

NonfairSync覆寫了AQS#tryAcquire()：

    static final class NonfairSync extends Sync {
        ...
        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
        ...
    }
    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;
        }
        ...
    }

12-17行重復了NonfairSync#lock()中state=0時的狀態轉換，18行進行排他性判斷，如果當前執行緒等于ownerThread，則直接回傳false，否則，19-22行進行重入，state加1（acquires=1），表示所有者執行緒重復獲取該鎖的次數增加1，

實際上，NonfairSync#lock()不需要特殊處理state=0時的狀態轉換，可通過NonfairSync#tryAcquire()、Sync#nonfairTryAcquire()完成，

為什么19-22行不需要同步

注意，如果18行判斷當前執行緒等于ownerThread，則根據程式順序規則，19-22行不需要同步，因為同一執行緒中，第二次呼叫NonfairSync#tryAcquire()時（會進入19-22行），第一次呼叫鎖寫入的state、ownerThread一定是可見的，

為什么要用state表示重入次數

如果沒有記錄重入次數，則第一次釋放鎖時，會一次性把ownerThread多次重入的鎖都釋放掉，而此時“鎖中的代碼”還沒有執行完成，造成混亂，

addWaiter

如果tryAcquire()獲取失敗，則要通過AQS#addWaiter()向等待佇列中添加一個獨占模式的節點，并回傳該節點：

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    ...
    private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
        // Try the fast path of enq; backup to full enq on failure
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }
    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;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }
    ...
}

AQS#enq()中重復了9-15行的邏輯，直接看enq()，

如果尾指標為null，則頭指標也一定為null，表示等待佇列未初始化，就CAS初始化佇列（常見于無阻塞佇列的設計中，如原始碼|并發一枝花之BlockingQueue），然后繼續回圈，如果尾指標非null，則佇列已初始化，就CAS嘗試在尾節點后插入新的節點node，

在插入程序中，會出現“node.prev指向舊的尾節點，但舊的尾節點.next為null未指向node（盡管，尾指標指向node）”的狀態，即“佇列在prev方向一致，next方向不一致”，記住該狀態，分析ReentrantLock#unlock()時會用到，

最后，enq()回傳舊的尾節點，但外層的AQS#addWaiter()仍然回傳新節點node，

佇列剛完成初始化時，存在一個dummy node，插入節點時，tail后移指向新節點，head不變仍然指向dummy node，直到呼叫AQS#acquireQueued()時，head才會后移，消除了dummy node，后面分析，

acquireQueued

插入新節點node后，通過AQS#acquireQueued()阻塞的等待該節點被呼叫（即當前執行緒被喚醒）：

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    ...
    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);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
    ...
}

該方法是lock程序的核心難點，需要結合AQS#addWaiter()理解AQS內部基于等待佇列的同步模型，

AQS的核心是狀態依賴，可概括為兩條規則：

當狀態還沒有滿足的時候，節點會進入等待佇列，
特別的，獲取成功的節點成為佇列的頭結點，

首先，AQS#addWaiter()會將新節點node加入隊尾（維護規則“當狀態還沒有滿足的時候，節點會進入等待佇列”），然后，AQS#acquireQueued()檢查node的前繼節點是否是頭節點，如果是，則嘗試獲取鎖；如果不是，或獲取所失敗，都會嘗試阻塞等待，

如果11行獲取鎖成功，則更新頭節點（維護規則“獲取成功的節點成為佇列的頭結點”），修改failed標志，并回傳interrupted標志，interrupted初始化為false，可能在17-19行被修改，

初始化佇列后的第一次更新頭結點，直接setHead消除了dummy node，消除之后，實際節點代替了dummy node的作用，但與dummy node不同的是，該節點是持有鎖的，

如果11行判斷前繼節點不是頭節點或獲取鎖失敗，則進入17-19行，AQS.shouldParkAfterFailedAcquire()判斷是否需要阻塞等待，如果需要，則通過AQS#parkAndCheckInterrupt()阻塞等待，直到被喚醒或被中斷，

shouldParkAfterFailedAcquire

AQS.shouldParkAfterFailedAcquire()根據pred.waitStatus判斷新節點node是否應該被阻塞：

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    ...
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            return true;
        if (ws > 0) {
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }
    ...
}

AQS#addWaiter()構造新節點時，pred.waitStatus使用了默認值0，此時，進入14-16行，CAS設定pred.waitStatus為SIGNAL==-1，最后回傳false，

回到AQS#acquireQueued()中后，由于AQS#parkAndCheckInterrupt()回傳false，回圈會繼續進行，假設node的前繼節點pred仍然不是頭結點或鎖獲取失敗，則會再次進入AQS#parkAndCheckInterrupt()，上一輪回圈中，已經將pred.waitStatus設定為SIGNAL==-1，則這次會進入7-8行，直接回傳true，表示應該阻塞，

什么時候會遇到ws > 0的case呢？當pred所維護的獲取請求被取消時，pred.waitStatus會被設定為CANCELLED==1，從而進入9-14行，改寫：

if (ws > 0) {
    do {
        pred = pred.prev;
        node.prev = pred;
    } while (pred.waitStatus > 0);
    pred.next = node;
}

邏輯很簡單，回圈移除所有被取消的前繼節點pred，直到找到未被取消的pred，移除所有被取消的前繼節點后，直接回傳false，

注意，在執行6行之前，佇列處于“node.prev指向最新的前繼節點，但pred.next指向已經移除的后繼節點”的狀態，即“佇列在prev方向一致，next方向不一致”，記住該狀態，分析ReentrantLock#unlock()時會用到，

此處不需要檢查前繼節點是否為null，因為等待佇列的頭結點要么是dummy node，滿足dummy.waitStatus==0；要么是剛替換的real node，滿足real.waitStatus==0；要么是后繼節點已經阻塞的節點，滿足real.waitStatus==SIGNAL==-1，則最晚遍歷到頭結點時，一定會退出回圈，不會出現pred為null的情況，

回到AQS#acquireQueued()后，重新檢查前繼節點是否為頭節點，并作出相應處理，

經過多次回圈執行AQS.shouldParkAfterFailedAcquire()后，等待佇列趨于穩定，最終的穩定狀態為：

除了頭節點，剩余節點都會回傳true，表示需要阻塞等待，
除了尾節點，剩余節點都滿足waitStatus==SIGNAL，表示釋放后需要喚醒后繼節點，

parkAndCheckInterrupt

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
...
    private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }
...
}

AQS#parkAndCheckInterrupt()借助LockSupport.park()實作阻塞等待，最后呼叫Thread.interrupted()檢查是否被中斷，并清除中斷狀態，并回傳中斷標志，

如果是被中斷的，則需要在外層AQS#acquireQueued()中重新設定中斷標志interrupted，并在下一次回圈中回傳，然后在更外層的AQS#acquire()中呼叫AQS.selfInterrupt()重放中斷，

為什么不能直接在AQS#parkAndCheckInterrupt()回傳后中斷？因為回傳中轉標志能提供更大的靈活性，外界可以自行決定是即時重放、稍后重放還是壓根不重放，Condition在得知AQS#acquireQueued()是被中斷的之后，便沒有直接復現中斷，而是根據REINTERRUPT配置決定是否重放，

cancelAcquire

最后，如果在執行AQS#acquire()的程序中拋出任何例外，則取消任務：

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    ...
    private void cancelAcquire(Node node) {
        ...
        node.waitStatus = Node.CANCELLED;
        ...
    }
    ...
}

因此，如果指考慮ReentrantLock#lock()方法的話，那么被標記為CACELLED狀態的節點一定在獲取鎖時拋出了例外，AQS.shouldParkAfterFailedAcquire()中清理了這部分CACELLED節點，

超時版ReentrantLock#tryLock()中，還可以由于超時觸發取消，

lock小結

ReentrantLock#lock()收斂后，AQS內部的等待佇列如圖：

image.png

unlock

ReentrantLock#unlcok()與ReentrantLock#lcok()是對偶的，

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

獲取鎖以單位1進行，釋放鎖時也以單位1進行，

release

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;
    }
    ...
}

首先，通過tryRelease()嘗試釋放鎖，按照AQS的約定，tryRelease()回傳true表示已完全釋放，可喚醒所有阻塞執行緒；否則沒有完全釋放，不需要喚醒，如果已完全釋放，則只需要喚醒頭結點的后繼節點，該節點的ownerThread必然與頭結點不同（如果相同，則之前lock時能夠重入，不需要排隊）；否則沒有完全釋放，不需要喚醒任何節點，

對于獨占鎖，“完全釋放”表示ownerThread的所有重入操作均已結束，

解釋8行的判斷

如果h == null，則佇列還未初始化（回憶AQS#enq()），如果h.waitStatus == 0，則要么剛剛初始化佇列，只有一個dummy node，沒有后繼節點（回憶AQS#enq()）；要么后繼節點還沒被阻塞，不需要喚醒（回憶等待佇列的穩定狀態），

tryRelease

對照tryAcquire()分析tryRelease()：

    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;
        }
        ...
    }

5-6行很重要，如果存在某執行緒持有鎖，則可以檢查unlock是否被ownerThread觸發；如果不存在執行緒持有鎖，則ownerThread==null，可以檢查是否在未lock的情況下進行unlock，或者重復執行了unlock，

因此，使用ReentrantLock時，try-finally要這么寫：
Lock lock = new ReentrantLock();
lock.lock();
try {
 // do sth
} finally {
 lock.unlcok();
}
確保在呼叫lock()成功之后，才能呼叫unlock()，

接下來，8行判斷是否將要進行1->0的狀態轉換，如果是，則可以完全釋放鎖，將ownerThread置為null，然后設定state，

最后，回傳是否可完全釋放的標志free，

可見性問題

為了抓住核心功能，前面一直忽略了一個很重要的問題——可見性，忽略可見性問題的話，閱讀原始碼基本沒有影響，但自己實作同步器時將帶來噩夢，

以此處為例，是應該先執行8-11行，還是先執行12行呢？或者是無所謂呢？

為保障可見性，必須先執行8-11行，再執行12行，因為exclusiveOwnerThread的可見性要借助于（piggyback）于volatile變數state：

    ...
    private transient Thread exclusiveOwnerThread;
    ...
    private volatile int state;
    ...

配套的，也必須先讀state，再讀exclusiveOwnerThread：

    abstract static class Sync extends AbstractQueuedSynchronizer {
        ...
        final boolean nonfairTryAcquire(int acquires) {
            ...
            int c = getState(); // 先讀state
            if (c == 0) {
                ...
            }
            else if (current == getExclusiveOwnerThread()) {    // 再讀exclusiveOwnerThread
                ...
            }
            return false;
        }
        ...
    }

核心是三條Happens-Before規則：

程式順序規則：如果程式中操作A在操作B之前，那么在執行緒中操作A將在操作B之前執行，
傳遞性：如果操作A在操作B之前執行，并且操作B在操作C之前執行，那么操作A必須在操作C之前執行，
volatile變數規則：對volatile變數的寫入操作必須在對該變數的讀操作之前執行，

具體來說，“先寫exclusiveOwnerThread再寫state；先讀state再讀exclusiveOwnerThread”的方案，保證了“在讀state之后，發生在寫state之前的寫exclusiveOwnerThread操作對發生在讀state之后的讀exclusiveOwnerThread操作一定是可見的”，

程式順序規則、傳遞性兩潭訓本規則，經常與監視器鎖規則、volatile變數規則顯示的搭配，一定要掌握，

相對的，執行緒啟動規則、執行緒結束規則、中斷規則、終結器規則則通常被隱式的使用，

unparkSuccessor

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    ...
    private void unparkSuccessor(Node node) {
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);

        Node s = node.next;
        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);
    }
    ...
}

對lock程序的分析中，我們得知，佇列中所有節點的waitStatus要么為0，要么為SIGNAL==-1，當node.waitStatus==SIGNAL時，表示node的后繼節點s已被阻塞或正在被阻塞，現在需要喚醒s，則7-8行將node.waitStatus置0，

注意，node.waitStatus一定不為CANCELLED==1，因為如果lock()方法沒有執行成功，就無法通過unlock()方法呼叫AQS#unparkSuccessor()，

接下來，10-18行從尾節點向前遍歷，找到node后最靠近node的未取消的節點，如果存在該節點s(s!=null)，就喚醒s.thread以競爭鎖，

一致性問題

為什么要從尾節點向前遍歷，而不能從node向后遍歷？這是因為，AQS中的等待佇列基于一個弱一致性雙向鏈表實作，允許某些時刻下，佇列在prev方向一致，next方向不一致，

理想情況下，佇列每時每刻都處于一致的狀態（強一致性模型），從node向后遍歷找第一個未取消節點是更高效的做法，然而，維護一致性通常需要犧牲部分性能，為了進一步的提升性能，腦洞大開的神牛們想出了各種高性能的弱一致性模型，盡管模型允許了更多弱一致狀態，但所有弱一致狀態都在控制之下，不會出現一致性問題，

回憶lock程序的分析，有兩個地方出現了這個弱一致狀態：

AQS#enq()插入新節點（包括AQS#addWaiter()）的程序中，舊的尾節點next為null未指向新節點，對應條件s == null，如圖：

image.png

AQS.shouldParkAfterFailedAcquire()移除CACELLED節點的程序中，中間節點指向已被移除的CACELLED節點，對應條件s.waitStatus > 0，如圖：

image.png

因此，從node開始，沿著next方向向后遍歷是行不通的，只能從tail開始，沿著prev方向向前遍歷，直到找到未取消的節點（s != null），或遍歷完node的所有后繼子孫（s == null），

當然，s == null也可能表示node恰好是尾節點，該狀態是強一致的，但仍然可以復用該段代碼，

unlock小結

ReentrantLock#unlock()收斂后，AQS內部的等待佇列如圖：

總結

本文的目標是借助對ReentrantLock#lock()與ReentrantLock#unlock()的分析，理解AQS的等待佇列模型，

實際上，ReentrantLock還有一些重要的特性和API，如ReentrantLock#lockInterruptibly()、ReentrantLock#newCondition()，后面分先后分析這兩個API，加深對AQS的理解，

到此這篇關于文章就結束了！

點關注，不迷路！如果本文對你有幫助的話不要忘記點贊支持哦！

上述面試題答案都整理成檔案筆記，也還整理了一些面試資料&最新2020收集的一些大廠的面試真題（都整理成檔案，小部分截圖），有需要的可以點擊進入暗號：csdn ，

希望對大家有所幫助，有用的話點贊給我支持！

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/16715.html

標籤：其他

上一篇：面臨挑戰：2020最新Spring面試題詳解（上）

下一篇：使用DOCKER 在WINDOWS 上編譯 OpenHarmony

不要死磕Java并發了,阿里P7架構師帶你深入剖析synchronized的實作原理

介面宣告

實作原理

繼承AQS

AQS復習

繼承

lock

acquire

lock小結

unlock

release

unlock小結

總結