深入理解讀寫鎖ReentrantReadWriteLock

本人免費(fèi)整理了Java高級(jí)資料，涵蓋了Java、Redis、MongoDB、MySQL、Zookeeper、Spring Cloud、Dubbo高并發(fā)分布式等教程，一共30G，需要自己領(lǐng)取。
傳送門：https://mp.weixin.qq.com/s/JzddfH-7yNudmkjT0IRL8Q


1.讀寫鎖的介紹

1.讀寫鎖的介紹

在并發(fā)場(chǎng)景中用于解決線程安全的問(wèn)題，我們幾乎會(huì)高頻率的使用到獨(dú)占式鎖，通常使用java提供的關(guān)鍵字synchronized或者concurrents包中實(shí)現(xiàn)了Lock接口的。它們都是獨(dú)占式獲取鎖，也就是在同一時(shí)刻只有一個(gè)線程能夠獲取鎖。而在一些業(yè)務(wù)場(chǎng)景中，大部分只是讀數(shù)據(jù)，寫數(shù)據(jù)很少，如果僅僅是讀數(shù)據(jù)的話并不會(huì)影響數(shù)據(jù)正確性（出現(xiàn)臟讀），而如果在這種業(yè)務(wù)場(chǎng)景下，依然使用獨(dú)占鎖的話，很顯然這將是出現(xiàn)性能瓶頸的地方。

針對(duì)這種讀多寫少的情況，java還提供了另外一個(gè)實(shí)現(xiàn)Lock接口的ReentrantReadWriteLock(讀寫鎖)。讀寫所允許同一時(shí)刻被多個(gè)讀線程訪問(wèn)，但是在寫線程訪問(wèn)時(shí)，所有的讀線程和其他的寫線程都會(huì)被阻塞。在分析WirteLock和ReadLock的互斥性時(shí)可以按照WriteLock與WriteLock之間，WriteLock與ReadLock之間以及ReadLock與ReadLock之間進(jìn)行分析。更多關(guān)于讀寫鎖特性介紹大家可以看源碼上的介紹（閱讀源碼時(shí)最好的一種學(xué)習(xí)方式，我也正在學(xué)習(xí)中，與大家共勉），這里做一個(gè)歸納總結(jié)：

1. 公平性選擇：支持非公平性（默認(rèn)）和公平的鎖獲取方式，吞吐量還是非公平優(yōu)于公平；

2. 重入性：支持重入，讀鎖獲取后能再次獲取，寫鎖獲取之后能夠再次獲取寫鎖，同時(shí)也能夠獲取讀鎖；

3. 鎖降級(jí)：遵循獲取寫鎖，獲取讀鎖再釋放寫鎖的次序，寫鎖能夠降級(jí)成為讀鎖

要想能夠徹底的理解讀寫鎖必須能夠理解這樣幾個(gè)問(wèn)題：1. 讀寫鎖是怎樣實(shí)現(xiàn)分別記錄讀寫狀態(tài)的？2. 寫鎖是怎樣獲取和釋放的？3.讀鎖是怎樣獲取和釋放的？我們帶著這樣的三個(gè)問(wèn)題，再去了解下讀寫鎖。

2.寫鎖詳解

2.1.寫鎖的獲取

同步組件的實(shí)現(xiàn)聚合了同步器（AQS），并通過(guò)重寫重寫同步器（AQS）中的方法實(shí)現(xiàn)同步組件的同步語(yǔ)義，AQS的底層實(shí)現(xiàn)分析可以。因此，寫鎖的實(shí)現(xiàn)依然也是采用這種方式。在同一時(shí)刻寫鎖是不能被多個(gè)線程所獲取，很顯然寫鎖是獨(dú)占式鎖，而實(shí)現(xiàn)寫鎖的同步語(yǔ)義是通過(guò)重寫AQS中的tryAcquire方法實(shí)現(xiàn)的。源碼為:

protected?final?boolean?tryAcquire(int?acquires)?{
????/*
?????*?Walkthrough:
?????*?1\.?If?read?count?nonzero?or?write?count?nonzero
?????*????and?owner?is?a?different?thread,?fail.
?????*?2\.?If?count?would?saturate,?fail.?(This?can?only
?????*????happen?if?count?is?already?nonzero.)
?????*?3\.?Otherwise,?this?thread?is?eligible?for?lock?if
?????*????it?is?either?a?reentrant?acquire?or
?????*????queue?policy?allows?it.?If?so,?update?state
?????*????and?set?owner.
?????*/
????Thread?current?=?Thread.currentThread();
????//?1\.?獲取寫鎖當(dāng)前的同步狀態(tài)
????int?c?=?getState();
????//?2\.?獲取寫鎖獲取的次數(shù)
????int?w?=?exclusiveCount(c);
????if?(c?!=?0)?{
????????//?(Note:?if?c?!=?0?and?w?==?0?then?shared?count?!=?0)
????????//?3.1?當(dāng)讀鎖已被讀線程獲取或者當(dāng)前線程不是已經(jīng)獲取寫鎖的線程的話
????????//?當(dāng)前線程獲取寫鎖失敗
????????if?(w?==?0?||?current?!=?getExclusiveOwnerThread())
????????????return?false;
????????if?(w?+?exclusiveCount(acquires)?>?MAX_COUNT)
????????????throw?new?Error("Maximum?lock?count?exceeded");
????????//?Reentrant?acquire
????????//?3.2?當(dāng)前線程獲取寫鎖，支持可重復(fù)加鎖
????????setState(c?+?acquires);
????????return?true;
????}
????//?3.3?寫鎖未被任何線程獲取，當(dāng)前線程可獲取寫鎖
????if?(writerShouldBlock()?||
????????!compareAndSetState(c,?c?+?acquires))
????????return?false;
????setExclusiveOwnerThread(current);
????return?true;
}

這段代碼的邏輯請(qǐng)看注釋，這里有一個(gè)地方需要重點(diǎn)關(guān)注，exclusiveCount(c)方法，該方法源碼為：

static?int?exclusiveCount(int?c)?{?return?c?&?EXCLUSIVE_MASK;?}

其中EXCLUSIVE_MASK為:?static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;EXCLUSIVE _MASK為1左移16位然后減1，即為0x0000FFFF。而exclusiveCount方法是將同步狀態(tài)（state為int類型）與0x0000FFFF相與，即取同步狀態(tài)的低16位。那么低16位代表什么呢？根據(jù)exclusiveCount方法的注釋為獨(dú)占式獲取的次數(shù)即寫鎖被獲取的次數(shù)，現(xiàn)在就可以得出來(lái)一個(gè)結(jié)論同步狀態(tài)的低16位用來(lái)表示寫鎖的獲取次數(shù)。同時(shí)還有一個(gè)方法值得我們注意：

static?int?sharedCount(int?c)????{?return?c?>>>?SHARED_SHIFT;?}

該方法是獲取讀鎖被獲取的次數(shù)，是將同步狀態(tài)（int c）右移16次，即取同步狀態(tài)的高16位，現(xiàn)在我們可以得出另外一個(gè)結(jié)論同步狀態(tài)的高16位用來(lái)表示讀鎖被獲取的次數(shù)。現(xiàn)在還記得我們開篇說(shuō)的需要弄懂的第一個(gè)問(wèn)題嗎？讀寫鎖是怎樣實(shí)現(xiàn)分別記錄讀鎖和寫鎖的狀態(tài)的，現(xiàn)在這個(gè)問(wèn)題的答案就已經(jīng)被我們弄清楚了，其示意圖如下圖所示：

現(xiàn)在我們回過(guò)頭來(lái)看寫鎖獲取方法tryAcquire，其主要邏輯為：當(dāng)讀鎖已經(jīng)被讀線程獲取或者寫鎖已經(jīng)被其他寫線程獲取，則寫鎖獲取失敗；否則，獲取成功并支持重入，增加寫狀態(tài)。

2.2.寫鎖的釋放

寫鎖釋放通過(guò)重寫AQS的tryRelease方法，源碼為：

protected?final?boolean?tryRelease(int?releases)?{
????if?(!isHeldExclusively())
????????throw?new?IllegalMonitorStateException();
????//1\.?同步狀態(tài)減去寫狀態(tài)
????int?nextc?=?getState()?-?releases;
????//2\.?當(dāng)前寫狀態(tài)是否為0，為0則釋放寫鎖
????boolean?free?=?exclusiveCount(nextc)?==?0;
????if?(free)
????????setExclusiveOwnerThread(null);
????//3\.?不為0則更新同步狀態(tài)
????setState(nextc);
????return?free;
}

源碼的實(shí)現(xiàn)邏輯請(qǐng)看注釋，不難理解與ReentrantLock基本一致，這里需要注意的是，減少寫狀態(tài)int nextc = getState() - releases;只需要用當(dāng)前同步狀態(tài)直接減去寫狀態(tài)的原因正是我們剛才所說(shuō)的寫狀態(tài)是由同步狀態(tài)的低16位表示的。

3.讀鎖詳解

3.1.讀鎖的獲取

