詳解Java?ReentrantReadWriteLock讀寫鎖的原理與實(shí)現(xiàn)

更新時(shí)間：2022年10月16日 14:19:41 作者：JAVA旭陽

ReentrantReadWriteLock讀寫鎖是使用AQS的集大成者，用了獨(dú)占模式和共享模式。本文和大家一起理解下ReentrantReadWriteLock讀寫鎖的實(shí)現(xiàn)原理，需要的可以了解一下

概述

ReentrantReadWriteLock讀寫鎖是使用AQS的集大成者，用了獨(dú)占模式和共享模式。本文和大家一起理解下ReentrantReadWriteLock讀寫鎖的實(shí)現(xiàn)原理。在這之前建議大家閱讀下下面3篇關(guān)聯(lián)文章：

深入淺出理解Java并發(fā)AQS的獨(dú)占鎖模式

深入淺出理解Java并發(fā)AQS的共享鎖模式

通俗易懂讀寫鎖ReentrantReadWriteLock的使用

原理概述

上圖是ReentrantReadWriteLock讀寫鎖的類結(jié)構(gòu)圖：

實(shí)現(xiàn)了ReadWriteLock接口，該接口提供了獲取讀鎖和寫鎖的API。
ReentrantReadWriteLock讀寫鎖內(nèi)部的成員變量readLock是讀鎖，指向內(nèi)部類ReadLock。
ReentrantReadWriteLock讀寫鎖內(nèi)部的成員變量writeLock是寫鎖，指向內(nèi)部類WriteLock。
ReentrantReadWriteLock讀寫鎖內(nèi)部的成員變量sync是繼承AQS的同步器，他有兩個(gè)子類FairSync公平同步器和NoFairSync非公平同步器，讀寫鎖內(nèi)部也有一個(gè)sync，他們使用的是同一個(gè)sync。

讀寫鎖用的同一個(gè)sync同步器，那么他們共享同一個(gè)state, 這樣不會(huì)混淆嗎？

不會(huì)，ReentrantReadWriteLock讀寫鎖使用了AQS中state值得低16位表示寫鎖得計(jì)數(shù)，用高16位表示讀鎖得計(jì)數(shù)，這樣就可以使用同一個(gè)AQS同時(shí)管理讀鎖和寫鎖。

1.ReentrantReadWriteLock類重要成員變量

// 讀鎖
private final ReentrantReadWriteLock.ReadLock readerLock;
// 寫鎖
private final ReentrantReadWriteLock.WriteLock writerLock;
// 同步器
final Sync sync;

2.ReentrantReadWriteLock構(gòu)造方法

//默認(rèn)是非公平鎖，可以指定參數(shù)創(chuàng)建公平鎖
public ReentrantReadWriteLock(boolean fair) {
    // true 為公平鎖
    sync = fair ? new FairSync() : new NonfairSync();
    // 這兩個(gè) lock 共享同一個(gè) sync 實(shí)例，都是由 ReentrantReadWriteLock 的 sync 提供同步實(shí)現(xiàn)
    readerLock = new ReadLock(this);
    writerLock = new WriteLock(this);
}

3.Sync類重要成員變量

// 用來移位
static final int SHARED_SHIFT   = 16;
// 高16位的1
static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
// 65535，16個(gè)1，代表寫鎖的最大重入次數(shù)
static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
// 低16位掩碼：0b 1111 1111 1111 1111，用來獲取寫鎖重入的次數(shù)
static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;

// 獲取讀寫鎖的讀鎖分配的總次數(shù)
static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
// 寫鎖（獨(dú)占）鎖的重入次數(shù)
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

加鎖原理

圖解過程

設(shè)計(jì)一個(gè)加鎖場景，t1線程加寫鎖，t2線程加讀鎖，我們看下它們整個(gè)加鎖得流程。

1.t1 加寫鎖w.lock()成功，占了 state 的低 16 位。

這里得state分為兩部分0_1,0表示高16位的值，1表示低16位的值。
AQS當(dāng)前占用線程exclusiveOwnerThread屬性指向t1線程。

2.t2線程執(zhí)行加讀鎖 r.lock()，嘗試獲取鎖，發(fā)現(xiàn)已經(jīng)被寫鎖占據(jù)了，加鎖失敗。

3.t2線程被封裝成一個(gè)共享模式Node.SHARED的節(jié)點(diǎn)，加入到AQS的隊(duì)列中。

