一、回顧基本的讀寫鎖

我們知道讀寫鎖 #java.util.concurrent.locks.ReentrantReadWriteLock 是一個 “讀寫互斥，寫寫互斥，讀讀共享” 的鎖。

讀寫鎖的使用非常簡單，那就是：

我們只需要保證讀鎖和寫鎖來自同一個 ReentrantReadWriteLock 即可，我們知道基於 AQS 實現的鎖都是使用一個 原子state 來進行資源控制，那麼讀寫鎖是如何去控制這個 原子state 的呢？

注：對重入鎖 ReentrantLock或者 AQS 原始碼不熟悉的讀者往下閱讀會有一定的困難，請閱讀 圖文深入解析Java顯式鎖底層原始碼 —— 加解鎖是如何實現的

！

---

二、讀寫鎖概述

如果閱讀過前面文章或者說對普通重入鎖 #java.util.concurrent.locks.ReentrantLock 有一定了解的小夥伴應該知道，重入鎖的實現，就是

1、使用 CAS 對 原子state 進行操作，再根據操作的結果來進行資源控制，且當獲取資源失敗後，

2、使用 On Sync Queue 來進行阻塞等待排隊，並等待喚醒以便進行再次對 原子state 進行 CAS 操作來嘗試獲取資源

的這麼一個反覆迴圈的過程。

這裡有一個好訊息，讀寫鎖中寫鎖的資源獲取acquire與釋放release，和重入鎖及其類似。讀鎖在流程上也是分類上面說的兩步，但是邏輯則出入較大，不過有了前面的基礎，看這篇文章應該不會太吃力。讀鎖和寫鎖共用同一個 原子state

和 On Sync Queue 來進行資源控制，那麼接下來我們來看看這是如何實現的吧。

三、讀鎖中對於 原子state 的操作

由於寫鎖和重入鎖基本上是一樣的，所以我們先講讀鎖。上面說到，讀鎖的實現也可拆為兩個階段，我們先說說第一個階段：讀鎖中對於 原子state 的操作。

讀鎖中對 原子state 的操作，也就是 tryAcquireShared 方法，我們結合原始碼和原始碼中的文件，得出如下三步：

• step1：如果寫鎖持鎖，直接獲取資源acquire失敗（返回 -1），持有寫鎖的是本執行緒除外。
• step2：寫鎖不持鎖，則首先根據 queue policy（公平鎖或非公平鎖） 判斷一下要不要阻塞。不需要阻塞則有其次，通過修改 原子state
  來嘗試獲取資源，成功則要修改一下重入計數。
• step3：上面的都失敗了，則進入到fullTryAcquireShared中。

        protected final int tryAcquireShared(int unused) {
            /*
             * Walkthrough:
             * 1. If write lock held by another thread, fail.
             * 2. Otherwise, this thread is eligible for
             *    lock wrt state, so ask if it should block
             *    because of queue policy. If not, try
             *    to grant by CASing state and updating count.
             *    Note that step does not check for reentrant
             *    acquires, which is postponed to full version
             *    to avoid having to check hold count in
             *    the more typical non-reentrant case.
             * 3. If step 2 fails either because thread
             *    apparently not eligible or CAS fails or count
             *    saturated, chain to version with full retry loop.
             */
            Thread current = Thread.currentThread();
            int c = getState();
            if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
                return -1;
            int r = sharedCount(c);
            if (!readerShouldBlock() &&
                r < MAX_COUNT &&
                compareAndSetState(c, c + SHARED_UNIT)) {
                if (r == 0) {
                    firstReader = current;
                    firstReaderHoldCount = 1;
                } else if (firstReader == current) {
                    firstReaderHoldCount++;
                } else {
                    HoldCounter rh = cachedHoldCounter;
                    if (rh == null || rh.tid != getThreadId(current))
                        cachedHoldCounter = rh = readHolds.get();
                    else if (rh.count == 0)
                        readHolds.set(rh);
                    rh.count++;
                }
                return 1;
            }
            return fullTryAcquireShared(current);
        }

