在上一篇《你真的懂ReentrantReadWriteLock嗎？》中我給大家留了一個引子，一個更高效同時可以避免寫飢餓的讀寫鎖---StampedLock。StampedLock實現了不僅多個讀不互相阻塞，同時在讀操作時不會阻塞寫操作。
為什么StampedLock這么神奇？能夠達到這種效果，它的核心思想在於，在讀的時候如果發生了寫，應該通過重試的方式來獲取新的值，而不應該阻塞寫操作。這種模式也就是典型的無鎖編程思想，和CAS自旋的思想一樣。這種操作方式決定了StampedLock在讀線程非常多而寫線程非常少的場景下非常適用，同時還避免了寫飢餓情況的發生。這篇文章將通過以下幾點來分析StampedLock。
- StampedLock的官方使用示例分析
- 源碼分析：讀寫鎖共享的狀態量
- 源碼分析：寫鎖的釋放和獲取
- 源碼分析：悲觀讀鎖的釋放和獲取
- 性能測試
StampedLock的官方使用示例分析
先來看一個官方給出的StampedLock使用案例：
public class Point {
private double x, y;
private final StampedLock stampedLock = new StampedLock();
//寫鎖的使用
void move(double deltaX, double deltaY){
long stamp = stampedLock.writeLock(); //獲取寫鎖
try {
x += deltaX;
y += deltaY;
} finally {
stampedLock.unlockWrite(stamp); //釋放寫鎖
}
}
//樂觀讀鎖的使用
double distanceFromOrigin() {
long stamp = stampedLock.tryOptimisticRead(); //獲得一個樂觀讀鎖
double currentX = x;
double currentY = y;
if (!stampedLock.validate(stamp)) { //檢查樂觀讀鎖后是否有其他寫鎖發生，有則返回false
stamp = stampedLock.readLock(); //獲取一個悲觀讀鎖
try {
currentX = x;
} finally {
stampedLock.unlockRead(stamp); //釋放悲觀讀鎖
}
}
return Math.sqrt(currentX*currentX + currentY*currentY);
}
//悲觀讀鎖以及讀鎖升級寫鎖的使用
void moveIfAtOrigin(double newX,double newY) {
long stamp = stampedLock.readLock(); //悲觀讀鎖
try {
while (x == 0.0 && y == 0.0) {
long ws = stampedLock.tryConvertToWriteLock(stamp); //讀鎖轉換為寫鎖
if (ws != 0L) { //轉換成功
stamp = ws; //票據更新
x = newX;
y = newY;
break;
} else {
stampedLock.unlockRead(stamp); //轉換失敗釋放讀鎖
stamp = stampedLock.writeLock(); //強制獲取寫鎖
}
}
} finally {
stampedLock.unlock(stamp); //釋放所有鎖
}
}
}
首先看看第一個方法move，可以看到它和ReentrantReadWriteLock寫鎖的使用基本一樣，都是簡單的獲取釋放，可以猜測這里也是一個獨占鎖的實現。需要注意的是 在獲取寫鎖是會返回個只long類型的stamp，然后在釋放寫鎖時會將stamp傳入進去。這個stamp是做什么用的呢？如果我們在中間改變了這個值又會發生什么呢？這里先暫時不做解釋，后面分析源碼時會解答這個問題。
第二個方法distanceFromOrigin就比較特別了，它調用了tryOptimisticRead，根據名字判斷這是一個樂觀讀鎖。首先什么是樂觀鎖？樂觀鎖的意思就是先假定在樂觀鎖獲取期間，共享變量不會被改變，既然假定不會被改變，那就不需要上鎖。在獲取樂觀讀鎖之后進行了一些操作，然后又調用了validate方法，這個方法就是用來驗證tryOptimisticRead之后，是否有寫操作執行過，如果有，則獲取一個讀鎖，這里的讀鎖和ReentrantReadWriteLock中的讀鎖類似，猜測也是個共享鎖。
第三個方法moveIfAtOrigin，它做了一個鎖升級的操作，通過調用tryConvertToWriteLock嘗試將讀鎖轉換為寫鎖，轉換成功后相當於獲取了寫鎖，轉換失敗相當於有寫鎖被占用，這時通過調用writeLock來獲取寫鎖進行操作。
看過了上面的三個方法，估計大家對怎么使用StampedLock有了一個初步的印象。下面就通過對StampedLock源碼的分析來一步步了解它背后是怎么解決鎖飢餓問題的。
源碼分析：讀寫鎖共享的狀態量
從上面的使用示例中我們看到，在StampedLock中，除了提供了類似ReentrantReadWriteLock讀寫鎖的獲取釋放方法，還提供了一個樂觀讀鎖的獲取方式。那么這三種方式是如何交互的呢？根據AQS的經驗，StampedLock中應該也是使用了一個狀態量來標志鎖的狀態。通過下面的源碼可以證明這點：
// 用於操作state后獲取stamp的值
private static final int LG_READERS = 7;
private static final long RUNIT = 1L; //0000 0000 0001
private static final long WBIT = 1L << LG_READERS; //0000 1000 0000
private static final long RBITS = WBIT - 1L; //0000 0111 1111
private static final long RFULL = RBITS - 1L; //0000 0111 1110
private static final long ABITS = RBITS | WBIT; //0000 1111 1111
private static final long SBITS = ~RBITS; //1111 1000 0000
//初始化時state的值
private static final long ORIGIN = WBIT << 1; //0001 0000 0000
//鎖共享變量state
private transient volatile long state;
//讀鎖溢出時用來存儲多出的毒素哦
private transient int readerOverflow;
上面的源碼中除了定義state變量外，還提供了一系列變量用來操作state，用來表示讀鎖和寫鎖的各種狀態。為了方便理解，我將他們都表示成二進制的值，長度有限，這里用低12位來表示64的long，高位自動用0補齊。要理解這些狀態的作用，就需要具體分析三種鎖操作方式是怎么通過state這一個變量來表示的，首先來看看獲取寫鎖和釋放寫鎖。
源碼分析：寫鎖的釋放和獲取
public StampedLock() {
state = ORIGIN; //初始化state為 0001 0000 0000
}
public long writeLock() {
long s, next;
return ((((s = state) & ABITS) == 0L && //沒有讀寫鎖
U.compareAndSwapLong(this, STATE, s, next = s + WBIT)) ? //cas操作嘗試獲取寫鎖
next : acquireWrite(false, 0L)); //獲取成功后返回next，失敗則進行后續處理，排隊也在后續處理中
}
public void unlockWrite(long stamp) {
WNode h;
if (state != stamp || (stamp & WBIT) == 0L) //stamp值被修改，或者寫鎖已經被釋放，拋出錯誤
throw new IllegalMonitorStateException();
state = (stamp += WBIT) == 0L ? ORIGIN : stamp; //加0000 1000 0000來記錄寫鎖的變化，同時改變寫鎖狀態
if ((h = whead) != null && h.status != 0)
release(h);
}
這里先說明兩點結論：讀鎖通過前7位來表示，每獲取一個讀鎖，則加1。寫鎖通過除前7位后剩下的位來表示，每獲取一次寫鎖，則加1000 0000，這兩點在后面的源碼中都可以得倒證明。
初始化時將state變量設置為0001 0000 0000。寫鎖獲取通過((s = state) & ABITS)操作等於0時默認沒有讀鎖和寫鎖。寫鎖獲取分三種情況：
-
沒有讀鎖和寫鎖時，state為0001 0000 0000
0001 0000 0000 & 0000 1111 1111 = 0000 0000 0000 // 等於0L，可以嘗試獲取寫鎖 -
有一個讀鎖時，state為0001 0000 0001
0001 0000 0001 & 0000 1111 1111 = 0000 0000 0001 // 不等於0L -
有一個寫鎖，state為0001 1000 0000
0001 1000 0000 & 0000 1111 1111 = 0000 1000 0000 // 不等於0L
獲取到寫鎖，需要將s + WBIT設置到state，也就是說每次獲取寫鎖，都需要加0000 1000 0000。同時返回s + WBIT的值
0001 0000 0000 + 0000 1000 0000 = 0001 1000 0000
釋放寫鎖首先判斷stamp的值有沒有被修改過或者多次釋放，之后通過state = (stamp += WBIT) == 0L ? ORIGIN : stamp來釋放寫鎖，位操作表示如下：
stamp += WBIT
0010 0000 0000 = 0001 1000 0000 + 0000 1000 0000
這一步操作是重點！！！寫鎖的釋放並不是像ReentrantReadWriteLock一樣+1然后-1，而是通過再次加0000 1000 0000來使高位每次都產生變化，為什么要這樣做？直接減掉0000 1000 0000不就可以了嗎？這就是為了后面樂觀鎖做鋪墊，讓每次寫鎖都留下痕跡。
大家可以想象這樣一個場景，字母A變化為B能看到變化，如果在一段時間內從A變到B然后又變到A，在內存中自會顯示A，而不能記錄變化的過程，這也就是CAS中的ABA問題。在StampedLock中就是通過每次對高位加0000 1000 0000來達到記錄寫鎖操作的過程，可以通過下面的步驟理解：
第一次獲取寫鎖：
0001 0000 0000 + 0000 1000 0000 = 0001 1000 0000
第一次釋放寫鎖：
0001 1000 0000 + 0000 1000 0000 = 0010 0000 0000
第二次獲取寫鎖：
0010 0000 0000 + 0000 1000 0000 = 0010 1000 0000
第二次釋放寫鎖：
0010 1000 0000 + 0000 1000 0000 = 0011 0000 0000
第n次獲取寫鎖：
1110 0000 0000 + 0000 1000 0000 = 1110 1000 0000
第n次釋放寫鎖：
1110 1000 0000 + 0000 1000 0000 = 1111 0000 0000
可以看到第8位在獲取和釋放寫鎖時會產生變化，也就是說第8位是用來表示寫鎖狀態的，前7位是用來表示讀鎖狀態的，8位之后是用來表示寫鎖的獲取次數的。這樣就有效的解決了ABA問題，留下了每次寫鎖的記錄，也為后面樂觀鎖檢查變化提供了基礎。
關於acquireWrite方法這里不做具體分析，方法非常復雜，感興趣的同學可以網上搜索相關資料。這里只對該方法做下簡單總結，該方法分兩步來進行線程排隊，首先通過隨機探測的方式多次自旋嘗試獲取鎖，然后自旋一定次數失敗后再初始化節點進行插入。
源碼分析：悲觀讀鎖的釋放和獲取
public long readLock() {
long s = state, next;
return ((whead == wtail && (s & ABITS) < RFULL && //隊列為空，無寫鎖，同時讀鎖未溢出，嘗試獲取讀鎖
U.compareAndSwapLong(this, STATE, s, next = s + RUNIT)) ? //cas嘗試獲取讀鎖+1
next : acquireRead(false, 0L)); //獲取讀鎖成功，返回s + RUNIT，失敗進入后續處理，類似acquireWrite
}
public void unlockRead(long stamp) {
long s, m; WNode h;
for (;;) {
if (((s = state) & SBITS) != (stamp & SBITS) ||
(stamp & ABITS) == 0L || (m = s & ABITS) == 0L || m == WBIT)
throw new IllegalMonitorStateException();
if (m < RFULL) { //小於最大記錄值（最大記錄值127超過后放在readerOverflow變量中）
if (U.compareAndSwapLong(this, STATE, s, s - RUNIT)) { //cas嘗試釋放讀鎖-1
if (m == RUNIT && (h = whead) != null && h.status != 0)
release(h);
break;
}
}
else if (tryDecReaderOverflow(s) != 0L) //readerOverflow - 1
break;
}
}
悲觀讀鎖的獲取和ReentrantReadWriteLock類似，不同在於StampedLock的讀鎖很容易溢出，最大只有127，超過后通過一個額外的變量readerOverflow來存儲，這是為了給寫鎖留下更大的空間，因為寫鎖是在不停增加的。悲觀讀鎖獲取分下面四種情況：
-
沒有讀鎖和寫鎖時，state為0001 0000 0000
// 小於 0000 0111 1110，可以嘗試獲取讀鎖
0001 0000 0000 & 0000 1111 1111 = 0000 0000 0000 -
有一個讀鎖時，state為0001 0000 0001
// 小於 0000 0111 1110，可以嘗試獲取讀鎖
0001 0000 0001 & 0000 1111 1111 = 0000 0000 0001 -
有一個寫鎖，state為0001 1000 0000
// 大於 0000 0111 1110，不可以獲取讀鎖
0001 1000 0000 & 0000 1111 1111 = 0000 1000 0000 -
讀鎖溢出，state為0001 0111 1110
// 等於 0000 0111 1110，不可以獲取讀鎖
0001 0111 1110 & 0000 1111 1111 = 0000 0111 1110
讀鎖的釋放過程在沒有溢出的情況下是通過s - RUNIT操作也就是-1來釋放的，當溢出后則將readerOverflow變量-1。
樂觀讀鎖的獲取和驗證
樂觀讀鎖因為實際上沒有獲取過鎖，所以也就沒有釋放鎖的過程，只是在操作后通過驗證檢查和獲取前的變化。源碼如下：
//嘗試獲取樂觀鎖
public long tryOptimisticRead() {
long s;
return (((s = state) & WBIT) == 0L) ? (s & SBITS) : 0L;
}
//驗證樂觀鎖獲取之后是否有過寫操作
public boolean validate(long stamp) {
//該方法之前的所有load操作在內存屏障之前完成，對應的還有storeFence()及fullFence()
U.loadFence();
return (stamp & SBITS) == (state & SBITS); //比較是否有過寫操作
}
樂觀鎖基本原理就時獲取鎖時記錄state的寫狀態，然后在操作完成之后檢查寫狀態是否有變化，因為寫狀態每次都會在高位留下記錄，這樣就避免了寫鎖獲取又釋放后得不到准確數據。獲取寫鎖記錄有三種情況：
-
沒有讀鎖和寫鎖時，state為0001 0000 0000
//((s = state) & WBIT) == 0L) true
0001 0000 0000 & 0000 1000 0000 = 0000 0000 0000
//(s & SBITS)
0001 0000 0000 & 1111 1000 0000 = 0001 0000 0000 -
有一個讀鎖時，state為0001 0000 0001
//((s = state) & WBIT) == 0L) true
0001 0000 0001 & 0000 1000 0000 = 0000 0000 0000
//(s & SBITS)
0001 0000 0001 & 1111 1000 0000 = 0001 0000 0000 -
有一個寫鎖，state為0001 1000 0000
//((s = state) & WBIT) == 0L) false
0001 1000 0000 & 0000 1000 0000 = 0000 1000 0000
//0L
0000 0000 0000
驗證過程中是否有過寫操作，分四種情況
-
寫過一次
0001 0000 0000 & 1111 1000 0000 = 0001 0000 0000
0010 0000 0000 & 1111 1000 0000 = 0010 0000 0000 //false -
未寫過，但讀過
0001 0000 0000 & 1111 1000 0000 = 0001 0000 0000
0001 0000 1111 & 1111 1000 0000 = 0001 0000 0000 //true -
正在寫
0001 0000 0000 & 1111 1000 0000 = 0001 0000 0000
0001 1000 0000 & 1111 1000 0000 = 0001 1000 0000 //false -
之前正在寫，無論是否寫完都不會為0L
0000 0000 0000 & 1111 1000 0000 = 0000 0000 0000 //false
性能測試
分析完了StampedLock的實現原理，這里對StampedLock、ReentrantReadWriteLock以及Synchronized分別在各種場景下進行性能測試，測試的基准代碼采用https://blog.takipi.com/java-8-stampedlocks-vs-readwritelocks-and-synchronized/ 文章中的代碼，首先貼出上述博客中的測試結果,文章中的OPTIMISTIC模式由於采用了“臟讀”模式，這里不采用OPTIMISTIC的測試結果，只比較StampedLock、ReentrantReadWriteLock以及Synchronized。
5個讀線程和5個寫線程場景：表現最好的是StampedLock的正常模式以及ReentrantReadWriteLock。
10個讀線程和10個寫線程場景：表現最好的是StampedLock的正常模式以及Synchronized。
16個讀線程和4個寫線程場景：表現最好的是StampedLock的正常模式以及Synchronized。
19個讀線程和1個寫線程場景：表現最好的是Synchronized。
博客評論中還有一種測試場景2000讀線程和1個寫線程，測試結果如下：
StampedLock ... 12814.2 ReentrantReadWriteLock ... 18882.8 Synchronized ... 22696.4
表現最好的是StampedLock。
看完了上面的測試，前面3種場景表現最好的都為StampedLock，但第4種情況下StampedLock表現很差，於是我自己對代碼又進行了一遍測試，同時鑒於讀寫鎖的大量應用在緩存場景下，讀寫差距極大，我增加了100個讀和1個寫的場景。
測試機器：MAC OS(10.12.6),CPU : 2.4 GHz Intel Core i5,內存：8G 軟件版本：JDK1.8
測試結果如下：
19個讀線程和1個寫線程場景：表現最好的是StampedLock以及Synchronized。
讀線程: 19. 寫線程: 1. 循環次數: 5. 計算總和: 1000000
100個讀線程和1個寫線程場景：表現最好的是StampedLock以及Synchronized。
讀線程: 100. 寫線程: 1. 循環次數: 5. 計算總和: 100000
通過上述測試，可以發現整體性能平均而言StampedLock和Synchronized相差不大，StampedLock在讀寫差距加大時稍微有點優勢。而ReentrantReadWriteLock性能之差有點出乎意料，基本可以達到拋棄使用的地步了，不知道大家對ReentrantReadWriteLock的使用場景有什么建議？
同時鑒於原生的Synchronized后期可優化空間比較大，而且在代碼復雜性以及安全性上面都具有一定優勢，因此在絕大多數場景可以使用Synchronized來進行同步，對性能有一定要求的在某些特定場景下可以使用StampedLock。測試所用代碼在我所引用的博客中都可以找到，大家可以自行嘗試測試，如果對結果有什么疑問，歡迎在評論中提出。
參考資料：
https://blog.takipi.com/java-8-stampedlocks-vs-readwritelocks-and-synchronized/