4.在阻塞前，t2線程發(fā)現(xiàn)自己是隊(duì)列中的老二，會(huì)嘗試再次獲取讀鎖，因?yàn)閠1沒有釋放，它會(huì)失敗，然后它會(huì)把隊(duì)列的前驅(qū)節(jié)點(diǎn)的狀態(tài)改為-1，然后阻塞自身，也就是t2線程。

上面中黃色三角形就是等待狀態(tài)的值，前驅(qū)節(jié)點(diǎn)變成-1
上面中的灰色表示節(jié)點(diǎn)所在的線程阻塞了

5.后面如過有其他線程如t3,t4加讀鎖或者寫鎖，由于t1線程沒有釋放鎖，會(huì)變成下面的狀態(tài)。

上面是整個(gè)解鎖的流程，下面深入源碼驗(yàn)證這個(gè)流程。

源碼解析

1.寫鎖加鎖源碼

WriteLock類的lock()方法是加寫鎖的入口方法。

static final class NonfairSync extends Sync {
    // ... 省略無關(guān)代碼
 
    // 外部類 WriteLock 方法, 方便閱讀, 放在此處
    public void lock() {
        sync.acquire(1);
    }
 
    // AQS 繼承過來的方法, 方便閱讀, 放在此處
    public final void acquire(int arg) {
        if (
            // 嘗試獲得寫鎖失敗
                !tryAcquire(arg) &&
                        // 將當(dāng)前線程關(guān)聯(lián)到一個(gè) Node 對(duì)象上, 模式為獨(dú)占模式
                        // 進(jìn)入 AQS 隊(duì)列阻塞
                        acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
        ) {
            selfInterrupt();
        }
    }
 
    
    protected final boolean tryAcquire(int acquires) {
        // 獲取當(dāng)前線程
        Thread current = Thread.currentThread();
        //獲得鎖的狀態(tài)
        int c = getState();
        // 獲得低 16 位, 代表寫鎖的 state 計(jì)數(shù)
        int w = exclusiveCount(c);
         // c不等于0表示加了讀鎖或者寫鎖
        if (c != 0) {
            if (
                // c != 0 and w == 0 表示有讀鎖返回錯(cuò)誤，讀鎖不支持鎖升級(jí), 或者
                    w == 0 ||
                        	// w != 0 說明有寫鎖，寫鎖的擁有者不是自己，獲取失敗
                            current != getExclusiveOwnerThread()
            ) {
                // 獲得鎖失敗
                return false;
            }
            // 寫鎖計(jì)數(shù)超過低 16 位最大數(shù)量, 報(bào)異常
            if (w + exclusiveCount(acquires) > MAX_COUNT)
                throw new Error("Maximum lock count exceeded");
            // 寫鎖重入, 獲得鎖成功，沒有并發(fā)，所以不使用 CAS
            setState(c + acquires);
            return true;
        }
        if (
             // c == 0，說明沒有任何鎖，判斷寫鎖是否該阻塞，是 false 就嘗試獲取鎖，失敗返回 false
                writerShouldBlock() ||
                        // 嘗試更改計(jì)數(shù)失敗
                        !compareAndSetState(c, c + acquires)
        ) {
            // 獲得鎖失敗
            return false;
        }
        // 獲得鎖成功，設(shè)置鎖的持有線程為當(dāng)前線程
        setExclusiveOwnerThread(current);
        return true;
    }
 
    // 非公平鎖 writerShouldBlock 總是返回 false, 無需阻塞
    final boolean writerShouldBlock() {
        return false; 
    }
    // 公平鎖會(huì)檢查 AQS 隊(duì)列中是否有前驅(qū)節(jié)點(diǎn), 沒有(false)才去競爭
    final boolean writerShouldBlock() {
        return hasQueuedPredecessors();
    }
}

tryAcquire()方法是模板方法，由子類自定義實(shí)現(xiàn)獲取鎖的邏輯。
線程如果獲取寫鎖失敗的話，通過acquireQueued()方法封裝成獨(dú)占Node加入到AQS隊(duì)列中。

2.讀鎖加鎖源碼

ReadLock類的lock()方法是加讀鎖的入口方法，調(diào)用tryAcquireShared()方法嘗試獲取讀鎖，返回負(fù)數(shù)，失敗，加入到隊(duì)列中。