另外，原始碼中的文件註釋雖然說了不可重入，但實際上是可以重入的，這裡簡單的說下，如果對寫鎖（或者說重入鎖Reentrantlock）還有印象的小夥伴應該知道，寫鎖的重入實際上是對 原子state 進行++操作。而讀寫則是使用一個 HoldCounter 物件，它的功能很簡單，就是負責重入的計數。

但有一個特例，那就是讀鎖套寫鎖會死鎖，這實際上是讀寫鎖設計上的一個 “缺陷” ，疑問先放在這裡，後面我們會娓娓道來。

---

3.1、原子state 操作之 step1 解析：判斷寫鎖是否已經持鎖

這部分邏輯極其簡單，但是有一個特殊的設計需要特別關注。

            Thread current = Thread.currentThread();
            int c = getState();
            if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
                return -1;

其一，getExclusiveOwnerThread() != current， exclusiveOwnerThread 如字面意思，就是是否為獨佔執行緒擁有者，也就是判斷當前執行緒是否持有寫鎖。

其二，就是剛才提到的特殊設計，c & EXCLUSIVE_MASK != 0，c 是我們的 原子state，如果 c 和 EXCLUSIVE_MASK 按位與後不為零，代表無權獲取資源，即已經有執行緒持有了寫鎖。

        static final int SHARED_SHIFT   = 16;
        static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
...
        /** Returns the number of exclusive holds represented in count  */
        static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

其實很好理解，EXCLUSIVE_MASK ，也就是 01111111111111111，16 個 1。我們知道 原子state是一個 32 位 int，要滿足同其與後為 0，就必須滿足低位都為 0 才可。

那麼問題就來了，為什麼高位不需要做判斷呢？ 答案其實呼之欲出，高位是供讀鎖使用的。讀寫鎖共用一個 原子state，但讀鎖控制高 16 位，寫鎖控制低 16 位。 對於寫鎖來說，低 16 位都為 0 代表沒有持鎖，只要有一個 1 ，則代表某執行緒已經持有寫鎖。

---

3.2、原子state 操作之 step2 解析：公平策略與嘗試 CAS 操作

第二部分程式碼看起來看多，但實際上關鍵的就是幾行，可分為三個判斷。

            int r = sharedCount(c);
            if (!readerShouldBlock() &&
                r < MAX_COUNT &&
                compareAndSetState(c, c + SHARED_UNIT))

3.2.1、CAS 嘗試獲取資源

老生常談。我們知道，重入鎖的重入每次使得 原子state++。讀鎖也是如此，讀鎖被設計為，當有新的執行緒共享了資源，則 原子state++，當然，是高位進行 ++。比如 3.1 那張圖，高位為 3 (0000000000000011)，代表有三個執行緒共享了該鎖。compareAndSetState(c, c + SHARED_UNIT) 這個操作就很好理解了。

SHARED_UNIT 就是高位的 1 ( 1 << 16 )，CAS 操作嘗試將 原子state 由原值修改為原值高位++。

3.2.2、r < MAX_COUNT 避免原子 state 溢位

前面我們說到，讀鎖的高位用於記錄當前有多少個執行緒共享了此讀鎖，但是讀鎖只有 16 位，所以說，我們最多隻能表示 65535 ( 16 個 1 ) 個執行緒共享此讀鎖。一旦超過，那就要進入 step 3 了。

        /** Returns the number of shared holds represented in count. */
        static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }

3.2.3、!readerShouldBlock() 公平鎖與非公平鎖策略

公平鎖和非公平鎖相信大家都知道，這裡不多說。最大的區別就是，公平鎖保證一定 “先到先得”，非公平鎖則不保證。

我們這裡只簡單說說 非公平鎖的 readerShouldBlock() 是怎麼實現的。

    final boolean apparentlyFirstQueuedIsExclusive() {
        Node h, s;
        return (h = head) != null &&
            (s = h.next)  != null &&
            !s.isShared()         &&
            s.thread != null;
    }

滿足如上的邏輯，則直接進入 step3，跳過 step2。

看到 head，就應該聯想起之前文章說過的 On Sync Queue，四個判斷連在一起其實就是匯成一句話，當前 On Sync Queue 沒有正在等待的執行緒，或者正在等待的第一個 Node 是 SHARED 模式，代表可進入 step2。

3.2.4、step2 中做了什麼

程式碼看著很多，但其實很簡單，它的核心就是使用一個 HoldCounter 來儲存重入狀態，以便知道這個執行緒加了幾層鎖，道理和重入鎖的 原子state++ 是一樣的。

至於為什麼第一個持有讀鎖的執行緒使用本地變數 firstReaderHoldCount ++，而其他執行緒使用 HoldCounter ++，個人猜測避免共享讀很少的情況（比如大多數情況就只有一個執行緒持有讀鎖），反覆建立 HoldCounter 物件。

               if (r == 0) { // 為 0 代表沒有任何執行緒
                    firstReader = current;
                    firstReaderHoldCount = 1;
                } else if (firstReader == current) {
                    firstReaderHoldCount++;
                } else {
                    HoldCounter rh = cachedHoldCounter;
                    if (rh == null ||
                        rh.tid != LockSupport.getThreadId(current))
                        cachedHoldCounter = rh = readHolds.get();
                    else if (rh.count == 0)
                        readHolds.set(rh);
                    rh.count++;
                }
                return 1;

---

3.3、原子state 操作之 step3 解析：fullTryAcquireShared

fullTryAcquireShared 程式碼看起來冗長無比，但實際上和前面說的大體是一樣的，但它是一個死迴圈，這裡不過多贅述，就是 step 1 + step 2 的一個自旋版。

1. 判斷是否有寫鎖持鎖
2. 判斷 onSyncQueue 有無等待者之類
3. 高位不能超過 65535
4. 進行 CAS 操作

        /**
         * Full version of acquire for reads, that handles CAS misses
         * and reentrant reads not dealt with in tryAcquireShared.
         */
        final int fullTryAcquireShared(Thread current) {
            /*
             * This code is in part redundant with that in
             * tryAcquireShared but is simpler overall by not
             * complicating tryAcquireShared with interactions between
             * retries and lazily reading hold counts.
             */
            HoldCounter rh = null;
            for (;;) {
                int c = getState();
                if (exclusiveCount(c) != 0) { // 和 step1 一模一樣，通過低位判斷是否持鎖
                    if (getExclusiveOwnerThread() != current)
                        return -1;
                    // else we hold the exclusive lock; blocking here
                    // would cause deadlock.
                } else if (readerShouldBlock()) { // 判斷 onSyncQueue 裡有沒有執行緒在等待，如果有，代表要進入 onSyncQueue 中阻塞了
                    // Make sure we're not acquiring read lock reentrantly
                    if (firstReader == current) {
                        // assert firstReaderHoldCount > 0;
                    } else {
                        if (rh == null) {
                            rh = cachedHoldCounter;
                            if (rh == null ||
                                rh.tid != LockSupport.getThreadId(current)) {
                                rh = readHolds.get();
                                if (rh.count == 0)
                                    readHolds.remove();
                            }
                        }
                        if (rh.count == 0)
                            return -1;
                    }
                }
                if (sharedCount(c) == MAX_COUNT) // 高位不能超過 65535
                    throw new Error("Maximum lock count exceeded");
                if (compareAndSetState(c, c + SHARED_UNIT)) { // CAS 操作 + HoldCounter
                    if (sharedCount(c) == 0) {
                        firstReader = current;
                        firstReaderHoldCount = 1;
                    } else if (firstReader == current) {
                        firstReaderHoldCount++;
                    } else {
                        if (rh == null)
                            rh = cachedHoldCounter;
                        if (rh == null ||
                            rh.tid != LockSupport.getThreadId(current))
                            rh = readHolds.get();
                        else if (rh.count == 0)
                            readHolds.set(rh);
                        rh.count++;
                        cachedHoldCounter = rh; // cache for release
                    }
                    return 1;
                }
            }
        }

四、讀鎖中對於 onSyncQueue 的操作

大體過一遍程式碼，會發現 doAcquireShared 與 acquireQueued 無比的類似！

    /**
     * Acquires in shared uninterruptible mode.
     * @param arg the acquire argument
     */
    private void doAcquireShared(int arg) {
        final Node node = addWaiter(Node.SHARED);
        boolean interrupted = false;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node))
                    interrupted |= parkAndCheckInterrupt();
            }
        } catch (Throwable t) {
            cancelAcquire(node);
            throw t;
        } finally {
            if (interrupted)
                selfInterrupt();
        }
    }

doAcquireShared 與 acquireQueued 都有如下特徵：

1. 進行 addWaiter 將自己塞入 On Sync Queue 尾部
2. 如果於 On Sync Queue 中，且前節點為頭節點，則有機會對 原子state 進行操作
3. 如果前節點的 waitStatus 不為 -1(SIGNAL) 且不大於 0(CANCEL) 則將其改為 -1(SIGNAL)
4. 如果前節點的 waitStatus 是 -1(SIGNAL)，則阻塞等待喚醒。

那前面四點這裡就不多贅述，前面的文章講到 acquireQueued 時，花了大量筆墨和圖文對其進行了講解。我們這裡主要重點講講 setHeadAndPropagate(node, r);

4.1、讀鎖中對於 onSyncQueue 的操作之 setHeadAndPropagate

前面我們說到 acquireQueued（也就是重入鎖中對於 onSyncQueue 的操作的實現 ） 執行緒被喚醒後，會將自己設為 On Sync Queue 頭部，取代其成為新的頭部，這裡貼一張來自之前文章的截圖：

---

---

我們再看看，doAcquireShared 中有了哪些變化，它除了將自己設定為新的 HEAD，還做了一個額外的操作：

    private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        setHead(node);

        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }

前兩步其實大同小異，就是使得自己取代原先 on Sync Queue 的 HEAD 節點。關鍵是後面的步驟，我們可以看到，當 propagate > 0 或者佇列未空，或者頭節點狀態為 -1(SIGNAL) 或者新的頭節點狀態為 -1(SIGNAL) 時（其實已經涵蓋了大多數的情況），關鍵是後面的兩個判斷

s == null || s.isShared()

這代表 on Sync Queue 為空，或者 on Sync Queue 第一個等待的節點 Node 是一個共享鎖節點。