看完了寫鎖，現(xiàn)在來(lái)看看讀鎖，讀鎖不是獨(dú)占式鎖，即同一時(shí)刻該鎖可以被多個(gè)讀線程獲取也就是一種共享式鎖。按照之前對(duì)AQS介紹，實(shí)現(xiàn)共享式同步組件的同步語(yǔ)義需要通過(guò)重寫AQS的tryAcquireShared方法和tryReleaseShared方法。讀鎖的獲取實(shí)現(xiàn)方法為：

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();
????//1\.?如果寫鎖已經(jīng)被獲取并且獲取寫鎖的線程不是當(dāng)前線程的話，當(dāng)前
????//?線程獲取讀鎖失敗返回-1
????if?(exclusiveCount(c)?!=?0?&&
????????getExclusiveOwnerThread()?!=?current)
????????return?-1;
????int?r?=?sharedCount(c);
????if?(!readerShouldBlock()?&&
????????r?<?MAX_COUNT?&&
????????//2\.?當(dāng)前線程獲取讀鎖
????????compareAndSetState(c,?c?+?SHARED_UNIT))?{
????????//3\.?下面的代碼主要是新增的一些功能，比如getReadHoldCount()方法
????????//返回當(dāng)前獲取讀鎖的次數(shù)
????????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;
????}
????//4\.?處理在第二步中CAS操作失敗的自旋已經(jīng)實(shí)現(xiàn)重入性
????return?fullTryAcquireShared(current);
}

代碼的邏輯請(qǐng)看注釋，需要注意的是?當(dāng)寫鎖被其他線程獲取后，讀鎖獲取失敗，否則獲取成功利用CAS更新同步狀態(tài)。另外，當(dāng)前同步狀態(tài)需要加上SHARED_UNIT（(1 << SHARED_SHIFT)即0x00010000）的原因這是我們?cè)谏厦嫠f(shuō)的同步狀態(tài)的高16位用來(lái)表示讀鎖被獲取的次數(shù)。如果CAS失敗或者已經(jīng)獲取讀鎖的線程再次獲取讀鎖時(shí)，是靠fullTryAcquireShared方法實(shí)現(xiàn)的，這段代碼就不展開說(shuō)了，有興趣可以看看。

3.2.讀鎖的釋放

讀鎖釋放的實(shí)現(xiàn)主要通過(guò)方法tryReleaseShared，源碼如下，主要邏輯請(qǐng)看注釋：

protected?final?boolean?tryReleaseShared(int?unused)?{
????Thread?current?=?Thread.currentThread();
????//?前面還是為了實(shí)現(xiàn)getReadHoldCount等新功能
????if?(firstReader?==?current)?{
????????//?assert?firstReaderHoldCount?>?0;
????????if?(firstReaderHoldCount?==?1)
????????????firstReader?=?null;
????????else
????????????firstReaderHoldCount--;
????}?else?{
????????HoldCounter?rh?=?cachedHoldCounter;
????????if?(rh?==?null?||?rh.tid?!=?getThreadId(current))
????????????rh?=?readHolds.get();
????????int?count?=?rh.count;
????????if?(count?<=?1)?{
????????????readHolds.remove();
????????????if?(count?<=?0)
????????????????throw?unmatchedUnlockException();
????????}
????????--rh.count;
????}
????for?(;;)?{
????????int?c?=?getState();
????????//?讀鎖釋放?將同步狀態(tài)減去讀狀態(tài)即可
????????int?nextc?=?c?-?SHARED_UNIT;
????????if?(compareAndSetState(c,?nextc))
????????????//?Releasing?the?read?lock?has?no?effect?on?readers,
????????????//?but?it?may?allow?waiting?writers?to?proceed?if
????????????//?both?read?and?write?locks?are?now?free.
????????????return?nextc?==?0;
????}
}

4.鎖降級(jí)

讀寫鎖支持鎖降級(jí)，遵循按照獲取寫鎖，獲取讀鎖再釋放寫鎖的次序，寫鎖能夠降級(jí)成為讀鎖，不支持鎖升級(jí)，關(guān)于鎖降級(jí)下面的示例代碼摘自ReentrantWriteReadLock源碼中：

void?processCachedData()?{
????????rwl.readLock().lock();
????????if?(!cacheValid)?{
????????????//?Must?release?read?lock?before?acquiring?write?lock
????????????rwl.readLock().unlock();
????????????rwl.writeLock().lock();
????????????try?{
????????????????//?Recheck?state?because?another?thread?might?have
????????????????//?acquired?write?lock?and?changed?state?before?we?did.
????????????????if?(!cacheValid)?{
????????????????????data?=?...
????????????cacheValid?=?true;
??????????}
??????????//?Downgrade?by?acquiring?read?lock?before?releasing?write?lock
??????????rwl.readLock().lock();
????????}?finally?{
??????????rwl.writeLock().unlock();?//?Unlock?write,?still?hold?read
????????}
??????}

??????try?{
????????use(data);
??????}?finally?{
????????rwl.readLock().unlock();
??????}
????}
}