// 加讀鎖的方法入口
public void lock() {
    sync.acquireShared(1);
}
public final void acquireShared(int arg) {
    // tryAcquireShared 返回負(fù)數(shù), 表示獲取讀鎖失敗，加入到隊(duì)列中
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

tryAcquireShared()方法是一個(gè)模板方法，AQS類中定義語義，子類實(shí)現(xiàn)，如果返回1，表示獲取鎖成功，還有剩余資源，返回0表示獲取成功，沒有剩余資源，返回-1表示失敗。

// 嘗試以共享模式獲取，返回1表示獲取鎖成功，還有剩余資源,返回0表示獲取成功，沒有剩余資源，返回-1，表示失敗
protected final int tryAcquireShared(int unused) {
    Thread current = Thread.currentThread();
    int c = getState();
    // exclusiveCount(c) 代表低 16 位, 寫鎖的 state，成立說明有線程持有寫鎖
    // 寫鎖的持有者不是當(dāng)前線程，則獲取讀鎖失敗，【寫鎖允許降級(jí)】
    if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current)
        return -1;
    
    // 高 16 位，代表讀鎖的 state，共享鎖分配出去的總次數(shù)
    int r = sharedCount(c);
    // 讀鎖是否應(yīng)該阻塞
    if (!readerShouldBlock() &&	r < MAX_COUNT &&
        compareAndSetState(c, c + SHARED_UNIT)) {		// 嘗試增加讀鎖計(jì)數(shù)
        // 加鎖成功
        // 加鎖之前讀鎖為 0，說明當(dāng)前線程是第一個(gè)讀鎖線程
        if (r == 0) {
            firstReader = current;
            firstReaderHoldCount = 1;
        // 第一個(gè)讀鎖線程是自己就發(fā)生了讀鎖重入
        } else if (firstReader == current) {
            firstReaderHoldCount++;
        } else {
            // cachedHoldCounter 設(shè)置為當(dāng)前線程的 holdCounter 對(duì)象，即最后一個(gè)獲取讀鎖的線程
            HoldCounter rh = cachedHoldCounter;
            // 說明還沒設(shè)置 rh
            if (rh == null || rh.tid != getThreadId(current))
                // 獲取當(dāng)前線程的鎖重入的對(duì)象，賦值給 cachedHoldCounter
                cachedHoldCounter = rh = readHolds.get();
            // 還沒重入
            else if (rh.count == 0)
                readHolds.set(rh);
            // 重入 + 1
            rh.count++;
        }
        // 讀鎖加鎖成功
        return 1;
    }
    // 邏輯到這 應(yīng)該阻塞，或者 cas 加鎖失敗
    // 會(huì)不斷嘗試 for (;;) 獲取讀鎖, 執(zhí)行過程中無阻塞
    return fullTryAcquireShared(current);
}

// 非公平鎖 readerShouldBlock 偏向?qū)戞i一些，看 AQS 阻塞隊(duì)列中第一個(gè)節(jié)點(diǎn)是否是寫鎖，是則阻塞，反之不阻塞
// 防止一直有讀鎖線程，導(dǎo)致寫鎖線程饑餓
// true 則該阻塞, false 則不阻塞
final boolean readerShouldBlock() {
    return apparentlyFirstQueuedIsExclusive();
}

// 下面是公平鎖的readerShouldBlock
// 公平鎖會(huì)檢查 AQS 隊(duì)列中是否有前驅(qū)節(jié)點(diǎn), 沒有(false)才去競爭
final boolean readerShouldBlock() {
    return hasQueuedPredecessors();
}

fullTryAcquireShared()方法是通過自旋的方式不斷獲取讀鎖，因?yàn)橛捎谇懊娴?code>readerShouldBlock返回false或者cas失敗，導(dǎo)致沒有獲取到鎖，需要不斷重試。

