一，AQS源碼博文：并發程式設計：AbstractQueuedSynchronizer源碼分析
二，ReentrantReadWriteLock讀寫鎖介紹
1，讀寫鎖介紹
ReentrantReadWriteLock 雖然與 ReentrantLock 沒有直接關系，但是在功能上算是對 ReentrantLock 的擴充。在 ReentrantLock 重入獨占鎖的功能上，添加了共享鎖的擴充，分别對應 ReentrantReadWriteLock 的寫鎖（WriteLock）和讀鎖（ReadLock）。ReentrantReadWriteLock 内部定義 Sync 類繼承自 AQS 對加鎖方式進行擴充，在讀鎖寫鎖擷取以及讀寫鎖交叉擷取提供了自身機制，以此保證線程同步。ReentrantReadWriteLock 在加鎖過程中，對 Integer 的32位進行分割，以高16位表示共享鎖，以低16位表示獨占鎖。在進行加鎖時，通過左移右移以及與運算判斷目前加鎖狀态及重入狀态，并進線程進行排程。
2，類圖
* ReentrantReadWriteLock 實作自 ReadWriteLock 接口，并在内部定義了鎖實作了相關内部類。
* Sync，FairSync，NonfairSync 三個内部類是對 AQS 的層次擴充，用來擴充加鎖的一系列邏輯處理
* WriteLock，ReadLock 兩個内部類實作自 Lock 接口，并通過 ReentrantReadWriteLock 建立後對外提供，在應用層進行不同的加鎖處理
3，常用API
// 建立讀寫鎖, 通過參數指定公平鎖/非公平鎖
public ReentrantReadWriteLock();
public ReentrantReadWriteLock(boolean fair);
// 建立讀鎖
public ReentrantReadWriteLock.ReadLock readLock()
// 建立寫鎖
public ReentrantReadWriteLock.WriteLock writeLock();
// 嘗試加鎖
public boolean tryLock();
public boolean tryLock(long timeout, TimeUnit unit);
// 加鎖
public void lock();
public void lockInterruptibly() throws InterruptedException;
// 釋放鎖
public void unlock();
* 在上述API中，關于鎖操作的一些API，都存在讀鎖和寫鎖的差別；其中，在讀鎖和寫鎖的進行中，又存在公平鎖和非公平鎖的區分；
* 後續源碼分析中，讀鎖和寫鎖會分别分析，但是公平鎖和分公平鎖隻對非公平鎖進行分析，公平鎖參考重入鎖
4，ReentrantReadWriteLock 加鎖分析
4.1，整體鎖狀态存儲
* ReentrantReadWriteLock 在鎖狀态存儲中，對 Integer 32位進行拆分，使用高16位存儲共享鎖，使用低16位存儲獨占鎖。是以無論是共享鎖的共享次數還是獨占鎖的重入次數，最高為（(1 << 16) - 1）。
// 1 << 16就是1在二進制情況下右移16位結果為
10000 0000 0000 0000 = 2^16 = 65536
// (1 << 16) - 1也是在二進制下進行減操作
1111 1111 1111 1111 = 2^16 - 1 = 65535
* ReentrantReadWriteLock 在獨占鎖加鎖時，是對低16位進行加1，在擷取獨占鎖的重入次數時，也是對低16位進行與運算以擷取結果加鎖具體看代碼框
// 獨占鎖state遞增
compareAndSetState(c, c + 1)
// 獨占鎖擷取重入次數
// EXCLUSIVE_MASK：常量，最終結果為 (1111 1111 1111 1111)，也就是對應 Integer 的低16位
// 用c和 EXCLUSIVE_MASK 進行與運算，也就是用c的低16位和 EXCLUSIVE_MASK 進行與運算，
// 與運算同1為1，其餘為0，是以擷取到的結果就是c的低16位表示的數字，然後再預設轉換10進制，即為重入數量
static int exclusiveCount(int c) {
return c & EXCLUSIVE_MASK;
}
static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
static final int SHARED_SHIFT = 16;
// 比如，此時共享為3次，重入為0次，則c的二進制表示為
0000 0000 0000 0011 0000 0000 0000 0000
// EXCLUSIVE_MASK的二進制表示為
0000 0000 0000 0000 1111 1111 1111 1111
// 運算結果
0000 0000 0000 0011 0000 0000 0000 0000
& 0000 0000 0000 0000 1111 1111 1111 1111
-------------------------------------------
0000 0000 0000 0000 0000 0000 0000 0000 = 0
// 是以此時擷取到的獨占鎖數量為0
// 假設可以擷取成功，則對獨占鎖遞增，通過CAS，用c + 1代替 c，此時，c的二進制表示為：
0000 0000 0000 0011 0000 0000 0000 0001
// 這時候繼續計算，擷取獨占鎖的次數，運算如下
0000 0000 0000 0011 0000 0000 0000 0001
& 0000 0000 0000 0000 1111 1111 1111 1111
-------------------------------------------
0000 0000 0000 0000 0000 0000 0000 0001 = 1
* 同樣，ReentrantReadWriteLock 在對共享鎖加鎖時，是對高16位進行加鎖，在擷取共享鎖次數時，先對 state 無符号右移16位，然後直接擷取結果，具體過程如代碼框
// 加鎖成功後，c遞增，按照重入邏輯，遞增應該為1
// 而此時遞增了(1 << SHARED_SHIFT)，也就是二進制下 1 0000 0000 0000 0000
// 除去低16位，也就是剛好在高16位上遞增1
compareAndSetState(c, c + SHARED_UNIT)
static final int SHARED_UNIT = (1 << SHARED_SHIFT);
static final int SHARED_SHIFT = 16;
// 共享鎖擷取鎖重入次數
// 直接将state無符号右移16位，表示的就是共享鎖的重入次數
// 而右移去掉的16位，也就剛好是獨占鎖表示的低16位
static int sharedCount(int c) {
return c >>> SHARED_SHIFT;
}
static final int SHARED_SHIFT = 16;
// 繼續舉例，比如此時共享次數3，獨占次數為3，則state的二進制表示為
0000 0000 0000 0011 0000 0000 0000 0011
// 無符号右移16位後，高位補0，值如下
0000 0000 0000 0000 0000 0000 0000 0011 = 3
// 假設此時共享鎖擷取成功，state加上(1 << SHARED_SHIFT)，如下
0000 0000 0000 0011 0000 0000 0000 0011
+ 0000 0000 0000 0001 0000 0000 0000 0000
-------------------------------------------
0000 0000 0000 0100 0000 0000 0000 0000
// 此時繼續擷取數量，對state無符号右移16位，結果如下
0000 0000 0000 0000 0000 0000 0000 0100 = 4
5，功能DEMO
package com.gupao.concurrent;
import java.util.concurrent.locks.ReentrantReadWriteLock;
/**
* @author pj_zhang
* @create 2019-10-19 12:37
**/
public class ReadAndWriteLockTest {
private static ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();
private static ReentrantReadWriteLock.ReadLock readLock = reentrantReadWriteLock.readLock();
private static ReentrantReadWriteLock.WriteLock writeLock = reentrantReadWriteLock.writeLock();
public static void main(String[] args) throws InterruptedException {
testReadAndWriteLock();
}
public static void testWriteAndReadLock() throws InterruptedException {
new Thread(() -> {
System.out.println("開始擷取鎖。。。");
// 先擷取寫鎖後擷取讀鎖
// 讀鎖會順利擷取
writeLock.lock();
System.out.println("擷取寫鎖成功。。。");
readLock.lock();
System.out.println("擷取讀鎖成功。。。");
writeLock.unlock();
System.out.println("釋放讀鎖成功。。。");
readLock.unlock();
System.out.println("釋放寫鎖成功");
}, "READ_AND_WRITE").start();
}
public static void testReadAndWriteLock() throws InterruptedException {
new Thread(() -> {
System.out.println("開始擷取鎖。。。");
// 先擷取讀寫再擷取寫鎖
// 此時已經存在鎖，則獨占鎖擷取失敗
readLock.lock();
System.out.println("擷取讀鎖成功。。。");
writeLock.lock();
System.out.println("擷取寫鎖成功。。。");
writeLock.unlock();
System.out.println("釋放讀鎖成功。。。");
readLock.unlock();
System.out.println("釋放寫鎖成功");
}, "READ_AND_WRITE").start();
}
/**
* 不同線程測試讀寫鎖
*
* @throws InterruptedException
*/
public static void testLockWithDiffThread() throws InterruptedException {
// 先啟動一個讀鎖
new Thread(() -> {
System.out.println(Thread.currentThread().getName() + "加讀鎖前執行。。。, time: " + System.currentTimeMillis());
readLock.lock();
System.out.println(Thread.currentThread().getName() + "加讀鎖後執行。。。, time: " + System.currentTimeMillis());
sleep(1000);
readLock.unlock();
}, "READ_1").start();
Thread.sleep(10);
// 再一個一個讀鎖，模拟讀鎖不阻塞
new Thread(() -> {
System.out.println(Thread.currentThread().getName() + "加讀鎖前執行。。。, time: " + System.currentTimeMillis());
readLock.lock();
System.out.println(Thread.currentThread().getName() + "加讀鎖後執行。。。, time: " + System.currentTimeMillis());
sleep(1000);
readLock.unlock();
}, "READ_2").start();
Thread.sleep(10);
// 啟動一個寫鎖，模拟寫鎖阻塞
new Thread(() -> {
System.out.println(Thread.currentThread().getName() + "_加寫鎖前執行。。。, time: " + System.currentTimeMillis());
writeLock.lock();
System.out.println(Thread.currentThread().getName() + "_加寫鎖後執行。。。, time: " + System.currentTimeMillis());
sleep(1000);
writeLock.unlock();
}, "WRITE").start();
Thread.sleep(10);
// 自啟動一個讀鎖，模拟共享鎖執行時，阻塞一個寫鎖，之後再擷取讀鎖，在寫鎖後執行
new Thread(() -> {
System.out.println(Thread.currentThread().getName() + "加讀鎖前執行。。。, time: " + System.currentTimeMillis());
readLock.lock();
System.out.println(Thread.currentThread().getName() + "加讀鎖後執行。。。, time: " + System.currentTimeMillis());
sleep(1000);
readLock.unlock();
}, "READ_3").start();
}
private static void sleep(long millSeconds) {
try {
Thread.sleep(millSeconds);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
* testLockWithDiffThread() 執行結果：
1）在不同線程間，讀鎖間共享，不會進行鎖競争，通過CPU排程依次執行
2）寫鎖排斥，在嘗試擷取寫鎖時，會先判斷是否存線上程持有寫鎖或者讀寫，如果存在，則添加到AQS同步隊列中等待喚醒
3）寫鎖等待後，再啟動一個讀鎖，此時因為寫鎖等待，并且已經對低16位的獨占鎖辨別進行修改，且鎖線程與目前線程不一緻。則目前讀鎖線程擷取鎖失敗，依次進入AQS隊列等待
* testReadAndWriteLock 執行結果
1）在同一線程後，嘗試擷取兩種鎖，從列印結果可以看出，線程擷取讀鎖後，再擷取寫鎖時阻塞，等待讀鎖釋放
2）線程在擷取讀鎖後，對高16進行修改，并設定鎖線程為目前線程
3）線程繼續擷取寫鎖，因為高位已經被修改，是以寫鎖擷取失敗，等待讀鎖釋放
* testWriteAndReadLock 執行結果
1）在同一線程中，嘗試擷取兩種所，從列印結果可以看出，線程擷取寫鎖後，再擷取讀鎖成功
2）線程擷取寫鎖後，對低16位進行修改，并設定鎖線程為目前線程
3）線程繼續擷取讀鎖，雖然判斷獨占線程已經存在，但是加鎖線程表示目前線程，是以線程擷取鎖成功
三，ReentrantReadWriteLock部分源碼分析
1，初始化部分
* ReentrantReadWriteLock()：調用重載構造器，并傳遞參數表示預設非公平鎖
public ReentrantReadWriteLock() {
this(false);
}
* ReentrantReadWriteLock(boolean fair)
public ReentrantReadWriteLock(boolean fair) {
// 通過入參表示公平鎖還是非公平鎖
sync = fair ? new FairSync() : new NonfairSync();
// 初始化讀寫鎖對象
readerLock = new ReadLock(this);
writerLock = new WriteLock(this);
}
* 初始化讀寫鎖
// 初始化讀鎖
protected ReadLock(ReentrantReadWriteLock lock) {
sync = lock.sync;
}
// 初始化寫鎖
protected WriteLock(ReentrantReadWriteLock lock) {
sync = lock.sync;
}
* 擷取讀寫鎖
// 擷取寫鎖
public ReentrantReadWriteLock.WriteLock writeLock() {
return writerLock;
}
// 擷取讀鎖
public ReentrantReadWriteLock.ReadLock readLock() {
return readerLock;
}
四，ReentrantReadWriteLock.ReadLock源碼分析
1，嘗試加鎖
* tryLock()：直接通過 Sync 嘗試加讀鎖
public boolean tryLock() {
return sync.tryReadLock();
}
* tryReadLock()
final boolean tryReadLock() {
// 擷取目前線程
Thread current = Thread.currentThread();
// 自旋嘗試擷取鎖，可能存線上程競争導緻的CAS失敗問題
for (;;) {
// 擷取目前的加鎖次數，注意此處次數是經過高低位處理後的次數，并不是真實次數，需要進行解析
int c = getState();
// exclusiveCount：低16位解析，擷取獨占鎖數量，判斷是否存在獨占鎖
// getExclusiveOwnerThread：判斷目前線程是否是鎖線程
// 是以如果存在獨占鎖，并且鎖線程不是目前線程，則嘗試失敗
// 否則目前步驟成功
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return false;
// 高16位解析，擷取共享鎖數量
int r = sharedCount(c);
// 共享鎖最大值判斷，
// MAX_COUNT = (1 << SHARED_SHIFT) - 1;
if (r == MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// 按照重入邏輯，如果可以加鎖，則是對state+1
// 但是共享鎖是對高16位進行操作，是以應該在高16位的基礎上加1，
// SHARED_UNIT = (1 << SHARED_SHIFT);
if (compareAndSetState(c, c + SHARED_UNIT)) {
// r為0，表示目前并沒有共享鎖進入，則初始化目前線程為頭線程
if (r == 0) {
firstReader = current;
firstReaderHoldCount = 1;
// 目前線程重入，則對重入此時遞增
} else if (firstReader == current) {
firstReaderHoldCount++;
} else {
// 此處表示擷取共享鎖的線程不是目前鎖線程
// 存在新線程嘗試擷取共享鎖，則cachedHoldCounter必定為null
HoldCounter rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current))
// readHolds內建自ThreadLocal，為每一個新線程配置設定一個HoldCounter
cachedHoldCounter = rh = readHolds.get();
// 可能存在其他途徑建立線程的HoldCounter，這樣在readHolds中可能不存在（個人了解）
else if (rh.count == 0)
readHolds.set(rh);
// 初始化完成後，對count遞增，表示線程加鎖及重入
rh.count++;
}
return true;
}
}
}
* readHolds.get()：此處涉及 ThreadLocal 的初始化部分
// ThreadLocalHoldCounter繼承自ThreadLocal
// 泛型表示ThreadLocal為每一個線程配置設定一個 HoldCounter 對象
// 目前類重寫了父類的initialValue()方法，并初始化了一個HoldCounter對象
// 表示 ThreadLocalHoldCounter 預設為每一個線程配置設定一個已經初始化好的對象
// 每一次get時，如果擷取不到線程已經存儲的資訊，則直接傳回一個新對象
static final class ThreadLocalHoldCounter extends ThreadLocal<HoldCounter> {
public HoldCounter initialValue() {
return new HoldCounter();
}
}
2，加鎖
* lock()
public void lock() {
sync.acquireShared(1);
}
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
* tryAcquireShared(int unused)
protected final int tryAcquireShared(int unused) {
Thread current = Thread.currentThread();
int c = getState();
// 存在寫鎖，并且不是目前線程擷取，則擷取鎖失敗
// 此處不排除存在寫鎖，如果存在寫鎖則寫鎖為目前線程持有
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
// readerShouldBlock()：判斷是否存在下一個節點，且下一個節點不是共享節點
int r = sharedCount(c);
if (!readerShouldBlock() &&
r < MAX_COUNT &&
compareAndSetState(c, c + SHARED_UNIT)) {
// 下面為加鎖處理
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;
}
// 繼續嘗試擷取共享鎖
return fullTryAcquireShared(current);
}
* fullTryAcquireShared(Thread current)
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) {
} else {
// 此處判斷目前線程是否在執行中，rh.count表示重入
// 如果在執行中，則繼續可以擷取執行，如果不在執行中，則擷取失敗
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;
}
}
}
* readerShouldBlock()：存在下一個節點，并且下一個節點不是共享節點，傳回true，否則傳回false
final boolean readerShouldBlock() {
return apparentlyFirstQueuedIsExclusive();
}
// 此處判斷是否存在下一個節點，以及下一個節點是否為共享節點(取反)
final boolean apparentlyFirstQueuedIsExclusive() {
Node h, s;
return (h = head) != null &&
(s = h.next) != null &&
!s.isShared() &&
s.thread != null;
}
3，釋放鎖
* unlock()
public void unlock() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
* tryReleaseShared(int unused)
protected final boolean tryReleaseShared(int unused) {
Thread current = Thread.currentThread();
// 判斷目前線程是否firstReader
// 此處條件判斷，主要為了遞減線程重入次數，如果可以釋放，則對關鍵辨別對象置空，幫助GC，并友善後續操作判斷
if (firstReader == current) {
// 繼續判斷firstReaderHoldCount重入次數，如果為1，說明可以徹底釋放，重置firstReader
// 不為1，遞減即可
if (firstReaderHoldCount == 1)
firstReader = null;
else
firstReaderHoldCount--;
} else {
// 目前線程不是firstReader，則從readHolds中擷取，并判斷count重入次數
HoldCounter rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current))
rh = readHolds.get();
int count = rh.count;
// count <=1，同樣表示徹底釋放，從readHolds中移除
if (count <= 1) {
readHolds.remove();
if (count <= 0)
throw unmatchedUnlockException();
}
--rh.count;
}
for (;;) {
// 擷取共享鎖的重入次數，并遞減，是否為0辨別
int c = getState();
int nextc = c - SHARED_UNIT;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
五，ReentrantReadWriteLock.WriteLock源碼分析
1，嘗試加鎖
* tryLock( )
public boolean tryLock( ) {
return sync.tryWriteLock();
}
* tryWriteLock()
final boolean tryWriteLock() {
Thread current = Thread.currentThread();
int c = getState();
// c不為0，表示存在獨占鎖或者共享鎖
if (c != 0) {
// 擷取獨占鎖重入次數
int w = exclusiveCount(c);
// w為0，說明獨占鎖為空，則表示存在共享鎖，擷取鎖失敗
// w不為0，并且鎖線程不是目前線程，說明存在其他線程持有獨占鎖，擷取鎖失敗
// 此處證明在同一線程下，擷取共享鎖後，不可擷取獨占鎖
if (w == 0 || current != getExclusiveOwnerThread())
return false;
if (w == MAX_COUNT)
throw new Error("Maximum lock count exceeded");
}
// CAS設定重入狀态，設定成功後，添加鎖線程為目前線程
if (!compareAndSetState(c, c + 1))
return false;
setExclusiveOwnerThread(current);
return true;
}
2，加鎖
* lock()
public void lock() {
sync.acquire(1);
}
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
* tryAcquire(int acquires)：此處嘗試加鎖與上一步嘗試加鎖基本一緻，隻是加了一步判斷
protected final boolean tryAcquire(int acquires) {
Thread current = Thread.currentThread();
int c = getState();
int w = exclusiveCount(c);
if (c != 0) {
if (w == 0 || current != getExclusiveOwnerThread())
return false;
if (w + exclusiveCount(acquires) > MAX_COUNT)
throw new Error("Maximum lock count exceeded");
setState(c + acquires);
return true;
}
// writerShouldBlock：判斷寫鎖是否需要阻塞
if (writerShouldBlock() ||
!compareAndSetState(c, c + acquires))
return false;
setExclusiveOwnerThread(current);
return true;
}
* writerShouldBlock()：該方法在非公平鎖中預設傳回false，再公平鎖實作中，進行了AQS隊列存在性校驗，保證順序執行
// 非公平鎖
final boolean writerShouldBlock() {
return false;
}
// 公平鎖
final boolean writerShouldBlock() {
return hasQueuedPredecessors();
}
public final boolean hasQueuedPredecessors() {
Node t = tail;
Node h = head;
Node s;
return h != t &&
((s = h.next) == null || s.thread != Thread.currentThread());
}
3，釋放鎖
* unlock()
public void unlock() {
sync.release(1);
}
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
* tryRelease(int releases)
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;
}