概述
ReentrantReadWriteLock讀寫鎖是使用AQS的集大成者，用了獨(dú)占模式和共享模式。本文和大家一起理解下ReentrantReadWriteLock讀寫鎖的實(shí)現(xiàn)原理。
原理概述
上圖是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的同步器，他有兩個子類FairSync公平同步器和NoFairSync非公平同步器，讀寫鎖內(nèi)部也有一個sync，他們使用的是同一個sync。
讀寫鎖用的同一個sync同步器，那么他們共享同一個state, 這樣不會混淆嗎？
不會，ReentrantReadWriteLock讀寫鎖使用了AQS中state值得低16位表示寫鎖得計(jì)數(shù)，用高16位表示讀鎖得計(jì)數(shù)，這樣就可以使用同一個AQS同時管理讀鎖和寫鎖。
- ReentrantReadWriteLock類重要成員變量
// 讀鎖
private final ReentrantReadWriteLock.ReadLock readerLock;
// 寫鎖
private final ReentrantReadWriteLock.WriteLock writerLock;
// 同步器
final Sync sync;
- ReentrantReadWriteLock構(gòu)造方法
//默認(rèn)是非公平鎖，可以指定參數(shù)創(chuàng)建公平鎖
public ReentrantReadWriteLock(boolean fair) {
// true 為公平鎖
sync = fair ? new FairSync() : new NonfairSync();
// 這兩個 lock 共享同一個 sync 實(shí)例，都是由 ReentrantReadWriteLock 的 sync 提供同步實(shí)現(xiàn)
readerLock = new ReadLock(this);
writerLock = new WriteLock(this);
}
- Sync類重要成員變量
// 用來移位
static final int SHARED_SHIFT = 16;
// 高16位的1
static final int SHARED_UNIT = (1 << SHARED_SHIFT);
// 65535，16個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ì)一個加鎖場景，t1線程加寫鎖，t2線程加讀鎖，我們看下它們整個加鎖得流程。
- t1 加寫鎖w.lock()成功，占了 state 的低 16 位。
- 這里得state分為兩部分0_1,0表示高16位的值，1表示低16位的值。
- AQS當(dāng)前占用線程exclusiveOwnerThread屬性指向t1線程。
- t2線程執(zhí)行加讀鎖r.lock()，嘗試獲取鎖，發(fā)現(xiàn)已經(jīng)被寫鎖占據(jù)了，加鎖失敗。
- t2線程被封裝成一個共享模式Node.SHARED的節(jié)點(diǎn)，加入到AQS的隊(duì)列中。
- 在阻塞前，t2線程發(fā)現(xiàn)自己是隊(duì)列中的老二，會嘗試再次獲取讀鎖，因?yàn)閠1沒有釋放，它會失敗，然后它會把隊(duì)列的前驅(qū)節(jié)點(diǎn)的狀態(tài)改為-1，然后阻塞自身，也就是t2線程。
- 上面中黃色三角形就是等待狀態(tài)的值，前驅(qū)節(jié)點(diǎn)變成-1
- 上面中的灰色表示節(jié)點(diǎn)所在的線程阻塞了
- 后面如過有其他線程如t3,t4加讀鎖或者寫鎖，由于t1線程沒有釋放鎖，會變成下面的狀態(tài)。
上面是整個解鎖的流程，下面深入源碼驗(yàn)證這個流程。
源碼解析
寫鎖加鎖源碼
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)到一個 Node 對象上, 模式為獨(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 表示有讀鎖返回錯誤，讀鎖不支持鎖升級, 或者
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;
}
// 公平鎖會檢查 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()方法是一個模板方法，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)前線程，則獲取讀鎖失敗，【寫鎖允許降級】
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)前線程是第一個讀鎖線程
if (r == 0) {
firstReader = current;
firstReaderHoldCount = 1;
// 第一個讀鎖線程是自己就發(fā)生了讀鎖重入
} else if (firstReader == current) {
firstReaderHoldCount++;
} else {
// cachedHoldCounter 設(shè)置為當(dāng)前線程的 holdCounter 對象，即最后一個獲取讀鎖的線程
HoldCounter rh = cachedHoldCounter;
// 說明還沒設(shè)置 rh
if (rh == null || rh.tid != getThreadId(current))
// 獲取當(dāng)前線程的鎖重入的對象，賦值給 cachedHoldCounter
cachedHoldCounter = rh = readHolds.get();
// 還沒重入
else if (rh.count == 0)
readHolds.set(rh);
// 重入 + 1
rh.count++;
}
// 讀鎖加鎖成功
return 1;
}
// 邏輯到這 應(yīng)該阻塞，或者 cas 加鎖失敗
// 會不斷嘗試 for (;;) 獲取讀鎖, 執(zhí)行過程中無阻塞
return fullTryAcquireShared(current);
}
// 非公平鎖 readerShouldBlock 偏向?qū)戞i一些，看 AQS 阻塞隊(duì)列中第一個節(jié)點(diǎn)是否是寫鎖，是則阻塞，反之不阻塞
// 防止一直有讀鎖線程，導(dǎo)致寫鎖線程饑餓
// true 則該阻塞, false 則不阻塞
final boolean readerShouldBlock() {
return apparentlyFirstQueuedIsExclusive();
}
// 下面是公平鎖的readerShouldBlock
// 公平鎖會檢查 AQS 隊(duì)列中是否有前驅(qū)節(jié)點(diǎn), 沒有(false)才去競爭
final boolean readerShouldBlock() {
return hasQueuedPredecessors();
}
- fullTryAcquireShared()方法是通過自旋的方式不斷獲取讀鎖，因?yàn)橛捎谇懊娴膔eaderShouldBlock返回false或者cas失敗，導(dǎo)致沒有獲取到鎖，需要不斷重試。
final int fullTryAcquireShared(Thread current) {
// 當(dāng)前讀鎖線程持有的讀鎖次數(shù)對象
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) {
// 最后一個讀鎖的 HoldCounter
rh = cachedHoldCounter;
// 說明當(dāng)前線程也不是最后一個讀鎖
if (rh == null || rh.tid != getThreadId(current)) {
// 獲取當(dāng)前線程的 HoldCounter
rh = readHolds.get();
// 條件成立說明 HoldCounter 對象是上一步代碼新建的
// 當(dāng)前線程不是鎖重入，在 readerShouldBlock() 返回 true 時需要去排隊(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()是在獲取讀鎖失敗的時候加入AQS隊(duì)列的邏輯。
private void doAcquireShared(int arg) {
// 將當(dāng)前線程關(guān)聯(lián)到一個 Node 對象上, 模式為共享模式
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) {
//【這里會設(shè)置自己為頭節(jié)點(diǎn)，喚醒相連的后序的共享節(jié)點(diǎn)】
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
// 是否在獲取讀鎖失敗時阻塞 park 當(dāng)前線程
if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}- setHeadAndPropagate()方法是在后續(xù)讀鎖被喚醒后，搶到鎖要處理的邏輯，包括修改隊(duì)列的頭結(jié)點(diǎn)，以及喚醒隊(duì)列中的下一個共享節(jié)點(diǎn)。
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head;
// 設(shè)置自己為 head 節(jié)點(diǎn)
setHead(node);
// propagate 表示有共享資源（例如共享讀鎖或信號量），為 0 就沒有資源
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
// 獲取下一個節(jié)點(diǎn)
Node s = node.next;
// 如果當(dāng)前是最后一個節(jié)點(diǎn)，或者下一個節(jié)點(diǎn)是【等待共享讀鎖的節(jié)點(diǎn)】
if (s == null || s.isShared())
// 喚醒后繼節(jié)點(diǎn)
doReleaseShared();
}
}解鎖原理
圖解過程
由于上面t1線程加的寫鎖，所有其他的線程都被阻塞了，只有在t1線程解鎖以后，其他線程才能被喚醒，我們現(xiàn)在看下t1線程被喚醒了，會發(fā)生什么？
t1線程執(zhí)行解鎖w.unlock()成功，修改AQS中的state。
- 這里的state變?yōu)榱?_0。
- AQS當(dāng)前占用線程exclusiveOwnerThread屬性變?yōu)閚ull。
- t1線程喚醒隊(duì)列中等待的老二， 為什么不是老大，因?yàn)槔洗笫且粋€空節(jié)點(diǎn)，不會設(shè)置任何的線程。t2線程被喚醒后，搶鎖成功，修改state中高16位為1。
- 老二的線程節(jié)點(diǎn)變?yōu)樗{(lán)色節(jié)點(diǎn)
- AQS中的state變?yōu)?_0。
- t2線程恢復(fù)運(yùn)行，設(shè)置原來的老二節(jié)點(diǎn)為頭節(jié)點(diǎn)
- t2線程要做的事情還沒結(jié)束呢，因?yàn)槭枪蚕砟Ｊ剑F(xiàn)在釋放了，就此時也喚醒隊(duì)列中的下一個共享節(jié)點(diǎn)。
- t3線程恢復(fù)去競爭讀鎖成功，這時state的高位+1，變成2。
- 這時候t3線程所在的Node設(shè)置為頭節(jié)點(diǎn)，同時發(fā)現(xiàn)對列的下一個節(jié)點(diǎn)不是共享節(jié)點(diǎn)，而是獨(dú)占節(jié)點(diǎn)，就不會喚醒后面的節(jié)點(diǎn)了。
- 之后t2線程和t3線程進(jìn)入尾聲，執(zhí)行r.unlock操作，state的計(jì)數(shù)減一，直到變?yōu)?。
- 最后寫鎖線程t4被喚醒，去搶占鎖成功，整個流程結(jié)束。
上面是整個解鎖的流程，下面深入源碼驗(yàn)證這個流程。
源碼解析
- 寫鎖釋放流程
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)榭芍厝氲脑?span>, 寫鎖計(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ù)不會影響其它獲取讀鎖線程, 但會影響其它獲取寫鎖線程，計(jì)數(shù)為 0 才是真正釋放
if (compareAndSetState(c, nextc))
// 返回是否已經(jīng)完全釋放了
return nextc == 0;
}
}調(diào)用doReleaseShared()?方法喚醒等待的線程，這個方法調(diào)用的地方有兩處，還記得嗎，一個這是里的解鎖，還有一個是前面加共享鎖阻塞的地方，喚醒后獲取鎖成功，也會調(diào)用doReleaseShared()方法。
private void doReleaseShared() {
// 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一個節(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，
// 此時喚醒它的節(jié)點(diǎn)（前驅(qū)）執(zhí)行 h == head 不成立，所以不會跳出循環(huán)，會繼續(xù)喚醒新的 head 節(jié)點(diǎn)的后繼節(jié)點(diǎn)
if (h == head)
break;
}
}總結(jié)
本文講解了讀寫鎖ReentrantReadWriteLock的整個加鎖、解鎖的實(shí)現(xiàn)原理，并從源碼的角度深入分析，希望對大家有幫助。