final int fullTryAcquireShared(Thread current) {
    // 當(dāng)前讀鎖線程持有的讀鎖次數(shù)對(duì)象
    HoldCounter rh = null;
    for (;;) {
        int c = getState();
        // 說明有線程持有寫鎖
        if (exclusiveCount(c) != 0) {
            // 寫鎖不是自己則獲取鎖失敗
            if (getExclusiveOwnerThread() != current)
                return -1;
        } else if (readerShouldBlock()) {
            // 條件成立說明當(dāng)前線程是 firstReader，當(dāng)前鎖是讀忙碌狀態(tài)，而且當(dāng)前線程也是讀鎖重入
            if (firstReader == current) {
                // assert firstReaderHoldCount > 0;
            } else {
                if (rh == null) {
                    // 最后一個(gè)讀鎖的 HoldCounter
                    rh = cachedHoldCounter;
                    // 說明當(dāng)前線程也不是最后一個(gè)讀鎖
                    if (rh == null || rh.tid != getThreadId(current)) {
                        // 獲取當(dāng)前線程的 HoldCounter
                        rh = readHolds.get();
                        // 條件成立說明 HoldCounter 對(duì)象是上一步代碼新建的
                        // 當(dāng)前線程不是鎖重入，在 readerShouldBlock() 返回 true 時(shí)需要去排隊(duì)
                        if (rh.count == 0)
                            // 防止內(nèi)存泄漏
                            readHolds.remove();
                    }
                }
                if (rh.count == 0)
                    return -1;
            }
        }
        // 越界判斷
        if (sharedCount(c) == MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        // 讀鎖加鎖，條件內(nèi)的邏輯與 tryAcquireShared 相同
        if (compareAndSetState(c, c + SHARED_UNIT)) {
            if (sharedCount(c) == 0) {
                firstReader = current;
                firstReaderHoldCount = 1;
            } else if (firstReader == current) {
                firstReaderHoldCount++;
            } else {
                if (rh == null)
                    rh = cachedHoldCounter;
                if (rh == null || rh.tid != getThreadId(current))
                    rh = readHolds.get();
                else if (rh.count == 0)
                    readHolds.set(rh);
                rh.count++;
                cachedHoldCounter = rh; // cache for release
            }
            return 1;
        }
    }
}

doAcquireShared()是在獲取讀鎖失敗的時(shí)候加入AQS隊(duì)列的邏輯。

private void doAcquireShared(int arg) {
    // 將當(dāng)前線程關(guān)聯(lián)到一個(gè) Node 對(duì)象上, 模式為共享模式
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            // 獲取前驅(qū)節(jié)點(diǎn)
            final Node p = node.predecessor();
            // 如果前驅(qū)節(jié)點(diǎn)就頭節(jié)點(diǎn)就去嘗試獲取鎖
            if (p == head) {
                // 再一次嘗試獲取讀鎖
                int r = tryAcquireShared(arg);
                // r >= 0 表示獲取成功
                if (r >= 0) {
                    //【這里會(huì)設(shè)置自己為頭節(jié)點(diǎn)，喚醒相連的后序的共享節(jié)點(diǎn)】
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            // 是否在獲取讀鎖失敗時(shí)阻塞      					 park 當(dāng)前線程
            if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

setHeadAndPropagate()方法是在后續(xù)讀鎖被喚醒后，搶到鎖要處理的邏輯，包括修改隊(duì)列的頭結(jié)點(diǎn)，以及喚醒隊(duì)列中的下一個(gè)共享節(jié)點(diǎn)。

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; 
    // 設(shè)置自己為 head 節(jié)點(diǎn)
    setHead(node);
    // propagate 表示有共享資源（例如共享讀鎖或信號(hào)量），為 0 就沒有資源
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        // 獲取下一個(gè)節(jié)點(diǎn)
        Node s = node.next;
        // 如果當(dāng)前是最后一個(gè)節(jié)點(diǎn)，或者下一個(gè)節(jié)點(diǎn)是【等待共享讀鎖的節(jié)點(diǎn)】
        if (s == null || s.isShared())
            // 喚醒后繼節(jié)點(diǎn)
            doReleaseShared();
    }
}

解鎖原理

圖解過程

由于上面t1線程加的寫鎖，所有其他的線程都被阻塞了，只有在t1線程解鎖以后，其他線程才能被喚醒，我們現(xiàn)在看下t1線程被喚醒了，會(huì)發(fā)生什么？

1.t1線程執(zhí)行解鎖w.unlock()成功，修改AQS中的state。

這里的state變?yōu)榱?_0。
AQS當(dāng)前占用線程exclusiveOwnerThread屬性變?yōu)閚ull。

2.t1線程喚醒隊(duì)列中等待的老二，為什么不是老大，因?yàn)槔洗笫且粋€(gè)空節(jié)點(diǎn)，不會(huì)設(shè)置任何的線程。t2線程被喚醒后，搶鎖成功，修改state中高16位為1。

老二的線程節(jié)點(diǎn)變?yōu)樗{(lán)色節(jié)點(diǎn)
AQS中的state變?yōu)?_0。

3.t2線程恢復(fù)運(yùn)行，設(shè)置原來的老二節(jié)點(diǎn)為頭節(jié)點(diǎn)

4.t2線程要做的事情還沒結(jié)束呢，因?yàn)槭枪蚕砟Ｊ?，它現(xiàn)在釋放了，就此時(shí)也喚醒隊(duì)列中的下一個(gè)共享節(jié)點(diǎn)。

