Java併發程式設計-ReentrantReadWriteLock
基於AQS的前世今生，來學習併發工具類ReentrantReadWriteLock。本文將從ReentrantReadWriteLock的產生背景、原始碼原理解析和應用來學習這個併發工具類。
1、產生背景
前面我們學習的重入鎖ReentrantLock本質上還是互斥鎖，每次最多隻能有一個執行緒持有ReentrantLock。對於維護資料完整性來說，互斥通常是一種過於強硬的規則，因此也就不必要的限制了併發性。互斥是一種保守的加鎖策略，雖然可以避免“寫/寫”衝突和“寫/讀”衝突，但也同樣避免了“讀/讀”衝突。和網際網路的“二八法則”一樣，大部分資料都是讀資料，可以存放在快取中，資料結構的操作其實很多也是讀操作，可以考慮適當的放寬加鎖需求，允許多個讀操作執行緒同時訪問資料結構以提升程式的效能。在這樣的需求背景下，就產生了讀寫鎖ReadWriteLock，一個資源可以同時被多個讀操作訪問，或者被一個寫操作訪問，但是不能讀寫操作同時訪問。ReadWriteLock定義了介面規範，實際實現讀寫鎖控制的類是ReentrantReadWriteLock，該類為讀寫鎖提供了可重入的加鎖語義。
2、原始碼原理解析
2.1 讀寫鎖原理
既然是讀寫鎖，那就是有兩把鎖，可以用AQS的同步狀態表示其中的一把鎖，再引入一個新的屬性表示另外一把鎖，但是這麼做就變成了二元併發安全問題，使問題變得更加複雜。ReentrantReadWriteLock選擇了用一個屬性，即AQS的同步狀態來表示讀寫鎖，怎樣用一個屬性來表示讀寫鎖呢？那就是位運算，對位運算不熟悉的可以先看下此文。
ReentantReadWriteLock採用“按位切割”的方式，就是將這個32位的int型state變數分為高16位和低16位來使用，高16位代表讀狀態，低16位代表寫狀態 。讀鎖是可以共享的，而寫鎖是互斥的，對於寫鎖而言，用低16位表示執行緒的重入次數，但是讀鎖因為可以同時有多個執行緒，所以重入次數需要通過其他的方式來記錄，那就是ThreadLocal變數。從這也可以總結出來和ReentrantLock相比，寫鎖的重入次數會減少，最多不能超過65535次。讀鎖的執行緒數也有限制，最對不能超過65535個。
假設狀態變數是c，則讀狀態就是c>>>16（無符號右移16位），其實就是通過無符號右移運算抹掉低的16位，剩下的就是c的高16位。寫狀態是c&((1 << 16) - 1)，其實就是c&00000000000000001111111111111111，與運算之後，高的16位被抹掉，剩下的就是c的低16位。如果讀執行緒申請讀鎖，當前寫鎖重入次數不為 0 時，則等待，否則可以馬上分配；如果是寫執行緒申請寫鎖，當前狀態為 0 則可以馬上分配，否則等待。
2.2 讀鎖的獲取和釋放
讀鎖的獲取方法如下：
protected final int tryAcquireShared(int unused) {
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) {//讀鎖的第一個執行緒 此時可以不用記錄到ThreadLocal
firstReader = current;
firstReaderHoldCount = 1; //避免查詢ThreadLocal 提升效率
} 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); //獲取讀鎖失敗 迴圈重試
}
final int fullTryAcquireShared(Thread current) {
HoldCounter rh = null;
for (;;) {
int c = getState();
if (exclusiveCount(c) != 0) {//獲取到寫鎖
if (getExclusiveOwnerThread() != current)
return -1; //非寫鎖執行緒獲取失敗
// else we hold the exclusive lock; blocking here
// would cause deadlock.
} else if (readerShouldBlock()) {
// 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 != getThreadId(current)) {
rh = readHolds.get();
if (rh.count == 0)
readHolds.remove();
}
}
if (rh.count == 0)
return -1;
}
}
if (sharedCount(c) == MAX_COUNT) //讀鎖數量達到最大
throw new Error("Maximum lock count exceeded");
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;
}
}
}
讀鎖的釋放方法如下：
protected final boolean tryReleaseShared(int unused) {
Thread current = Thread.currentThread();
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();
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;
}
}
2.3 寫鎖的獲取和釋放
寫鎖的獲取方法如下：
protected final boolean tryAcquire(int acquires) {
Thread current = Thread.currentThread();
int c = getState();
int w = exclusiveCount(c);//寫鎖狀態
if (c != 0) {//表示鎖已經被分配出去了 if c != 0 and w == 0表示獲取讀鎖
// (Note: if c != 0 and w == 0 then shared count != 0)
//其他執行緒獲取到了寫鎖
if (w == 0 || current != getExclusiveOwnerThread())
return false;
//寫鎖重入次數超過最大值
if (w + exclusiveCount(acquires) > MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// Reentrant acquire更新寫鎖重入次數
setState(c + acquires);
return true;
}
if (writerShouldBlock() ||//子類實現寫鎖是否公平獲取
!compareAndSetState(c, c + acquires))
return false;//cas獲取寫鎖失敗
setExclusiveOwnerThread(current);//獲取寫鎖成功 獨佔
return true;
}
寫鎖的釋放方法如下：
protected final boolean tryRelease(int releases) {
if (!isHeldExclusively())//當前執行緒不持有寫鎖
throw new IllegalMonitorStateException();
int nextc = getState() - releases; //重入次數減少
boolean free = exclusiveCount(nextc) == 0; //減少到0寫鎖釋放
if (free)
setExclusiveOwnerThread(null); //寫鎖釋放
setState(nextc);
return free;
}
2.4 鎖降級
鎖降級指的是寫鎖降級為讀鎖，首先持有當前寫鎖，然後獲取到讀鎖，在tryAcquireShared方法中已經體現了該過程，隨後再釋放該寫鎖的過程。鎖降級主要是為了保持資料的可見性，如果當前執行緒不獲取讀鎖而是直接釋放寫鎖，假設此時有另外的執行緒獲取到了寫鎖並修改了資料，那麼當前執行緒是無法知曉資料已經更新了。如果當前執行緒遵循鎖降級的過程，則其他執行緒會被阻塞，直到當前執行緒操作完成其他執行緒才可以獲取寫鎖進行資料更新。RentrantReadWriteLock不支援鎖升級（把持讀鎖、獲取寫鎖，最後釋放讀鎖的過程）。目的也是保證資料可見性，如果讀鎖已被多個執行緒獲取，其中任意執行緒成功獲取了寫鎖並更新了資料，則其更新對其他獲取到讀鎖的執行緒是不可見的。
3、應用
概況性的總結RentrantReadWriteLock的應用，就是ReentrantLock能使用的地方，RentrantReadWriteLock都能使用，而且能提供更好的吞吐率。
參考資料:
ofollow,noindex" target="_blank">https://github.com/lingjiango/ConcurrentProgramPractice