您好,登錄后才能下訂單哦!
一、ReentrantReadWriteLock簡介
上一篇文章我們將講到的ReentrantLock和Synchronized鎖,都屬于排他鎖,也就是說只會有一個線程獲取鎖;而我們今天講的ReentrantReadWriteLock(讀寫鎖)是支持多個線程同時獲取鎖的在獲取讀鎖時;但是在獲取到寫鎖時,其它的寫鎖和讀鎖都會阻塞;其實可以看出讀寫鎖,維護了一對鎖,一個寫鎖,其實是個排他鎖,一個讀鎖,是共享鎖;通過分離讀鎖和寫鎖,使得并發(fā)性相比一般的排他鎖有很大的提升;讀寫鎖的性能比排他鎖好,因為在大多數(shù)場景中讀是多于寫的;讀寫鎖提供了,公平性的選擇、重新進(jìn)入(該鎖支持重進(jìn)入,以讀寫鎖線程為例:讀線程在獲取讀鎖之后,能夠再次獲取讀鎖。而寫線程在獲取了寫鎖之后能夠再次獲取寫鎖,同時也可以獲取讀鎖)和鎖降級(遵循獲取寫鎖、獲取讀鎖在釋放寫鎖的次序,寫鎖能夠降級成為讀鎖)等特性。
二、ReentrantReadWriteLock基本成員
我們先來看一張ReentrantReadWriteLock類圖
Sync是一個內(nèi)部類,繼承AQS,主要實現(xiàn)AQS的一些方法。
基本成員簡介
static final int SHARED_SHIFT = 16;
// 這個是讀鎖的原始累加值(也就是說每次獲取讀鎖都是獲取狀態(tài)state,然后用state加它),是2^16
// 舉個例子,假設(shè)現(xiàn)在state為1,那么現(xiàn)在來獲取讀鎖就是1+SHARED_UNIT
static final int SHARED_UNIT = (1 << SHARED_SHIFT);
// 讀鎖和寫鎖的最大數(shù)量,都是2^16 - 1
static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1;
// 寫鎖的掩碼,其實就是2^16 - 1,這個數(shù)的二進(jìn)制很特殊,16位全是1
static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
/** Returns the number of shared holds represented in count */
// 讀鎖的數(shù)量
static int sharedCount(int c) { return c >>> SHARED_SHIFT; }
/** Returns the number of exclusive holds represented in count */
// 寫鎖的數(shù)量
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
// 記錄每個線程獲取讀鎖的數(shù)量
// count是數(shù)量
// tid是線程的唯一標(biāo)識
static final class HoldCounter {
int count = 0;
// Use id, not reference, to avoid garbage retention
final long tid = getThreadId(Thread.currentThread());
}
// 繼承ThreadLocal,主要用來存儲HoldCounter
static final class ThreadLocalHoldCounter
extends ThreadLocal<java.util.concurrent.locks.ReentrantReadWriteLock.Sync.HoldCounter> {
public java.util.concurrent.locks.ReentrantReadWriteLock.Sync.HoldCounter initialValue() {
return new java.util.concurrent.locks.ReentrantReadWriteLock.Sync.HoldCounter();
}
}
// 可以理解為最后一個獲取讀鎖的線程(用于優(yōu)化性能,不需要去ThreadLocal查找)
private transient HoldCounter cachedHoldCounter;
// 第一個獲取讀鎖的線程和讀鎖的數(shù)量,作用是如果是第一個,不走ThreadLocal
private transient Thread firstReader = null;
private transient int firstReaderHoldCount;
解釋下讀寫鎖的狀態(tài)計算,state一個數(shù),怎么控制的讀和寫
先來看寫鎖,先來看寫鎖的掩碼EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1,這個數(shù)是65535,二進(jìn)制就是16個1,我們看見獲取寫鎖的個數(shù)c & 65535(exclusiveCount這個方法,c代表就是state,鎖的狀態(tài),不理解可以看看AQS對state的定義),但是由于二進(jìn)制16位是1,所以只要c在0-65535這個范圍,取&都還是c(由于這個掩碼的特殊性),根據(jù)寫鎖的定義只能有一個線程獲取寫鎖,寫鎖獲取了就要阻塞其它線程獲取讀或者寫,怎么判斷了,其實只要判斷c& 這個二進(jìn)制是不是等于0就可以了,所以了寫鎖的范圍其實就是0-65535,其實二進(jìn)制的范圍就是低16位(因為最大數(shù)量是MAX_COUNT = 2^16-1)。
再來分析下讀鎖,讀鎖我們主要關(guān)注下SHARED_UNIT就可以,獲取鎖其實是用c+SHARED_UNIT(c代表的就是sate),釋放鎖是用c-SHARED_UNIT,這個數(shù)是 65536(SHARED_UNIT 其實就是 2 ^ 16),我們每次獲取讀鎖其實就是把SHARED_UNIT累加,其實我們可以把SHARED_UNIT的一次累加就當(dāng)做一次讀鎖的獲取,我們看下讀鎖的值得范圍是0 - 負(fù)65536(負(fù)數(shù)和int的最大值有關(guān),int正數(shù)的最大值是2147483647,在給它累加其實就會變?yōu)樨?fù)數(shù),最后最大其實就是高16位全是1,因為讀鎖的最大數(shù)量是MAX_COUNT = 2^16-1,所以其實讀鎖的范圍可以看做是高16位),獲取讀鎖的個數(shù)就是無符號右移16位(就是sharedCount方法),因為可能是負(fù)數(shù)。
上面的圖片其實就是一個讀鎖,一個寫鎖,這種情況只有在同一個線程才會發(fā)生,如果你能算出一個寫鎖和讀鎖,說明你基本理解了讀寫鎖狀態(tài)控制的運算方法。
ReadLock和WriteLock,內(nèi)部類,繼承Lock,提供一些鎖的方法。
FairSync和NonfairSync,內(nèi)部類,是繼承Sync,主要實現(xiàn)公平和非公平的一些方法
三、ReentrantReadWriteLock基本方法
1)、構(gòu)造方法
// 無參,默認(rèn)非公平
public ReentrantReadWriteLock() {
this(false);
}
// 有參
public ReentrantReadWriteLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
readerLock = new ReadLock(this);
writerLock = new WriteLock(this);
}
2)、WriteLock:寫鎖一些方法,如下圖:
①、獲取鎖lock方法,我們可以看出調(diào)用的是sync.acquire方法,由于sync繼承自AQS所以調(diào)用的其實是AQS的acquire,但是AQS的tryAcquire需要子類自己實現(xiàn),所以我們看看tryAcquire(可以看出是個獨占方法,也符合寫鎖的定義,只會有一個線程獲?。?。
public void lock() {
sync.acquire(1);
}
子類sync的tryAcquire
// 寫鎖的狀態(tài)控制(state)
protected final boolean tryAcquire(int acquires) {
/*
* Walkthrough:
* 1. If read count nonzero or write count nonzero
* and owner is a different thread, fail.
* 2. If count would saturate, fail. (This can only
* happen if count is already nonzero.)
* 3. Otherwise, this thread is eligible for lock if
* it is either a reentrant acquire or
* queue policy allows it. If so, update state
* and set owner.
*/
// 獲取當(dāng)前線程
Thread current = Thread.currentThread();
// 獲取當(dāng)前鎖的狀態(tài)
int c = getState();
// 獲取寫鎖的狀態(tài),c & (2的16次方-1)
// 2的16次方-1 其實就是65535,變成二進(jìn)制就是16個1
int w = exclusiveCount(c);
// c不等于0,證明有線程獲取鎖了,不管是讀鎖或者寫鎖
if (c != 0) {
// (Note: if c != 0 and w == 0 then shared count != 0)
// w == 0,說明有讀鎖了,w!= 0證明有寫鎖
// current != getExclusiveOwnerThread()說明有寫鎖了,不是自己
if (w == 0 || current != getExclusiveOwnerThread())
return false;
// 證明了是自己獲取了寫鎖,如果大于鎖的最大數(shù)量,拋異常
if (w + exclusiveCount(acquires) > MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// Reentrant acquire
// 說明沒有超出限制,可以重入
setState(c + acquires);
return true;
}
// 走到這一步,證明還沒有線程獲取鎖
// writerShouldBlock 現(xiàn)在公平還是非公平,由FairSync和NonfairSync實現(xiàn)這個方法
if (writerShouldBlock() ||
// cas設(shè)置所得狀態(tài)
!compareAndSetState(c, c + acquires))
//失敗或者公平鎖在我的前面有排隊節(jié)點
return false;
// 設(shè)置擁有鎖的線程
setExclusiveOwnerThread(current);
return true;
}
在上面的寫鎖獲取鎖時我們需要注意下writerShouldBlock這個方法,它是實現(xiàn)公平和非公平的關(guān)鍵,它的公平和非公平的方法實現(xiàn)不同,公平是需要確認(rèn)自己前面是否有排隊節(jié)點,非公平直接返回false,具體查看這個方法。
②、寫鎖釋放鎖:unlock方法,它是其實也是調(diào)用的也是調(diào)用AQS的release方法,我們直接看子類的實現(xiàn)吧。
public void unlock() {
sync.release(1);
}
子類sync的tryRelease方法
protected final boolean tryRelease(int releases) {
// 判斷是否獲取鎖的是這個線程
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
// 釋放鎖,修改state
int nextc = getState() - releases;
// 判斷寫鎖是否完全釋放
boolean free = exclusiveCount(nextc) == 0;
// 完全釋放,修改鎖的擁有者為null
if (free)
setExclusiveOwnerThread(null);
// cas狀態(tài)
setState(nextc);
return free;
}
③、我們發(fā)現(xiàn)WriteLock里面還有一些獲取鎖的方法,lockInterruptibly響應(yīng)中斷,tryLock(long timeout, TimeUnit unit)支持中斷并且?guī)С瑫r時間,其實都是調(diào)用了AQS里面的這些方法,然后獲取鎖這部分的實現(xiàn)都是調(diào)用的子類sync的tryAcquire方法。
3)、ReadLock:讀鎖的一些方法,如下圖
①、獲取鎖lock方法,調(diào)用過程和寫鎖一樣,先走AQS,不過這一次調(diào)用的是共享鎖的方法acquireShared,然后AQS在調(diào)用子類sync的實現(xiàn)方法。
public void lock() {
sync.acquireShared(1);
}
sync的tryAcquireShared
protected final int tryAcquireShared(int unused) {
/*
* Walkthrough:
* 1. If write lock held by another thread, fail.
* 2. Otherwise, this thread is eligible for
* lock wrt state, so ask if it should block
* because of queue policy. If not, try
* to grant by CASing state and updating count.
* Note that step does not check for reentrant
* acquires, which is postponed to full version
* to avoid having to check hold count in
* the more typical non-reentrant case.
* 3. If step 2 fails either because thread
* apparently not eligible or CAS fails or count
* saturated, chain to version with full retry loop.
*/
// 獲取當(dāng)前線程
Thread current = Thread.currentThread();
// 獲取當(dāng)前鎖狀態(tài)
int c = getState();
// exclusiveCount(c) != 0 有線程獲取了寫鎖
// 并且這個獲取寫鎖的線程不是自己
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
// 現(xiàn)在只會有三種清理
// 1、還沒有任何線程獲取鎖,不管是讀鎖還是寫鎖
// 2、有線程獲取到了讀鎖
// 3、獲取寫鎖的線程時自己
int r = sharedCount(c);
// readerShouldBlock 由FairSync和NonfairSync實現(xiàn)
// FairSync 判斷是否前面有排隊節(jié)點
// NonfairSync排隊節(jié)點是否有寫節(jié)點
if (!readerShouldBlock() &&
r < MAX_COUNT &&
compareAndSetState(c, c + SHARED_UNIT)) {
// r == 0證明還沒有獲取到鎖
if (r == 0) {
firstReader = current;
firstReaderHoldCount = 1;
// 重入鎖
} else if (firstReader == current) {
firstReaderHoldCount++;
} else {
// 獲取最后一個獲取鎖的HoldCounter
java.util.concurrent.locks.ReentrantReadWriteLock.Sync.HoldCounter rh = cachedHoldCounter;
// 最后一個HoldCounter是空或者不是本線程,就設(shè)置一個
if (rh == null || rh.tid != getThreadId(current))
// 其實這一步做了兩件事情,其實是先set了一個HoldCounter,然后在get之后設(shè)置給rh
cachedHoldCounter = rh = readHolds.get();
else if (rh.count == 0) // 理解不了什么時候會是0
readHolds.set(rh);
rh.count++;
}
return 1;
}
// 執(zhí)行fullTryAcquireShared方法有幾種情況了
// 1.獲取鎖的下一個節(jié)點還是寫鎖,需要等待
// 2.到達(dá)獲取鎖的最大數(shù)量
// 3.可能存在多線程進(jìn)程來設(shè)置讀鎖,cas失敗了
return fullTryAcquireShared(current);
}
final int fullTryAcquireShared(Thread current) {
/*
* This code is in part redundant with that in
* tryAcquireShared but is simpler overall by not
* complicating tryAcquireShared with interactions between
* retries and lazily reading hold counts.
*/
java.util.concurrent.locks.ReentrantReadWriteLock.Sync.HoldCounter rh = null;
for (;;) { // 自旋獲取鎖
int c = getState(); // 獲取鎖狀態(tài)
if (exclusiveCount(c) != 0) { // 判斷有沒有寫鎖,不等于0證明有寫鎖
if (getExclusiveOwnerThread() != current) // 寫鎖不是自己
return -1; // 返回
// 寫鎖時自己,其實就可以獲取,其實就是鎖降級(可以看做是一種特殊的重入鎖)
// else we hold the exclusive lock; blocking here
// would cause deadlock.
// 到下面這個else if證明沒有寫鎖
// readerShouldBlock 由FairSync和NonfairSync實現(xiàn)公平和非公平原則
// FairSync 判斷是否前面有排隊節(jié)點
// NonfairSync排隊節(jié)點是否有寫節(jié)點
} else if (readerShouldBlock()) { // 到這一步證明了,是公平鎖或者非公平鎖的頭結(jié)點.next是寫鎖,
// 此線程需要進(jìn)入同步隊列了,下面就是判斷這個線程有沒有獲取過鎖
// Make sure we're not acquiring read lock reentrantly
// 第一個獲取鎖的是當(dāng)前線程,證明可以重入
if (firstReader == current) {
// assert firstReaderHoldCount > 0;
} else {
// 進(jìn)去這里說明,firstReader不是當(dāng)前線程,那就說明獲取讀鎖的不止一個了,因為firstReader不可能為null
// 獲取最后一個獲取讀鎖的HoldCounter
if (rh == null) { // rh == null 只會是第一次循環(huán)
rh = cachedHoldCounter; // 獲取緩存的HoldCounter
if (rh == null || rh.tid != getThreadId(current)) {
// 從 ThreadLocal 中取出計數(shù)器,如果沒有就會重新創(chuàng)建并設(shè)置
rh = readHolds.get();
if (rh.count == 0) // 那就證明沒有獲取到讀讀鎖
readHolds.remove(); // 刪除這個
}
}
if (rh.count == 0) // 這個是上面剛剛創(chuàng)建的證明獲取鎖失敗了,需要進(jìn)入隊列
return -1;
}
}
// 獲取讀鎖的數(shù)量==MAX_COUNT,拋異常
if (sharedCount(c) == MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// 使用cas設(shè)置讀鎖的狀態(tà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;
}
}
}
在讀鎖獲取鎖時需要注意下readerShouldBlock這個方法,和寫鎖類似,這個方法也是主要實現(xiàn)公平與非公平的關(guān)鍵,非公平鎖(NonfairSync)時需要注意,如果獲取讀鎖時,需要執(zhí)行apparentlyFirstQueuedIsExclusive這個方法,判斷隊列head的next是否是寫鎖(是寫鎖,讓這個寫鎖先來,避免寫鎖饑餓,就是避免寫線程獲取不到鎖,所以寫有很高的優(yōu)先權(quán)),則自己獲取讀鎖就需要排隊,公平鎖(FairSync)實現(xiàn)則還是需要判斷隊列里面是否有節(jié)點在排隊。
②、釋放鎖unlock,調(diào)用AQS的releaseShared方法,我們主要關(guān)注子類實現(xiàn)
public void unlock() {
sync.releaseShared(1);
}
子類sync的tryReleaseShared方法
protected final boolean tryReleaseShared(int unused) {
// 獲取當(dāng)前線程
Thread current = Thread.currentThread();
// 如果第一個讀鎖擁有者是當(dāng)前線程
if (firstReader == current) {
// assert firstReaderHoldCount > 0;
// 讀鎖的數(shù)量為1,沒有重入
if (firstReaderHoldCount == 1)
firstReader = null;
else
// 重入了,修改數(shù)量
firstReaderHoldCount--;
// 讀鎖已經(jīng)不是一個了
} else {
// 獲取最后一個緩存
java.util.concurrent.locks.ReentrantReadWriteLock.Sync.HoldCounter rh = cachedHoldCounter;
// 獲取緩存HoldCounter
if (rh == null || rh.tid != getThreadId(current))
// 不是就獲取
rh = readHolds.get();
// 獲取讀鎖線程的數(shù)量
int count = rh.count;
// 如果小于等于1
if (count <= 1) {
// 刪除這個線程的HoldCounter
readHolds.remove();
if (count <= 0)
throw unmatchedUnlockException();
}
// 讀鎖數(shù)量遞減
--rh.count;
}
for (;;) {
// 獲取狀態(tài)stase
int c = getState();
// 讀鎖狀態(tài)減1(其實就是減SHARED_UNIT)
int nextc = c - SHARED_UNIT;
// cas設(shè)置線程狀態(tài)
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.
// 為什么nextc == 0才會返回true了
// nextc == 0代表了什么了,沒有讀鎖了
// 返回true,就代表可以去喚醒下一個線程了,但是隊列的線程是寫線程或者線程了,不確定
// 但是這個對讀線程是沒有影響的,但是對寫鎖是有影響的,我們想象一下什么情況下才會下個節(jié)點會是寫鎖獲取的線程了
// 其實就是已經(jīng)有線程獲取了寫鎖,因為有線程獲取了寫鎖,所以可能發(fā)生鎖降級,寫鎖降級為讀鎖
// 為了保護鎖降級的語義,所以必須保護讀鎖,直到?jīng)]有讀鎖了才會去喚醒后面可能的寫鎖,也就是返回true
return nextc == 0;
}
}
③、lockInterruptibly和tryLock(long timeout, TimeUnit unit)和寫鎖的這些方法作用一樣。
四、總結(jié)
ReentrantReadWriteLock讀寫鎖,要想學(xué)習(xí)好這個類,就必須了解什么是讀鎖,什么是寫鎖,怎么區(qū)別讀鎖和寫鎖,因為我們都知道鎖都是通過AQS的state來控制的,但是現(xiàn)在是兩個鎖,所以理解ReentrantReadWriteLock對state拆分的運算很重要,也就是二進(jìn)制的低16位是寫鎖,高16位是讀鎖,也不得不說大神設(shè)計讓人嘆為觀止啊;還有就是理解讀寫鎖的一些特性,重入指的就是同一個線程獲取鎖之后,再次調(diào)用lock方法不會被阻塞,但是注意調(diào)用幾次lock,就要調(diào)用幾次unlock,因為我們通過源碼得知重入其實就是對state的累加,還有就是鎖降級,我個人更愿意理解為特殊的重入,鎖降級就是一個線程先獲得寫鎖,然后這個線程再去獲取讀鎖,這樣不會阻塞,然后寫鎖其實就降級為了讀鎖,然后在釋放寫鎖,最后釋放讀鎖,作者為什么這么設(shè)計了,書上說的,鎖降級主要是為了保證數(shù)據(jù)的可見性,如果當(dāng)前線程不獲取讀鎖而是直接釋放寫鎖,假設(shè)此刻另一個線程(記作線程T)獲取了寫鎖并修改了數(shù)據(jù),那么當(dāng)前線程無法感知線程T的數(shù)據(jù)更新,如果當(dāng)前線程獲取讀鎖,即遵循鎖降級的步驟,則線程T,就會阻塞,直到當(dāng)前線程使用數(shù)據(jù)并釋放讀鎖之后,線程T才會獲得寫鎖并更新數(shù)據(jù)。個人理解這個鎖降級其實是一種特殊的鎖的優(yōu)化策略,在我們需要在邊寫邊讀的這種業(yè)務(wù)場景中,保證數(shù)據(jù)可見性的同時(不讓其他線程獲取寫鎖),提升本線程讀鎖性能,因為不需要和寫鎖或者其他讀鎖(公平鎖)去競爭獲取鎖,而是直接降級為讀鎖。
參考 《Java 并發(fā)編程的藝術(shù)》
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報,并提供相關(guān)證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。