5.t3線程恢復(fù)去競爭讀鎖成功，這時(shí)state的高位+1，變成2。

6.這時(shí)候t3線程所在的Node設(shè)置為頭節(jié)點(diǎn)，同時(shí)發(fā)現(xiàn)對(duì)列的下一個(gè)節(jié)點(diǎn)不是共享節(jié)點(diǎn)，而是獨(dú)占節(jié)點(diǎn)，就不會(huì)喚醒后面的節(jié)點(diǎn)了。

7.之后t2線程和t3線程進(jìn)入尾聲，執(zhí)行r.unlock操作，state的計(jì)數(shù)減一，直到變?yōu)?。

8.最后寫鎖線程t4被喚醒，去搶占鎖成功，整個(gè)流程結(jié)束。

上面是整個(gè)解鎖的流程，下面深入源碼驗(yàn)證這個(gè)流程。

源碼解析

1.寫鎖釋放流程

WriteLock類的unlock()方法是入口方法，調(diào)用tryRelease()方法釋放鎖，如果成功，調(diào)用unparkSuccessor()方法喚醒線程。

public void unlock() {
    // 釋放鎖
    sync.release(1);
}
public final boolean release(int arg) {
    // 嘗試釋放鎖
    if (tryRelease(arg)) {
        Node h = head;
        // 頭節(jié)點(diǎn)不為空并且不是等待狀態(tài)不是 0，喚醒后繼的非取消節(jié)點(diǎn)
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

tryRelease()方法是AQS提供的模板方法，返回true表示成功，false失敗，由自定義同步器實(shí)現(xiàn)。

protected final boolean tryRelease(int releases) {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    int nextc = getState() - releases;
    // 因?yàn)榭芍厝氲脑? 寫鎖計(jì)數(shù)為 0, 才算釋放成功
    boolean free = exclusiveCount(nextc) == 0;
    if (free)
        // 設(shè)置占用線程為null
        setExclusiveOwnerThread(null);
    setState(nextc);
    return free;
}

2.讀鎖釋放流程

ReadLock類的unlock()方法是釋放共享鎖的入口方法。

public void unlock() {
    sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

tryReleaseShared()方法是由AQS提供的模板方法，由自定義同步器實(shí)現(xiàn)。

protected final boolean tryReleaseShared(int unused) {
    //自選
    for (;;) {
        int c = getState();
        int nextc = c - SHARED_UNIT;
        // 讀鎖的計(jì)數(shù)不會(huì)影響其它獲取讀鎖線程, 但會(huì)影響其它獲取寫鎖線程，計(jì)數(shù)為 0 才是真正釋放
        if (compareAndSetState(c, nextc))
            // 返回是否已經(jīng)完全釋放了 
            return nextc == 0;
    }
}

調(diào)用doReleaseShared()方法喚醒等待的線程，這個(gè)方法調(diào)用的地方有兩處，還記得嗎，一個(gè)這是里的解鎖，還有一個(gè)是前面加共享鎖阻塞的地方，喚醒后獲取鎖成功，也會(huì)調(diào)用doReleaseShared()方法。

private void doReleaseShared() {
    // 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一個(gè)節(jié)點(diǎn) unpark
	// 如果 head.waitStatus == 0 ==> Node.PROPAGATE
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            // SIGNAL 喚醒后繼
            if (ws == Node.SIGNAL) {
                // 因?yàn)樽x鎖共享，如果其它線程也在釋放讀鎖，那么需要將 waitStatus 先改為 0
            	// 防止 unparkSuccessor 被多次執(zhí)行
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;  
                // 喚醒后繼節(jié)點(diǎn)
                unparkSuccessor(h);
            }
            // 如果已經(jīng)是 0 了，改為 -3，用來解決傳播性
            else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                
        }
        // 條件不成立說明被喚醒的節(jié)點(diǎn)非常積極，直接將自己設(shè)置為了新的 head，
        // 此時(shí)喚醒它的節(jié)點(diǎn)（前驅(qū)）執(zhí)行 h == head 不成立，所以不會(huì)跳出循環(huán)，會(huì)繼續(xù)喚醒新的 head 節(jié)點(diǎn)的后繼節(jié)點(diǎn)
        if (h == head)                   
            break;
    }
}

以上就是詳解Java ReentrantReadWriteLock讀寫鎖的原理與實(shí)現(xiàn)的詳細(xì)內(nèi)容，更多關(guān)于Java ReentrantReadWriteLock讀寫鎖的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: