您好,登錄后才能下訂單哦!
這篇文章主要介紹了Java ReentrantReadWriteLock讀寫鎖如何實現(xiàn)的相關知識,內容詳細易懂,操作簡單快捷,具有一定借鑒價值,相信大家閱讀完這篇Java ReentrantReadWriteLock讀寫鎖如何實現(xiàn)文章都會有所收獲,下面我們一起來看看吧。
ReentrantReadWriteLock
讀寫鎖是使用AQS的集大成者,用了獨占模式和共享模式。
上圖是ReentrantReadWriteLock
讀寫鎖的類結構圖:
實現(xiàn)了ReadWriteLock
接口,該接口提供了獲取讀鎖和寫鎖的API。
ReentrantReadWriteLock
讀寫鎖內部的成員變量readLock是讀鎖,指向內部類ReadLock。
ReentrantReadWriteLock
讀寫鎖內部的成員變量writeLock是寫鎖,指向內部類WriteLock。
ReentrantReadWriteLock
讀寫鎖內部的成員變量sync是繼承AQS的同步器,他有兩個子類FairSync
公平同步器和NoFairSync
非公平同步器,讀寫鎖內部也有一個sync,他們使用的是同一個sync。
讀寫鎖用的同一個sync同步器,那么他們共享同一個state, 這樣不會混淆嗎?
不會,ReentrantReadWriteLock
讀寫鎖使用了AQS中state值得低16位表示寫鎖得計數(shù),用高16位表示讀鎖得計數(shù),這樣就可以使用同一個AQS同時管理讀鎖和寫鎖。
1.ReentrantReadWriteLock類重要成員變量
// 讀鎖 private final ReentrantReadWriteLock.ReadLock readerLock; // 寫鎖 private final ReentrantReadWriteLock.WriteLock writerLock; // 同步器 final Sync sync;
2.ReentrantReadWriteLock構造方法
//默認是非公平鎖,可以指定參數(shù)創(chuàng)建公平鎖 public ReentrantReadWriteLock(boolean fair) { // true 為公平鎖 sync = fair ? new FairSync() : new NonfairSync(); // 這兩個 lock 共享同一個 sync 實例,都是由 ReentrantReadWriteLock 的 sync 提供同步實現(xiàn) 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,代表寫鎖的最大重入次數(shù) static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1; // 低16位掩碼:0b 1111 1111 1111 1111,用來獲取寫鎖重入的次數(shù) static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1; // 獲取讀寫鎖的讀鎖分配的總次數(shù) static int sharedCount(int c) { return c >>> SHARED_SHIFT; } // 寫鎖(獨占)鎖的重入次數(shù) 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線程執(zhí)行加讀鎖 r.lock()
,嘗試獲取鎖,發(fā)現(xiàn)已經被寫鎖占據(jù)了,加鎖失敗。
3.t2線程被封裝成一個共享模式Node.SHARED的節(jié)點,加入到AQS的隊列中。
4.在阻塞前,t2線程發(fā)現(xiàn)自己是隊列中的老二,會嘗試再次獲取讀鎖,因為t1沒有釋放,它會失敗,然后它會把隊列的前驅節(jié)點的狀態(tài)改為-1,然后阻塞自身,也就是t2線程。
上面中黃色三角形就是等待狀態(tài)的值,前驅節(jié)點變成-1
上面中的灰色表示節(jié)點所在的線程阻塞了
5.后面如過有其他線程如t3,t4加讀鎖或者寫鎖,由于t1線程沒有釋放鎖,會變成下面的狀態(tài)。
上面是整個解鎖的流程,下面深入源碼驗證這個流程。
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) && // 將當前線程關聯(lián)到一個 Node 對象上, 模式為獨占模式 // 進入 AQS 隊列阻塞 acquireQueued(addWaiter(Node.EXCLUSIVE), arg) ) { selfInterrupt(); } } protected final boolean tryAcquire(int acquires) { // 獲取當前線程 Thread current = Thread.currentThread(); //獲得鎖的狀態(tài) int c = getState(); // 獲得低 16 位, 代表寫鎖的 state 計數(shù) int w = exclusiveCount(c); // c不等于0表示加了讀鎖或者寫鎖 if (c != 0) { if ( // c != 0 and w == 0 表示有讀鎖返回錯誤,讀鎖不支持鎖升級, 或者 w == 0 || // w != 0 說明有寫鎖,寫鎖的擁有者不是自己,獲取失敗 current != getExclusiveOwnerThread() ) { // 獲得鎖失敗 return false; } // 寫鎖計數(shù)超過低 16 位最大數(shù)量, 報異常 if (w + exclusiveCount(acquires) > MAX_COUNT) throw new Error("Maximum lock count exceeded"); // 寫鎖重入, 獲得鎖成功,沒有并發(fā),所以不使用 CAS setState(c + acquires); return true; } if ( // c == 0,說明沒有任何鎖,判斷寫鎖是否該阻塞,是 false 就嘗試獲取鎖,失敗返回 false writerShouldBlock() || // 嘗試更改計數(shù)失敗 !compareAndSetState(c, c + acquires) ) { // 獲得鎖失敗 return false; } // 獲得鎖成功,設置鎖的持有線程為當前線程 setExclusiveOwnerThread(current); return true; } // 非公平鎖 writerShouldBlock 總是返回 false, 無需阻塞 final boolean writerShouldBlock() { return false; } // 公平鎖會檢查 AQS 隊列中是否有前驅節(jié)點, 沒有(false)才去競爭 final boolean writerShouldBlock() { return hasQueuedPredecessors(); } }
tryAcquire()
方法是模板方法,由子類自定義實現(xiàn)獲取鎖的邏輯。
線程如果獲取寫鎖失敗的話,通過acquireQueued()
方法封裝成獨占Node加入到AQS隊列中。
2.讀鎖加鎖源碼
ReadLock
類的lock()
方法是加讀鎖的入口方法,調用tryAcquireShared()
方法嘗試獲取讀鎖,返回負數(shù),失敗,加入到隊列中。
// 加讀鎖的方法入口 public void lock() { sync.acquireShared(1); } public final void acquireShared(int arg) { // tryAcquireShared 返回負數(shù), 表示獲取讀鎖失敗,加入到隊列中 if (tryAcquireShared(arg) < 0) doAcquireShared(arg); }
tryAcquireShared()
方法是一個模板方法,AQS類中定義語義,子類實現(xiàn),如果返回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,共享鎖分配出去的總次數(shù) int r = sharedCount(c); // 讀鎖是否應該阻塞 if (!readerShouldBlock() && r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)) { // 嘗試增加讀鎖計數(shù) // 加鎖成功 // 加鎖之前讀鎖為 0,說明當前線程是第一個讀鎖線程 if (r == 0) { firstReader = current; firstReaderHoldCount = 1; // 第一個讀鎖線程是自己就發(fā)生了讀鎖重入 } 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 (;;) 獲取讀鎖, 執(zhí)行過程中無阻塞 return fullTryAcquireShared(current); } // 非公平鎖 readerShouldBlock 偏向寫鎖一些,看 AQS 阻塞隊列中第一個節(jié)點是否是寫鎖,是則阻塞,反之不阻塞 // 防止一直有讀鎖線程,導致寫鎖線程饑餓 // true 則該阻塞, false 則不阻塞 final boolean readerShouldBlock() { return apparentlyFirstQueuedIsExclusive(); } // 下面是公平鎖的readerShouldBlock // 公平鎖會檢查 AQS 隊列中是否有前驅節(jié)點, 沒有(false)才去競爭 final boolean readerShouldBlock() { return hasQueuedPredecessors(); }
fullTryAcquireShared()
方法是通過自旋的方式不斷獲取讀鎖,因為由于前面的readerShouldBlock
返回false或者cas失敗,導致沒有獲取到鎖,需要不斷重試。
final int fullTryAcquireShared(Thread current) { // 當前讀鎖線程持有的讀鎖次數(shù)對象 HoldCounter rh = null; for (;;) { int c = getState(); // 說明有線程持有寫鎖 if (exclusiveCount(c) != 0) { // 寫鎖不是自己則獲取鎖失敗 if (getExclusiveOwnerThread() != current) return -1; } else if (readerShouldBlock()) { // 條件成立說明當前線程是 firstReader,當前鎖是讀忙碌狀態(tài),而且當前線程也是讀鎖重入 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) { // 將當前線程關聯(lián)到一個 Node 對象上, 模式為共享模式 final Node node = addWaiter(Node.SHARED); boolean failed = true; try { boolean interrupted = false; for (;;) { // 獲取前驅節(jié)點 final Node p = node.predecessor(); // 如果前驅節(jié)點就頭節(jié)點就去嘗試獲取鎖 if (p == head) { // 再一次嘗試獲取讀鎖 int r = tryAcquireShared(arg); // r >= 0 表示獲取成功 if (r >= 0) { //【這里會設置自己為頭節(jié)點,喚醒相連的后序的共享節(jié)點】 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()
方法是在后續(xù)讀鎖被喚醒后,搶到鎖要處理的邏輯,包括修改隊列的頭結點,以及喚醒隊列中的下一個共享節(jié)點。
private void setHeadAndPropagate(Node node, int propagate) { Node h = head; // 設置自己為 head 節(jié)點 setHead(node); // propagate 表示有共享資源(例如共享讀鎖或信號量),為 0 就沒有資源 if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) { // 獲取下一個節(jié)點 Node s = node.next; // 如果當前是最后一個節(jié)點,或者下一個節(jié)點是【等待共享讀鎖的節(jié)點】 if (s == null || s.isShared()) // 喚醒后繼節(jié)點 doReleaseShared(); } }
由于上面t1線程加的寫鎖,所有其他的線程都被阻塞了,只有在t1線程解鎖以后,其他線程才能被喚醒,我們現(xiàn)在看下t1線程被喚醒了,會發(fā)生什么?
1.t1線程執(zhí)行解鎖w.unlock()
成功,修改AQS中的state。
這里的state變?yōu)榱?_0。
AQS當前占用線程exclusiveOwnerThread屬性變?yōu)閚ull。
2.t1線程喚醒隊列中等待的老二, 為什么不是老大,因為老大是一個空節(jié)點,不會設置任何的線程。t2線程被喚醒后,搶鎖成功,修改state中高16位為1。
老二的線程節(jié)點變?yōu)樗{色節(jié)點
AQS中的state變?yōu)?_0。
3.t2線程恢復運行,設置原來的老二節(jié)點為頭節(jié)點
4.t2線程要做的事情還沒結束呢,因為是共享模式,它現(xiàn)在釋放了,就此時也喚醒隊列中的下一個共享節(jié)點。
5.t3線程恢復去競爭讀鎖成功,這時state的高位+1,變成2。
6.這時候t3線程所在的Node設置為頭節(jié)點,同時發(fā)現(xiàn)對列的下一個節(jié)點不是共享節(jié)點,而是獨占節(jié)點,就不會喚醒后面的節(jié)點了。
7.之后t2線程和t3線程進入尾聲,執(zhí)行r.unlock
操作,state的計數(shù)減一,直到變?yōu)?。
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; // 頭節(jié)點不為空并且不是等待狀態(tài)不是 0,喚醒后繼的非取消節(jié)點 if (h != null && h.waitStatus != 0) unparkSuccessor(h); return true; } return false; }
tryRelease()
方法是AQS提供的模板方法,返回true表示成功,false失敗,由自定義同步器實現(xiàn)。
protected final boolean tryRelease(int releases) { if (!isHeldExclusively()) throw new IllegalMonitorStateException(); int nextc = getState() - releases; // 因為可重入的原因, 寫鎖計數(shù)為 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提供的模板方法,由自定義同步器實現(xiàn)。
protected final boolean tryReleaseShared(int unused) { //自選 for (;;) { int c = getState(); int nextc = c - SHARED_UNIT; // 讀鎖的計數(shù)不會影響其它獲取讀鎖線程, 但會影響其它獲取寫鎖線程,計數(shù)為 0 才是真正釋放 if (compareAndSetState(c, nextc)) // 返回是否已經完全釋放了 return nextc == 0; } }
調用doReleaseShared()
方法喚醒等待的線程,這個方法調用的地方有兩處,還記得嗎,一個這是里的解鎖,還有一個是前面加共享鎖阻塞的地方,喚醒后獲取鎖成功,也會調用doReleaseShared()
方法。
private void doReleaseShared() { // 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一個節(jié)點 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 被多次執(zhí)行 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // 喚醒后繼節(jié)點 unparkSuccessor(h); } // 如果已經是 0 了,改為 -3,用來解決傳播性 else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; } // 條件不成立說明被喚醒的節(jié)點非常積極,直接將自己設置為了新的 head, // 此時喚醒它的節(jié)點(前驅)執(zhí)行 h == head 不成立,所以不會跳出循環(huán),會繼續(xù)喚醒新的 head 節(jié)點的后繼節(jié)點 if (h == head) break; } }
關于“Java ReentrantReadWriteLock讀寫鎖如何實現(xiàn)”這篇文章的內容就介紹到這里,感謝各位的閱讀!相信大家對“Java ReentrantReadWriteLock讀寫鎖如何實現(xiàn)”知識都有一定的了解,大家如果還想學習更多知識,歡迎關注億速云行業(yè)資訊頻道。
免責聲明:本站發(fā)布的內容(圖片、視頻和文字)以原創(chuàng)、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關證據(jù),一經查實,將立刻刪除涉嫌侵權內容。