詳解Java ReentrantReadWriteLock讀寫鎖的原理與實現
概述
ReentrantReadWriteLock讀寫鎖是使用AQS的集大成者，用瞭獨占模式和共享模式。本文和大傢一起理解下ReentrantReadWriteLock讀寫鎖的實現原理。在這之前建議大傢閱讀下下面3篇關聯文章：
深入淺出理解Java並發AQS的獨占鎖模式
深入淺出理解Java並發AQS的共享鎖模式
通俗易懂讀寫鎖ReentrantReadWriteLock的使用
原理概述
上圖是ReentrantReadWriteLock讀寫鎖的類結構圖：
- 實現瞭
ReadWriteLock接口，該接口提供瞭獲取讀鎖和寫鎖的API。 ReentrantReadWriteLock讀寫鎖內部的成員變量readLock是讀鎖，指向內部類ReadLock。ReentrantReadWriteLock讀寫鎖內部的成員變量writeLock是寫鎖，指向內部類WriteLock。ReentrantReadWriteLock讀寫鎖內部的成員變量sync是繼承AQS的同步器，他有兩個子類FairSync公平同步器和NoFairSync非公平同步器，讀寫鎖內部也有一個sync，他們使用的是同一個sync。
讀寫鎖用的同一個sync同步器，那麼他們共享同一個state, 這樣不會混淆嗎？
不會，ReentrantReadWriteLock讀寫鎖使用瞭AQS中state值得低16位表示寫鎖得計數，用高16位表示讀鎖得計數，這樣就可以使用同一個AQS同時管理讀鎖和寫鎖。
1.ReentrantReadWriteLock類重要成員變量
// 讀鎖 private final ReentrantReadWriteLock.ReadLock readerLock; // 寫鎖 private final ReentrantReadWriteLock.WriteLock writerLock; // 同步器 final Sync sync;
2.ReentrantReadWriteLock構造方法
//默認是非公平鎖，可以指定參數創建公平鎖
public ReentrantReadWriteLock(boolean fair) {
// true 為公平鎖
sync = fair ? new FairSync() : new NonfairSync();
// 這兩個 lock 共享同一個 sync 實例，都是由 ReentrantReadWriteLock 的 sync 提供同步實現
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個1，代表寫鎖的最大重入次數
static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1;
// 低16位掩碼：0b 1111 1111 1111 1111，用來獲取寫鎖重入的次數
static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
// 獲取讀寫鎖的讀鎖分配的總次數
static int sharedCount(int c) { return c >>> SHARED_SHIFT; }
// 寫鎖（獨占）鎖的重入次數
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
加鎖原理
圖解過程
設計一個加鎖場景，t1線程加寫鎖，t2線程加讀鎖，我們看下它們整個加鎖得流程。
1.t1 加寫鎖w.lock()成功，占瞭 state 的低 16 位。
- 這裡得state分為兩部分
0_1,0表示高16位的值，1表示低16位的值。 - AQS當前占用線程
exclusiveOwnerThread屬性指向t1線程。
2.t2線程執行加讀鎖 r.lock()，嘗試獲取鎖，發現已經被寫鎖占據瞭，加鎖失敗。
3.t2線程被封裝成一個共享模式Node.SHARED的節點，加入到AQS的隊列中。
4.在阻塞前，t2線程發現自己是隊列中的老二，會嘗試再次獲取讀鎖，因為t1沒有釋放，它會失敗，然後它會把隊列的前驅節點的狀態改為-1，然後阻塞自身，也就是t2線程。
- 上面中黃色三角形就是等待狀態的值，前驅節點變成-1
- 上面中的灰色表示節點所在的線程阻塞瞭
5.後面如過有其他線程如t3,t4加讀鎖或者寫鎖，由於t1線程沒有釋放鎖，會變成下面的狀態。
上面是整個解鎖的流程，下面深入源碼驗證這個流程。
源碼解析
1.寫鎖加鎖源碼
WriteLock類的lock()方法是加寫鎖的入口方法。
static final class NonfairSync extends Sync {
// ... 省略無關代碼
// 外部類 WriteLock 方法, 方便閱讀, 放在此處
public void lock() {
sync.acquire(1);
}
// AQS 繼承過來的方法, 方便閱讀, 放在此處
public final void acquire(int arg) {
if (
// 嘗試獲得寫鎖失敗
!tryAcquire(arg) &&
// 將當前線程關聯到一個 Node 對象上, 模式為獨占模式
// 進入 AQS 隊列阻塞
acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
) {
selfInterrupt();
}
}
protected final boolean tryAcquire(int acquires) {
// 獲取當前線程
Thread current = Thread.currentThread();
//獲得鎖的狀態
int c = getState();
// 獲得低 16 位, 代表寫鎖的 state 計數
int w = exclusiveCount(c);
// c不等於0表示加瞭讀鎖或者寫鎖
if (c != 0) {
if (
// c != 0 and w == 0 表示有讀鎖返回錯誤，讀鎖不支持鎖升級, 或者
w == 0 ||
// w != 0 說明有寫鎖，寫鎖的擁有者不是自己，獲取失敗
current != getExclusiveOwnerThread()
) {
// 獲得鎖失敗
return false;
}
// 寫鎖計數超過低 16 位最大數量, 報異常
if (w + exclusiveCount(acquires) > MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// 寫鎖重入, 獲得鎖成功，沒有並發，所以不使用 CAS
setState(c + acquires);
return true;
}
if (
// c == 0，說明沒有任何鎖，判斷寫鎖是否該阻塞，是 false 就嘗試獲取鎖，失敗返回 false
writerShouldBlock() ||
// 嘗試更改計數失敗
!compareAndSetState(c, c + acquires)
) {
// 獲得鎖失敗
return false;
}
// 獲得鎖成功，設置鎖的持有線程為當前線程
setExclusiveOwnerThread(current);
return true;
}
// 非公平鎖 writerShouldBlock 總是返回 false, 無需阻塞
final boolean writerShouldBlock() {
return false;
}
// 公平鎖會檢查 AQS 隊列中是否有前驅節點, 沒有(false)才去競爭
final boolean writerShouldBlock() {
return hasQueuedPredecessors();
}
}
tryAcquire()方法是模板方法，由子類自定義實現獲取鎖的邏輯。- 線程如果獲取寫鎖失敗的話，通過
acquireQueued()方法封裝成獨占Node加入到AQS隊列中。
2.讀鎖加鎖源碼
ReadLock類的lock()方法是加讀鎖的入口方法，調用tryAcquireShared()方法嘗試獲取讀鎖，返回負數，失敗，加入到隊列中。
// 加讀鎖的方法入口
public void lock() {
sync.acquireShared(1);
}
public final void acquireShared(int arg) {
// tryAcquireShared 返回負數, 表示獲取讀鎖失敗，加入到隊列中
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
tryAcquireShared()方法是一個模板方法，AQS類中定義語義，子類實現，如果返回1，表示獲取鎖成功，還有剩餘資源，返回0表示獲取成功，沒有剩餘資源，返回-1表示失敗。
// 嘗試以共享模式獲取，返回1表示獲取鎖成功，還有剩餘資源,返回0表示獲取成功，沒有剩餘資源，返回-1，表示失敗
protected final int tryAcquireShared(int unused) {
Thread current = Thread.currentThread();
int c = getState();
// exclusiveCount(c) 代表低 16 位, 寫鎖的 state，成立說明有線程持有寫鎖
// 寫鎖的持有者不是當前線程，則獲取讀鎖失敗，【寫鎖允許降級】
if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current)
return -1;
// 高 16 位，代表讀鎖的 state，共享鎖分配出去的總次數
int r = sharedCount(c);
// 讀鎖是否應該阻塞
if (!readerShouldBlock() && r < MAX_COUNT &&
compareAndSetState(c, c + SHARED_UNIT)) { // 嘗試增加讀鎖計數
// 加鎖成功
// 加鎖之前讀鎖為 0，說明當前線程是第一個讀鎖線程
if (r == 0) {
firstReader = current;
firstReaderHoldCount = 1;
// 第一個讀鎖線程是自己就發生瞭讀鎖重入
} else if (firstReader == current) {
firstReaderHoldCount++;
} else {
// cachedHoldCounter 設置為當前線程的 holdCounter 對象，即最後一個獲取讀鎖的線程
HoldCounter rh = cachedHoldCounter;
// 說明還沒設置 rh
if (rh == null || rh.tid != getThreadId(current))
// 獲取當前線程的鎖重入的對象，賦值給 cachedHoldCounter
cachedHoldCounter = rh = readHolds.get();
// 還沒重入
else if (rh.count == 0)
readHolds.set(rh);
// 重入 + 1
rh.count++;
}
// 讀鎖加鎖成功
return 1;
}
// 邏輯到這 應該阻塞，或者 cas 加鎖失敗
// 會不斷嘗試 for (;;) 獲取讀鎖, 執行過程中無阻塞
return fullTryAcquireShared(current);
}
// 非公平鎖 readerShouldBlock 偏向寫鎖一些，看 AQS 阻塞隊列中第一個節點是否是寫鎖，是則阻塞，反之不阻塞
// 防止一直有讀鎖線程，導致寫鎖線程饑餓
// true 則該阻塞, false 則不阻塞
final boolean readerShouldBlock() {
return apparentlyFirstQueuedIsExclusive();
}
// 下面是公平鎖的readerShouldBlock
// 公平鎖會檢查 AQS 隊列中是否有前驅節點, 沒有(false)才去競爭
final boolean readerShouldBlock() {
return hasQueuedPredecessors();
}
fullTryAcquireShared()方法是通過自旋的方式不斷獲取讀鎖，因為由於前面的readerShouldBlock返回false或者cas失敗，導致沒有獲取到鎖，需要不斷重試。
final int fullTryAcquireShared(Thread current) {
// 當前讀鎖線程持有的讀鎖次數對象
HoldCounter rh = null;
for (;;) {
int c = getState();
// 說明有線程持有寫鎖
if (exclusiveCount(c) != 0) {
// 寫鎖不是自己則獲取鎖失敗
if (getExclusiveOwnerThread() != current)
return -1;
} else if (readerShouldBlock()) {
// 條件成立說明當前線程是 firstReader，當前鎖是讀忙碌狀態，而且當前線程也是讀鎖重入
if (firstReader == current) {
// assert firstReaderHoldCount > 0;
} else {
if (rh == null) {
// 最後一個讀鎖的 HoldCounter
rh = cachedHoldCounter;
// 說明當前線程也不是最後一個讀鎖
if (rh == null || rh.tid != getThreadId(current)) {
// 獲取當前線程的 HoldCounter
rh = readHolds.get();
// 條件成立說明 HoldCounter 對象是上一步代碼新建的
// 當前線程不是鎖重入，在 readerShouldBlock() 返回 true 時需要去排隊
if (rh.count == 0)
// 防止內存泄漏
readHolds.remove();
}
}
if (rh.count == 0)
return -1;
}
}
// 越界判斷
if (sharedCount(c) == MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// 讀鎖加鎖，條件內的邏輯與 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隊列的邏輯。
private void doAcquireShared(int arg) {
// 將當前線程關聯到一個 Node 對象上, 模式為共享模式
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
// 獲取前驅節點
final Node p = node.predecessor();
// 如果前驅節點就頭節點就去嘗試獲取鎖
if (p == head) {
// 再一次嘗試獲取讀鎖
int r = tryAcquireShared(arg);
// r >= 0 表示獲取成功
if (r >= 0) {
//【這裡會設置自己為頭節點，喚醒相連的後序的共享節點】
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
// 是否在獲取讀鎖失敗時阻塞 park 當前線程
if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
setHeadAndPropagate()方法是在後續讀鎖被喚醒後，搶到鎖要處理的邏輯，包括修改隊列的頭結點，以及喚醒隊列中的下一個共享節點。
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head;
// 設置自己為 head 節點
setHead(node);
// propagate 表示有共享資源（例如共享讀鎖或信號量），為 0 就沒有資源
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
// 獲取下一個節點
Node s = node.next;
// 如果當前是最後一個節點，或者下一個節點是【等待共享讀鎖的節點】
if (s == null || s.isShared())
// 喚醒後繼節點
doReleaseShared();
}
}
解鎖原理
圖解過程
由於上面t1線程加的寫鎖，所有其他的線程都被阻塞瞭，隻有在t1線程解鎖以後，其他線程才能被喚醒，我們現在看下t1線程被喚醒瞭，會發生什麼？
1.t1線程執行解鎖w.unlock()成功，修改AQS中的state。
- 這裡的state變為瞭0_0。
- AQS當前占用線程exclusiveOwnerThread屬性變為null。
2.t1線程喚醒隊列中等待的老二， 為什麼不是老大，因為老大是一個空節點，不會設置任何的線程。t2線程被喚醒後，搶鎖成功，修改state中高16位為1。
- 老二的線程節點變為藍色節點
- AQS中的state變為1_0。
3.t2線程恢復運行，設置原來的老二節點為頭節點
4.t2線程要做的事情還沒結束呢，因為是共享模式，它現在釋放瞭，就此時也喚醒隊列中的下一個共享節點。
5.t3線程恢復去競爭讀鎖成功，這時state的高位+1，變成2。
6.這時候t3線程所在的Node設置為頭節點，同時發現對列的下一個節點不是共享節點，而是獨占節點，就不會喚醒後面的節點瞭。
7.之後t2線程和t3線程進入尾聲，執行r.unlock操作，state的計數減一，直到變為0。
8.最後寫鎖線程t4被喚醒，去搶占鎖成功，整個流程結束。
上面是整個解鎖的流程，下面深入源碼驗證這個流程。
源碼解析
1.寫鎖釋放流程
WriteLock類的unlock()方法是入口方法，調用tryRelease()方法釋放鎖，如果成功，調用unparkSuccessor()方法喚醒線程。
public void unlock() {
// 釋放鎖
sync.release(1);
}
public final boolean release(int arg) {
// 嘗試釋放鎖
if (tryRelease(arg)) {
Node h = head;
// 頭節點不為空並且不是等待狀態不是 0，喚醒後繼的非取消節點
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
tryRelease()方法是AQS提供的模板方法，返回true表示成功，false失敗，由自定義同步器實現。
protected final boolean tryRelease(int releases) {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
int nextc = getState() - releases;
// 因為可重入的原因, 寫鎖計數為 0, 才算釋放成功
boolean free = exclusiveCount(nextc) == 0;
if (free)
// 設置占用線程為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提供的模板方法，由自定義同步器實現。
protected final boolean tryReleaseShared(int unused) {
//自選
for (;;) {
int c = getState();
int nextc = c - SHARED_UNIT;
// 讀鎖的計數不會影響其它獲取讀鎖線程, 但會影響其它獲取寫鎖線程，計數為 0 才是真正釋放
if (compareAndSetState(c, nextc))
// 返回是否已經完全釋放瞭
return nextc == 0;
}
}
調用doReleaseShared()方法喚醒等待的線程，這個方法調用的地方有兩處，還記得嗎，一個這是裡的解鎖，還有一個是前面加共享鎖阻塞的地方，喚醒後獲取鎖成功，也會調用doReleaseShared()方法。
private void doReleaseShared() {
// 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一個節點 unpark
// 如果 head.waitStatus == 0 ==> Node.PROPAGATE
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
// SIGNAL 喚醒後繼
if (ws == Node.SIGNAL) {
// 因為讀鎖共享，如果其它線程也在釋放讀鎖，那麼需要將 waitStatus 先改為 0
// 防止 unparkSuccessor 被多次執行
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue;
// 喚醒後繼節點
unparkSuccessor(h);
}
// 如果已經是 0 瞭，改為 -3，用來解決傳播性
else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue;
}
// 條件不成立說明被喚醒的節點非常積極，直接將自己設置為瞭新的 head，
// 此時喚醒它的節點（前驅）執行 h == head 不成立，所以不會跳出循環，會繼續喚醒新的 head 節點的後繼節點
if (h == head)
break;
}
}
以上就是詳解Java ReentrantReadWriteLock讀寫鎖的原理與實現的詳細內容，更多關於Java ReentrantReadWriteLock讀寫鎖的資料請關註WalkonNet其它相關文章！
推薦閱讀：
- Java多線程之ReentrantReadWriteLock源碼解析
- Java 多線程並發 ReentrantReadWriteLock詳情
- Java多線程讀寫鎖ReentrantReadWriteLock類詳解
- Java並發編程之淺談ReentrantLock
- ReentrantLock從源碼解析Java多線程同步學習