滿足這些條件後來到我們的共享鎖（讀鎖）核心程式碼：

    private void doReleaseShared() {
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
                    if (!h.compareAndSetWaitStatus(Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !h.compareAndSetWaitStatus(0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

我們知道第一個等待的節點 Node 是一個共享鎖節點。此時進入 doReleaseShared()，如果其前節點 waitStatus 為 -1(SIGNAL)，則直接將其喚醒（unparkSuccessor(h); 其實就是 release 的核心操作，即將其喚醒）。

如果說其前節點 waitStatus 為 0(-)，則嘗試將其改為 -3(PROPAGATE)，並繼續。

如果說後續的 Node 都是 SHARE 節點，那麼將會被層層喚醒。

---

4.3、讀鎖中對於 onSyncQueue 的操作圖解

如果說你已經懵了，那麼這個小章節來幫你緩解一下，我們通過實操來講解一下這個過程。我們使用了六個執行緒來進行模擬，程式碼很好理解：

為了更好的模擬整個步驟，我們使用 Thread.sleep(500); 來保證整個過程按照我們預想的來進行。

    public static void main(String[] args) throws InterruptedException {
        ReentrantReadWriteLock l = new ReentrantReadWriteLock();
        ReadLock readLock = l.readLock();
        WriteLock writeLock = l.writeLock();

        writeLock.lock();

        Thread t = new Thread(() -> { readLock.lock();System.out.println("A"); });
        t.setName("A");
        t.start();
        Thread.sleep(500);

        Thread t1 = new Thread(() -> { readLock.lock();System.out.println("B"); });
        t1.setName("B");
        t1.start();
        Thread.sleep(500);

        Thread t2 = new Thread(() -> { readLock.lock();System.out.println("C"); });
        t2.setName("C");
        t2.start();
        Thread.sleep(500);

        Thread tx = new Thread(() -> { writeLock.lock();System.out.println("另一個寫鎖執行緒"); });
        tx.setName("另一個寫鎖執行緒");
        tx.start();
        Thread.sleep(500);

        Thread t3 = new Thread(() -> { readLock.lock();System.out.println("D"); });
        t3.setName("D");
        t3.start();
        Thread.sleep(500);

        writeLock.unlock();
    }

---

1. 主執行緒首先呼叫 writeLock.lock(); ，也就是將 原子state 由 0 替換到 1，因為沒有其他執行緒和它競爭，這個肯定是成功的。
2. 後續的執行緒，無論讀鎖還是寫鎖，修改 原子state 必定失敗，因為 原子state & EXCLUSIVE_MASK 不為 0，因為低位目前是 1，按位與後得 1，然後進入佇列尾部進行阻塞，且標記自己為 SHARE 節點。
3. 主執行緒呼叫 writeLock.unlock(); 喚醒佇列中第二個節點。

---

4.3.1、setHeadAndPropagate 進行狀態傳播前的一些列狀態圖解

當一個共享節點被喚醒後，就來到了我們的核心方法：setHeadAndPropagate，我們說過，讀鎖中對於 On Sync Queue 的操作，大體的流程和重入鎖是一樣的，這裡就不贅述，我們來看看直到 setHeadAndPropagate 被呼叫之前， On Sync Queue 中發生了什麼：

首先可以預想到的是，後續的 5 個執行緒，都將在 On Sync Queue 中等待喚醒，如上圖所示。圖中將佇列第一個節點標記為寫鎖執行緒是為了更好的理解，實際上佇列中第一個節點是一個空節點。我們斷點看看具體是怎麼樣的表現形式：

上圖很明顯可以看出此時 HEAD 的 thread 為空，next 指向 Thread A，層層指入，直到隊尾的 Thread D。

且我們可以知道，被喚醒的執行緒就是 Thread A，也就是 HEAD.next 。

4.3.2、setHeadAndPropagate 圖解！

先回想一下前面說的，setHeadAndPropagate，也就是被喚醒 且 CAS 操作 原子state（高位+1） 成功之後，第一步和之前文章說的一樣，把舊的佇列頭刪去，並將自己設為頭節點，如下圖所示：

後續如果發現佇列未空，或者下一個等待的節點為 SHARE(也就是 Thread B) 節點，則呼叫 doReleaseShared(); 進行解鎖狀態的傳播，實際上就是喚醒 Thread B 去操作 原子state（高位+1），直到發現佇列頭節點的 next 節點不是 SHARE 節點，如下圖所示：

我們佇列中第一個等待的節點變成了 另一個寫鎖執行緒，並且有三個讀執行緒 A、B、C 獲取了鎖，此時 原子state 高位為 3，低位為 0，我們斷點看下是不是這樣：

我們看到最後一次傳遞喚醒，HEAD 的 next 為 另一個寫鎖執行緒 ，且如下圖所示，s.isShared() 為 false，因為 另一個寫鎖執行緒 並不是一個共享節點，是一個獨佔節點，所以不再執行 doReleaseShared(); 喚醒下一個執行緒（其實喚醒了也沒用，因為此時寫執行緒去 CAS 必定失敗）。

此時 原子state 的值為 196608，換算成二進位制，則為 0000000000000011 0000000000000000，符合我們的預期，就是高位為 3，低位為 0。

收起所有的除錯斷點，我們發現剛才的程式碼列印為：

五、讀鎖中如何釋放資源？

前面說到讀鎖獲取資源 acquire 的流程，那麼讀鎖如何釋放資源 release 呢？

回想一下重入鎖，重入鎖每次釋放資源使得 原子state--，當 原子state 為零時，則喚醒佇列中下一個等待的執行緒。

讀鎖實際上也是如此，但要注意讀鎖是有兩個狀態來維護鎖的層級和持有共享鎖的執行緒數量的，一個是 HoldCounter 負責重入計數，一個是 原子state高位，負責記錄有多少個執行緒持有了共享鎖。也就是說，我們釋放資源release時，需要首先使得 HoldCounter--，直到其為零，再去操作 原子state高位。

這些其實猜都能猜的出來，直到 原子state高位歸零，就會去喚醒下一個等待的執行緒，而原始碼中也確實如此：

程式碼很簡單，這裡就不囉嗦了。

六、寫鎖與重入鎖的區別

前面花了很多筆墨寫讀鎖，因為寫鎖和之前的重入鎖基本一致，所以這裡不花太多篇幅去寫寫鎖，寫鎖只有一個地方和重入鎖有區別，那就是重入鎖整個 原子state 都是為鎖服務的，而對於寫鎖來說，只有低位服務於自己，且如果高位不為 0（代表有讀鎖已經持有了資源），也無法 acquire 成功，需要入佇列中進行等待。

七、小彩蛋：為什麼讀鎖套寫鎖會死鎖？

這裡再次囉嗦一次，我們知道對於寫鎖來說，可以對 原子state低位 進行 ++ 來實現重入。對於讀鎖來說呢，可以對 HoldCounter 進行 ++ 來實現重入，講道理互相套不應該死鎖，因為他們互不影響。

那麼為什麼讀鎖套死鎖會死鎖呢？

然而這樣卻不會死鎖：

其實原因很簡單，如果要實現讀鎖套寫鎖，其實十分簡單，只需要在入寫鎖 acquire 時判斷本執行緒是否已經有 HoldCounter 即可。但是這樣會帶來一個問題：

• 假設有十個執行緒正在共享讀鎖，此時其中一個讀執行緒重入了寫鎖，這將導致寫鎖的 獨享形同虛設，正常應該是此執行緒進入等待佇列，等待所有執行緒都釋放了讀鎖後，才能實現獲得獨佔鎖。

然而這個持有讀鎖的執行緒已經在執行邏輯了，它無法進行阻塞，如果進入阻塞狀態，便會有一個問題：那就是這個阻塞的讀執行緒將 永遠無法釋放讀鎖！

這就是讀鎖套寫鎖會死鎖的原因，但我個人認為這種情況，JDK應該將異常丟擲才對，之前的專案中便是碰到了這個情況，找了很久才發現是這個問題，這也算是個小 tips 吧。請大家要注意這個問題 ~ 專案中儘量避免讀寫鎖互相套。

---

7.1、寫鎖套讀鎖卻不會死鎖

因為持有寫鎖的執行緒已經擁有了獨佔鎖，此時再去獲取讀鎖，這個對於整個讀寫鎖來說無關緊要，我們只要保證其他的執行緒無法獲取到讀鎖即可。

---

參考資料：

JAVA12原始碼

---

歡迎討論~ 如果覺得寫得不錯 請點贊喲~ 最後推銷一波自己的 Q群，現在基本沒人... 期待大牛與萌新進來一起討論 JAVA 相關技術 ~ 群 《我萌的征途是架構獅！》 歡迎您 ~

相關推薦