2019獨角獸企業重金招聘Python工程師標準>>>

最近做的一個小項目中有這樣的需求：整個項目有一份config.json保存著項目的一些配置，是存儲在本地文件的一個資源，并且應用中存在讀寫（讀>>寫）更新問題。既然讀寫并發操作，那么就涉及到操作互斥，這里自然想到了讀寫鎖，本文對讀寫鎖方面的知識做個梳理。

為什么需要讀寫鎖？

與傳統鎖不同的是讀寫鎖的規則是可以共享讀，但只能一個寫，總結起來為：讀讀不互斥，讀寫互斥，寫寫互斥，而一般的獨占鎖是：讀讀互斥，讀寫互斥，寫寫互斥，而場景中往往讀遠遠大于寫，讀寫鎖就是為了這種優化而創建出來的一種機制。

注意是讀遠遠大于寫，一般情況下獨占鎖的效率低來源于高并發下對臨界區的激烈競爭導致線程上下文切換。因此當并發不是很高的情況下，讀寫鎖由于需要額外維護讀鎖的狀態，可能還不如獨占鎖的效率高。因此需要根據實際情況選擇使用。

一個簡單的讀寫鎖實現

根據上面理論可以利用兩個int變量來簡單實現一個讀寫鎖，實現雖然爛，但是原理都是差不多的，值得閱讀下。

public class ReadWriteLock {/*** 讀鎖持有個數*/private int readCount = 0;/*** 寫鎖持有個數*/private int writeCount = 0;/*** 獲取讀鎖,讀鎖在寫鎖不存在的時候才能獲取*/public synchronized void lockRead() throws InterruptedException {// 寫鎖存在,需要waitwhile (writeCount > 0) {wait();}readCount++;}/*** 釋放讀鎖*/public synchronized void unlockRead() {readCount--;notifyAll();}/*** 獲取寫鎖,當讀鎖存在時需要wait.*/public synchronized void lockWrite() throws InterruptedException {// 先判斷是否有寫請求while (writeCount > 0) {wait();}// 此時已經不存在獲取寫鎖的線程了,因此占坑,防止寫鎖饑餓writeCount++;// 讀鎖為0時獲取寫鎖while (readCount > 0) {wait();}}/*** 釋放讀鎖*/public synchronized void unlockWrite() {writeCount--;notifyAll();}}

ReadWriteLock的實現原理

在Java中ReadWriteLock的主要實現為ReentrantReadWriteLock，其提供了以下特性：

公平性選擇：支持公平與非公平（默認）的鎖獲取方式，吞吐量非公平優先于公平。

可重入：讀線程獲取讀鎖之后可以再次獲取讀鎖，寫線程獲取寫鎖之后可以再次獲取寫鎖

可降級：寫線程獲取寫鎖之后，其還可以再次獲取讀鎖，然后釋放掉寫鎖，那么此時該線程是讀鎖狀態，也就是降級操作。

ReentrantReadWriteLock的結構

ReentrantReadWriteLock的核心是由一個基于AQS的同步器Sync構成，然后由其擴展出ReadLock（共享鎖），WriteLock（排它鎖）所組成。

并且從ReentrantReadWriteLock的構造函數中可以發現ReadLock與WriteLock使用的是同一個Sync，具體怎么實現同一個隊列既可以為共享鎖，又可以表示排他鎖下文會具體分析。

清單一：ReentrantReadWriteLock構造函數

public ReentrantReadWriteLock(boolean fair) {sync = fair ? new FairSync() : new NonfairSync();readerLock = new ReadLock(this);writerLock = new WriteLock(this);}

Sync的實現

sync是讀寫鎖實現的核心，sync是基于AQS實現的，在AQS中核心是state字段和雙端隊列，那么一個一個問題來分析。

Sync如何同時表示讀鎖與寫鎖？

清單2：讀寫鎖狀態獲取

static final int SHARED_SHIFT = 16; static final int SHARED_UNIT = (1 << SHARED_SHIFT); static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1; static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;/** Returns the number of shared holds represented in count */ static int sharedCount(int c) { return c >>> SHARED_SHIFT; } /** Returns the number of exclusive holds represented in count */ static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

從代碼中獲取讀寫狀態可以看出其是把state（int32位）字段分成高16位與低16位，其中高16位表示讀鎖個數，低16位表示寫鎖個數，如下圖所示（圖來自Java并發編程藝術）。

?

該圖表示當前一個線程獲取到了寫鎖，并且重入了兩次，因此低16位是3，并且該線程又獲取了讀鎖，并且重入了一次，所以高16位是2，當寫鎖被獲取時如果讀鎖不為0那么讀鎖一定是獲取寫鎖的這個線程。

讀鎖的獲取

讀鎖的獲取主要實現是AQS中的acquireShared方法，其調用過程如下代碼。

清單3：讀鎖獲取入口

// ReadLock public void lock() {sync.acquireShared(1); } // AQS public final void acquireShared(int arg) {if (tryAcquireShared(arg) < 0)doAcquireShared(arg); }

其中doAcquireShared(arg)方法是獲取失敗之后AQS中入隊操作，等待被喚醒后重新獲取，那么關鍵點就是tryAcquireShared(arg)方法，方法有點長，因此先總結出獲取讀鎖所經歷的步驟，獲取的第一部分步驟如下：

• 操作1：讀寫需要互斥，因此當存在寫鎖并且持有寫鎖的線程不是該線程時獲取失敗。
• 操作2：是否存在等待寫鎖的線程，存在的話則獲取讀鎖需要等待，避免寫鎖饑餓。(寫鎖優先級是比較高的)
• 操作3：CAS獲取讀鎖，實際上是state字段的高16位自增。
• 操作4：獲取成功后再ThreadLocal中記錄當前線程獲取讀鎖的次數。

清單4：讀鎖獲取的第一部分

protected final int tryAcquireShared(int unused) {Thread current = Thread.currentThread();int c = getState();// 操作1：存在寫鎖，并且寫鎖不是當前線程則直接去排隊if (exclusiveCount(c) != 0 &&getExclusiveOwnerThread() != current)return -1;int r = sharedCount(c);// 操作2：讀鎖是否該阻塞，對于非公平模式下寫鎖獲取優先級會高，如果存在要獲取寫鎖的線程則讀鎖需要讓步，公平模式下則先來先到if (!readerShouldBlock() && // 讀鎖使用高16位，因此存在獲取上限為2^16-1r < MAX_COUNT &&// 操作3：CAS修改讀鎖狀態，實際上是讀鎖狀態+1compareAndSetState(c, c + SHARED_UNIT)) {// 操作4：執行到這里說明讀鎖已經獲取成功，因此需要記錄線程狀態。if (r == 0) {firstReader = current; // firstReader是把讀鎖狀態從0變成1的那個線程firstReaderHoldCount = 1;} else if (firstReader == current) { firstReaderHoldCount++;} else {// 這些代碼實際上是從ThreadLocal中獲取當前線程重入讀鎖的次數，然后自增下。HoldCounter rh = cachedHoldCounter; // cachedHoldCounter是上一個獲取鎖成功的線程if (rh == null || rh.tid != getThreadId(current))cachedHoldCounter = rh = readHolds.get();else if (rh.count == 0)readHolds.set(rh);rh.count++;}return 1;}// 當操作2，操作3失敗時執行該邏輯return fullTryAcquireShared(current);}

當操作2，操作3失敗時會執行fullTryAcquireShared(current)，為什么會這樣寫呢？個人認為是一種補償操作，操作2與操作3失敗并不代表當前線程沒有讀鎖的資格，并且這里的讀鎖是共享鎖，有資格就應該被獲取成功，因此給予補償獲取讀鎖的操作。在fullTryAcquireShared(current)中是一個循環獲取讀鎖的過程，大致步驟如下：

• 操作5：等同于操作2，存在寫鎖，且寫鎖線程并非當前線程則直接返回失敗
• 操作6：當前線程是重入讀鎖，這里只會偏向第一個獲取讀鎖的線程以及最后一個獲取讀鎖的線程，其他都需要去AQS中排隊。
• 操作7：CAS改變讀鎖狀態
• 操作8：同操作4，獲取成功后再ThreadLocal中記錄當前線程獲取讀鎖的次數。

清單5：讀鎖獲取的第二部分

final int fullTryAcquireShared(Thread current) {HoldCounter rh = null;// 最外層嵌套循環for (;;) {int c = getState();// 操作5：存在寫鎖，且寫鎖并非當前線程則直接返回失敗if (exclusiveCount(c) != 0) {if (getExclusiveOwnerThread() != current)return -1;// else we hold the exclusive lock; blocking here// would cause deadlock.// 操作6：如果當前線程是重入讀鎖則放行} else if (readerShouldBlock()) {// Make sure we're not acquiring read lock reentrantly// 當前是firstReader，則直接放行,說明是已獲取的線程重入讀鎖if (firstReader == current) {// assert firstReaderHoldCount > 0;} else {// 執行到這里說明是其他線程，如果是cachedHoldCounter（其count不為0）也就是上一個獲取鎖的線程則可以重入，否則進入AQS中排隊// **這里也是對寫鎖的讓步**，如果隊列中頭結點為寫鎖，那么當前獲取讀鎖的線程要進入隊列中排隊if (rh == null) {rh = cachedHoldCounter;if (rh == null || rh.tid != getThreadId(current)) {rh = readHolds.get();if (rh.count == 0)readHolds.remove();}}// 說明是上述剛初始化的rh，所以直接去AQS中排隊if (rh.count == 0)return -1;}}if (sharedCount(c) == MAX_COUNT)throw new Error("Maximum lock count exceeded");// 操作7：修改讀鎖狀態，實際上讀鎖自增操作if (compareAndSetState(c, c + SHARED_UNIT)) {// 操作8：對ThreadLocal中維護的獲取鎖次數進行更新。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;}}}

讀鎖的釋放

清單6：讀鎖釋放入口

// ReadLock public void unlock() {sync.releaseShared(1); } // Sync public final boolean releaseShared(int arg) {if (tryReleaseShared(arg)) {doReleaseShared(); // 這里實際上是釋放讀鎖后喚醒寫鎖的線程操作return true;}return false; }

讀鎖的釋放主要是tryReleaseShared(arg)函數，因此拆解其步驟如下：

• 操作1：清理ThreadLocal中保存的獲取鎖數量信息
• 操作2：CAS修改讀鎖個數，實際上是自減一

清單7：讀鎖的釋放流程

protected final boolean tryReleaseShared(int unused) {Thread current = Thread.currentThread();// 操作1：清理ThreadLocal對應的信息if (firstReader == current) {;if (firstReaderHoldCount == 1)firstReader = null;elsefirstReaderHoldCount--;} 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;}// 操作2：循環中利用CAS修改讀鎖狀態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;}}

寫鎖的獲取

清單8：寫鎖的獲取入口

// WriteLockpublic void lock() {sync.acquire(1);} // AQSpublic final void acquire(int arg) {// 嘗試獲取，獲取失敗后入隊，入隊失敗則interrupt當前線程if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();}

寫鎖的獲取也主要是tryAcquire(arg)方法，這里也拆解步驟：

• 操作1：如果讀鎖數量不為0或者寫鎖數量不為0，并且不是重入操作，則獲取失敗。
• 操作2：如果當前鎖的數量為0，也就是不存在操作1的情況，那么該線程是有資格獲取到寫鎖，因此修改狀態，設置獨占線程為當前線程

清單9：寫鎖的獲取

protected final boolean tryAcquire(int acquires) {Thread current = Thread.currentThread();int c = getState();int w = exclusiveCount(c);// 操作1：c != 0，說明存在讀鎖或者寫鎖if (c != 0) {// (Note: if c != 0 and w == 0 then shared count != 0) // 寫鎖為0，讀鎖不為0 或者獲取寫鎖的線程并不是當前線程，直接失敗if (w == 0 || current != getExclusiveOwnerThread())return false;if (w + exclusiveCount(acquires) > MAX_COUNT)throw new Error("Maximum lock count exceeded");// Reentrant acquire// 執行到這里說明是寫鎖線程的重入操作，直接修改狀態，也不需要CAS因為沒有競爭setState(c + acquires);return true;}// 操作2：獲取寫鎖，writerShouldBlock對于非公平模式直接返回fasle，對于公平模式則線程需要排隊，因此需要阻塞。if (writerShouldBlock() ||!compareAndSetState(c, c + acquires))return false;setExclusiveOwnerThread(current);return true; }

寫鎖的釋放

清單10：寫鎖的釋放入口

// WriteLock public void unlock() {sync.release(1);} // AQS 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(arg)方法，其邏輯就比較簡單了，注釋很詳細。

清單11：寫鎖的釋放

protected final boolean tryRelease(int releases) {// 如果當前線程沒有獲取寫鎖卻釋放，則直接拋異常if (!isHeldExclusively())throw new IllegalMonitorStateException();// 狀態變更至nextcint nextc = getState() - releases;// 因為寫鎖是可以重入，所以在都釋放完畢后要把獨占標識清空boolean free = exclusiveCount(nextc) == 0;if (free)setExclusiveOwnerThread(null);// 修改狀態setState(nextc);return free;}

一些其他問題

鎖降級操作哪里體現？

鎖降級操作指的是一個線程獲取寫鎖之后再獲取讀鎖，然后讀鎖釋放掉寫鎖的過程。在tryAcquireShared(arg)獲取讀鎖的代碼中有如下代碼。
清單12：寫鎖降級策略

Thread current = Thread.currentThread();// 當前狀態int c = getState();// 存在寫鎖，并且寫鎖不等于當前線程時返回，換句話說等寫鎖為當前線程時則可以繼續往下獲取讀鎖。if (exclusiveCount(c) != 0 &&getExclusiveOwnerThread() != current)return -1; 。。。。。讀鎖獲取。。。。。

那么鎖降級有什么用？答案是為了可見性的保證。在ReentrantReadWriteLock的javadoc中有如下代碼，其是鎖降級的一個應用示例。

class CachedData {Object data;volatile boolean cacheValid;final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();void processCachedData() {// 獲取讀鎖rwl.readLock().lock();if (!cacheValid) {// Must release read lock before acquiring write lock，不釋放的話下面寫鎖會獲取不成功，造成死鎖rwl.readLock().unlock();// 獲取寫鎖rwl.writeLock().lock();try {// Recheck state because another thread might have// acquired write lock and changed state before we did.if (!cacheValid) {data = ...cacheValid = true;}// Downgrade by acquiring read lock before releasing write lock// 這里再次獲取讀鎖，如果不獲取那么當寫鎖釋放后可能其他寫線程再次獲得寫鎖，導致下方`use(data)`時出現不一致的現象// 這個操作就是降級rwl.readLock().lock();} finally {rwl.writeLock().unlock(); // Unlock write, still hold read}}try {// 使用完后釋放讀鎖use(data);} finally {rwl.readLock().unlock();}}}}

公平與非公平的區別

清單13：公平下的Sync

static final class FairSync extends Sync {private static final long serialVersionUID = -2274990926593161451L;final boolean writerShouldBlock() {return hasQueuedPredecessors(); // 隊列中是否有元素，有責當前操作需要block}final boolean readerShouldBlock() {return hasQueuedPredecessors();// 隊列中是否有元素，有責當前操作需要block}}

公平下的Sync實現策略是所有獲取的讀鎖或者寫鎖的線程都需要入隊排隊，按照順序依次去嘗試獲取鎖。

清單14：非公平下的Sync

static final class NonfairSync extends Sync {private static final long serialVersionUID = -8159625535654395037L;final boolean writerShouldBlock() {// 非公平下不考慮排隊，因此寫鎖可以競爭獲取return false; // writers can always barge}final boolean readerShouldBlock() {/* As a heuristic to avoid indefinite writer starvation,* block if the thread that momentarily appears to be head* of queue, if one exists, is a waiting writer. This is* only a probabilistic effect since a new reader will not* block if there is a waiting writer behind other enabled* readers that have not yet drained from the queue.*/// 這里實際上是一個優先級，如果隊列中頭部元素時寫鎖，那么讀鎖需要等待，避免寫鎖饑餓。return apparentlyFirstQueuedIsExclusive();}}

非公平下由于搶占式獲取鎖，寫鎖是可能產生饑餓，因此解決辦法就是提高寫鎖的優先級，換句話說獲取寫鎖之前先占坑。

?

作者：牛李，一個正在努力學習的碼農，主要關注后端領域、代碼設計，以及一些有趣的技術。GitHub: https://github.com/mrdear

本文系作者投稿文章。歡迎投稿。

投稿內容要求

• 互聯網技術相關，包括但不限于開發語言、網絡、數據庫、架構、運維、前端、DevOps（DevXXX）、AI、區塊鏈、存儲、移動、安全、技術團隊管理等內容。
• 文章不需要首發，可以是已經在開源中國博客或網上其它平臺發布過的。但是鼓勵首發，首發內容被收錄可能性較大。
• 如果你是記錄某一次解決了某一個問題（這在博客中占絕大比例），那么需要將問題的前因后果描述清楚，最直接的就是結合圖文等方式將問題復現，同時完整地說明解決思路與最終成功的方案。
• 如果你是分析某一技術理論知識，請從定義、應用場景、實際案例、關鍵技術細節、觀點等方面，對其進行較為全面地介紹。
• 如果你是以實際案例分享自己或者公司對諸如某一架構模型、通用技術、編程語言、運維工具的實踐，那么請將事件相關背景、具體技術細節、演進過程、思考、應用效果等方面描述清楚。
• 其它未盡 case 具體情況具體分析，不虛的，文章投過來試試先，比如我們并不拒絕就某個熱點事件對其進行的報導、深入解析。

投稿方式

• 以 Word 或者 Markdown 文檔的形式將稿件投遞到?oscbianji@oschina.cn?郵箱

重要說明

• 作者需要擁有所投文章的所有權，不能將別人的文章拿過來投遞。
• 投遞的文章需要經過審核，如果開源中國編輯覺得需要的話，將與作者一起進一步完善文章，意在使文章更佳、傳播更廣。
• 文章版權歸作者所有，開源中國獲得文章的傳播權，可在開源中國各個平臺進行文章傳播，同時保留文章原始出處和作者信息，可在官方博客中標原創標簽。

轉載于:https://my.oschina.net/editorial-story/blog/1928306

總結

以上是生活随笔為你收集整理的Java 读写锁实现原理的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。