並發編程從零開始（十二）-Lock與Condition

2021 年 10 月 31 日
筆記
juc, 並發編程

並發編程從零開始（十二）-Lock與Condition

8 Lock與Condition

8.1 互斥鎖

8.1.1 鎖的可重入性

「可重入鎖」是指當一個執行緒調用 object.lock()獲取到鎖，進入臨界區後，再次調用object.lock()，仍然可以獲取到該鎖。顯然，通常的鎖都要設計成可重入的，否則就會發生死鎖。

synchronized關鍵字，就是可重入鎖。在一個synchronized方法method1()裡面調用另外一個synchronized方法method2()。如果synchronized關鍵字不可重入，那麼在method2()處就會發生阻塞，這顯然不可行。

8.1.2 類繼承層次

在正式介紹鎖的實現原理之前，先看一下 Concurrent 包中的與互斥鎖（ReentrantLock）相關類之間的繼承層次，如下圖所示：

Lock是一個介面，其定義如下：

常用的方法是lock()/unlock()。lock()不能被中斷，對應的lockInterruptibly()可以被中斷。

ReentrantLock本身沒有程式碼邏輯，實現都在其內部類Sync中：

8.1.3 鎖的公平性VS非公平性

Sync是一個抽象類，它有兩個子類FairSync與NonfairSync，分別對應公平鎖和非公平鎖。從下面的ReentrantLock構造方法可以看出，會傳入一個布爾類型的變數fair指定鎖是公平的還是非公平的，默認為非公平的。

什麼叫公平鎖和非公平鎖呢？先舉個現實生活中的例子，一個人去火車站售票窗口買票，發現現場有人排隊，於是他排在隊伍末尾，遵循先到者優先服務的規則，這叫公平；如果他去了不排隊，直接衝到窗口買票，這叫作不公平。

對應到鎖的例子，一個新的執行緒來了之後，看到有很多執行緒在排隊，自己排到隊伍末尾，這叫公平；執行緒來了之後直接去搶鎖，這叫作不公平。默認設置的是非公平鎖，其實是為了提高效率，減少執行緒切換。

鎖實現的基本原理

Sync的父類AbstractQueuedSynchronizer經常被稱作隊列同步器（AQS），這個類非常重要，該類的父類是AbstractOwnableSynchronizer。

此處的鎖具備synchronized功能，即可以阻塞一個執行緒。為了實現一把具有阻塞或喚醒功能的鎖，需要幾個核心要素：

需要一個state變數，標記該鎖的狀態。state變數至少有兩個值：0、1。對state變數的操作，使用CAS保證執行緒安全。
需要記錄當前是哪個執行緒持有鎖。
需要底層支援對一個執行緒進行阻塞或喚醒操作。
需要有一個隊列維護所有阻塞的執行緒。這個隊列也必須是執行緒安全的無鎖隊列，也需要使用CAS。

針對要素1和2，在上面兩個類中有對應的體現：

state取值不僅可以是0、1，還可以大於1，就是為了支援鎖的可重入性。例如，同樣一個執行緒，調用5次lock，state會變成5；然後調用5次unlock，state減為0。

當state=0時，沒有執行緒持有鎖，exclusiveOwnerThread=null；

當state=1時，有一個執行緒持有鎖，exclusiveOwnerThread=該執行緒；

當state > 1時，說明該執行緒重入了該鎖。

對於要素3，Unsafe類提供了阻塞或喚醒執行緒的一對操作原語，也就是park/unpark。

有一個LockSupport的工具類，對這一對原語做了簡單封裝：

在當前執行緒中調用park()，該執行緒就會被阻塞；在另外一個執行緒中，調用unpark(Thread thread)，傳入一個被阻塞的執行緒，就可以喚醒阻塞在park()地方的執行緒。

unpark(Thread thread)，它實現了一個執行緒對另外一個執行緒的「精準喚醒」。notify也只是喚醒某一個執行緒，但無法指定具體喚醒哪個執行緒。

針對要素4，在AQS中利用雙向鏈表和CAS實現了一個阻塞隊列。如下所示：

阻塞隊列是整個AQS核心中的核心。如下圖所示，head指向雙向鏈表頭部，tail指向雙向鏈表尾部。入隊就是把新的Node加到tail後面，然後對tail進行CAS操作；出隊就是對head進行CAS操作，把head向後移一個位置。

初始的時候，head=tail=NULL；然後，在往隊列中加入阻塞的執行緒時，會新建一個空的Node，讓head和tail都指向這個空Node；之後，在後面加入被阻塞的執行緒對象。所以，當head=tail的時候，說明隊列為空。

8.1.4 公平與非公平的lock()實現差異

下面分析基於AQS，ReentrantLock在公平性和非公平性上的實現差異。

8.1.5 阻塞隊列與喚醒機制

下面進入鎖的最為關鍵的部分，即acquireQueued(…)方法內部一探究竟。

先說addWaiter(…)方法，就是為當前執行緒生成一個Node，然後把Node放入雙向鏈表的尾部。要注意的是，這只是把Thread對象放入了一個隊列中而已，執行緒本身並未阻塞。

創建節點，嘗試將節點追加到隊列尾部。獲取tail節點，將tail節點的next設置為當前節點。

如果tail不存在，就初始化隊列。

在addWaiter(…)方法把Thread對象加入阻塞隊列之後的工作就要靠acquireQueued(…)方法完成。執行緒一旦進入acquireQueued(…)就會被無限期阻塞，即使有其他執行緒調用interrupt()方法也不能將其喚醒，除非有其他執行緒釋放了鎖，並且該執行緒拿到了鎖，才會從accquireQueued(…)返回。

進入acquireQueued(…)，該執行緒被阻塞。在該方法返回的一刻，就是拿到鎖的那一刻，也就是被喚醒的那一刻，此時會刪除隊列的第一個元素（head指針前移1個節點）。

首先，acquireQueued(…)方法有一個返回值，表示什麼意思呢？雖然該方法不會中斷響應，但它會記錄被阻塞期間有沒有其他執行緒向它發送過中斷訊號。如果有，則該方法會返回true；否則，返回false。

基於這個返回值，才有了下面的程式碼：

當 acquireQueued(…)返回 true 時，會調用 selfInterrupt()，自己給自己發送中斷訊號，也就是自己把自己的中斷標誌位設為true。之所以要這麼做，是因為自己在阻塞期間，收到其他執行緒中斷訊號沒有及時響應，現在要進行補償。這樣一來，如果該執行緒在lock程式碼塊內部有調用sleep()之類的阻塞方法，就可以拋出異常，響應該中斷訊號。

阻塞就發生在下面這個方法中：

執行緒調用 park()方法，自己把自己阻塞起來，直到被其他執行緒喚醒，該方法返回。

park()方法返回有兩種情況。

其他執行緒調用了unpark(Thread t)。
其他執行緒調用了t.interrupt()。這裡要注意的是，lock()不能響應中斷，但LockSupport.park()會響應中斷。

也正因為LockSupport.park()可能被中斷喚醒，acquireQueued(…)方法才寫了一個for死循環。喚醒之後，如果發現自己排在隊列頭部，就去拿鎖；如果拿不到鎖，則再次自己阻塞自己。不斷重複此過程，直到拿到鎖。

被喚醒之後，通過Thread.interrupted()來判斷是否被中斷喚醒。如果是情況1，會返回false；如果是情況2，則返回true。

8.1.6 unlock()實現分析

說完了lock，下面分析unlock的實現。unlock不區分公平還是非公平。

上圖中，當前執行緒要釋放鎖，先調用tryRelease(arg)方法，如果返回true，則取出head，讓head獲取鎖。

對於tryRelease方法：

首先計算當前執行緒釋放鎖後的state值。

如果當前執行緒不是排他執行緒，則拋異常，因為只有獲取鎖的執行緒才可以進行釋放鎖的操作。

此時設置state，沒有使用CAS，因為是單執行緒操作。

再看unparkSuccessor方法：

release()裡面做了兩件事：tryRelease(…)方法釋放鎖；unparkSuccessor(…)方法喚醒隊列中的後繼者。

8.1.7 lockInterruptibly()實現分析

上面的 lock 不能被中斷，這裡的 lockInterruptibly()可以被中斷：

這裡的 acquireInterruptibly(…)也是 AQS 的模板方法，裡面的 tryAcquire(…)分別被 FairSync和NonfairSync實現。

主要看doAcquireInterruptibly(…)方法：

當parkAndCheckInterrupt()返回true的時候，說明有其他執行緒發送中斷訊號，直接拋出InterruptedException，跳出for循環，整個方法返回。

8.1.8 tryLock()實現分析

tryLock()實現基於調用非公平鎖的tryAcquire(…)，對state進行CAS操作，如果操作成功就拿到鎖；

如果操作不成功則直接返回false，也不阻塞。

8.2 讀寫鎖

和互斥鎖相比，讀寫鎖（ReentrantReadWriteLock）就是讀執行緒和讀執行緒之間不互斥。

讀讀不互斥，讀寫互斥，寫寫互斥

8.2.1 類繼承層次

ReadWriteLock是一個介面，內部由兩個Lock介面組成。

ReentrantReadWriteLock實現了該介面，使用方式如下：

也就是說，當使用 ReadWriteLock 的時候，並不是直接使用，而是獲得其內部的讀鎖和寫鎖，然後分別調用lock/unlock。

8.2.2 讀寫鎖實現的基本原理

個視圖呢？可以理解為是一把鎖，執行緒分成兩類：讀執行緒和寫執行緒。讀執行緒和寫執行緒之間不互斥（可以同時拿到這把鎖），讀執行緒之間不互斥，寫執行緒之間互斥。

從下面的構造方法也可以看出，readerLock和writerLock實際共用同一個sync對象。sync對象同互斥鎖一樣，分為非公平和公平兩種策略，並繼承自AQS。

同互斥鎖一樣，讀寫鎖也是用state變數來表示鎖狀態的。只是state變數在這裡的含義和互斥鎖完全不同。在內部類Sync中，對state變數進行了重新定義，如下所示：

也就是把 state 變數拆成兩半，低16位，用來記錄寫鎖。但同一時間既然只能有一個執行緒寫，為什麼還需要16位呢？這是因為一個寫執行緒可能多次重入。例如，低16位的值等於5，表示一個寫執行緒重入了5次。

高16位，用來「讀」鎖。例如，高16位的值等於5，既可以表示5個讀執行緒都拿到了該鎖；也可以表示一個讀執行緒重入了5次。

為什麼要把一個int類型變數拆成兩半，而不是用兩個int型變數分別表示讀鎖和寫鎖的狀態呢？

這是因為無法用一次CAS 同時操作兩個int變數，所以用了一個int型的高16位和低16位分別表示讀鎖和寫鎖的狀態。

當state=0時，說明既沒有執行緒持有讀鎖，也沒有執行緒持有寫鎖；當state != 0時，要麼有執行緒持有讀鎖，要麼有執行緒持有寫鎖，兩者不能同時成立，因為讀和寫互斥。這時再進一步通過sharedCount(state)和exclusiveCount(state)判斷到底是讀執行緒還是寫執行緒持有了該鎖。

8.2.3 AQS的兩對模板方法

下面介紹在ReentrantReadWriteLock的兩個內部類ReadLock和WriteLock中，是如何使用state變數的。

acquire/release、acquireShared/releaseShared 是AQS裡面的兩對模板方法。互斥鎖和讀寫鎖的寫鎖都是基於acquire/release模板方法來實現的。讀寫鎖的讀鎖是基於acquireShared/releaseShared這對模板方法來實現的。這兩對模板方法的程式碼如下：

將讀/寫、公平/非公平進行排列組合，就有4種組合。如下圖所示，上面的兩個方法都是在Sync中實現的。Sync中的兩個方法又是模板方法，在NonfairSync和FairSync中分別有實現。最終的對應關係如下：

讀鎖的公平實現：Sync.tryAccquireShared()+FairSync中的兩個重寫的子方法。
讀鎖的非公平實現：Sync.tryAccquireShared()+NonfairSync中的兩個重寫的子方法。
寫鎖的公平實現：Sync.tryAccquire()+FairSync中的兩個重寫的子方法。
寫鎖的非公平實現：Sync.tryAccquire()+NonfairSync中的兩個重寫的子方法。

對於公平，比較容易理解，不論是讀鎖，還是寫鎖，只要隊列中有其他執行緒在排隊（排隊等讀鎖，或者排隊等寫鎖），就不能直接去搶鎖，要排在隊列尾部。

對於非公平，讀鎖和寫鎖的實現策略略有差異。

寫執行緒能搶鎖，前提是state=0，只有在沒有其他執行緒持有讀鎖或寫鎖的情況下，它才有機會去搶鎖。或者state != 0，但那個持有寫鎖的執行緒是它自己，再次重入。寫執行緒是非公平的，即writerShouldBlock()方法一直返回false。

對於讀執行緒，假設當前執行緒被讀執行緒持有，然後其他讀執行緒還非公平地一直去搶，可能導致寫執行緒永遠拿不到鎖，所以對於讀執行緒的非公平，要做一些「約束」。當發現隊列的第1個元素是寫執行緒的時候，讀執行緒也要阻塞，不能直接去搶。即偏向寫執行緒。

8.2.4 WriteLock公平vs非公平實現

寫鎖是排他鎖，實現策略類似於互斥鎖。

1.tryLock()實現分析

lock()方法：

在互斥鎖部分講過了。tryLock和lock方法不區分公平/非公平。

2.unlock()實現分析

unlock()方法不區分公平/非公平。

8.2.5 ReadLock公平vs非公平實現

讀鎖是共享鎖，其實現策略和排他鎖有很大的差異。

1.tryLock()實現分析

2.unlock()實現分析

tryReleaseShared()的實現：

因為讀鎖是共享鎖，多個執行緒會同時持有讀鎖，所以對讀鎖的釋放不能直接減1，而是需要通過一個for循環+CAS操作不斷重試。這是tryReleaseShared和tryRelease的根本差異所在。

8.3 Condition

8.3.1 Condition與Lock的關係

Condition本身也是一個介面，其功能和wait/notify類似，如下所示：

wait()/notify()必須和synchronized一起使用，Condition也必須和Lock一起使用。因此，在Lock的介面中，有一個與Condition相關的介面：

8.3.2 Condition的使用場景

以ArrayBlockingQueue為例。如下所示為一個用數組實現的阻塞隊列，執行put(…)操作的時候，隊列滿了，生產者執行緒被阻塞；執行take()操作的時候，隊列為空，消費者執行緒被阻塞。

8.3.3 Condition實現原理

可以發現，Condition的使用很方便，避免了wait/notify的生產者通知生產者、消費者通知消費者的問題。具體實現如下：

由於Condition必須和Lock一起使用，所以Condition的實現也是Lock的一部分。首先查看互斥鎖和讀寫鎖中Condition的構造方法：

首先，讀寫鎖中的 ReadLock 是不支援 Condition 的，讀寫鎖的寫鎖和互斥鎖都支援Condition。雖然它們各自調用的是自己的內部類Sync，但內部類Sync都繼承自AQS。因此，上面的程式碼sync.newCondition最終都調用了AQS中的newCondition：

每一個Condition對象上面，都阻塞了多個執行緒。因此，在ConditionObject內部也有一個雙向鏈表組成的隊列，如下所示：

下面來看一下在await()/notify()方法中，是如何使用這個隊列的。

8.3.4 await()實現分析

關於await，有幾個關鍵點要說明：

執行緒調用 await()的時候，肯定已經先拿到了鎖。所以，在 addConditionWaiter()內部，對這個雙向鏈表的操作不需要執行CAS操作，執行緒天生是安全的，程式碼如下：
在執行緒執行wait操作之前，必須先釋放鎖。也就是fullyRelease(node)，否則會發生死鎖。這個和wait/notify與synchronized的配合機制一樣。
執行緒從wait中被喚醒後，必須用acquireQueued(node, savedState)方法重新拿鎖。
checkInterruptWhileWaiting(node)程式碼在park(this)程式碼之後，是為了檢測在park期間是否收到過中斷訊號。當執行緒從park中醒來時，有兩種可能：一種是其他執行緒調用了unpark，另一種是收到中斷訊號。這裡的await()方法是可以響應中斷的，所以當發現自己被中斷喚醒的，而不是被unpark喚醒的時，會直接退出while循環，await()方法也會返回。
isOnSyncQueue(node)用於判斷該Node是否在AQS的同步隊列裡面。初始的時候，Node只在Condition的隊列里，而不在AQS的隊列里。但執行notity操作的時候，會放進AQS的同步隊列。

8.3.5 awaitUninterruptibly()實現分析

與await()不同，awaitUninterruptibly()不會響應中斷，其方法的定義中不會有中斷異常拋出，下面分析其實現和await()的區別。

可以看出，整體程式碼和 await()類似，區別在於收到異常後，不會拋出異常，而是繼續執行while循環。

8.3.6 notify()實現分析

同 await()一樣，在調用 notify()的時候，必須先拿到鎖（否則就會拋出上面的異常），是因為前面執行await()的時候，把鎖釋放了。

然後，從隊列中取出firstWaiter，喚醒它。在通過調用unpark喚醒它之前，先用enq(node)方法把這個Node放入AQS的鎖對應的阻塞隊列中。也正因為如此，才有了await()方法裡面的判斷條件：

while( ! isOnSyncQueue(node))

這個判斷條件滿足，說明await執行緒不是被中斷，而是被unpark喚醒的。

notifyAll()與此類似。

8.4 StampedLock

8.4.1 為什麼引入StampedLock

StampedLock是在JDK8中新增的，有了讀寫鎖，為什麼還要引入StampedLock呢？

可以看到，從ReentrantLock到StampedLock，並發度依次提高。

另一方面，因為ReentrantReadWriteLock採用的是「悲觀讀」的策略，當第一個讀執行緒拿到鎖之後，第二個、第三個讀執行緒還可以拿到鎖，使得寫執行緒一直拿不到鎖，可能導致寫執行緒「餓死」。雖然在其公平或非公平的實現中，都盡量避免這種情形，但還有可能發生。

StampedLock引入了「樂觀讀」策略，讀的時候不加讀鎖，讀出來發現數據被修改了，再升級為「悲觀讀」，相當於降低了「讀」的地位，把搶鎖的天平往「寫」的一方傾斜了一下，避免寫執行緒被餓死。

8.4.2 使用場景

在剖析其原理之前，下面先以官方的一個例子來看一下StampedLock如何使用。

如上面程式碼所示，有一個Point類，多個執行緒調用move()方法，修改坐標；還有多個執行緒調用distanceFromOrigin()方法，求距離。

首先，執行move操作的時候，要加寫鎖。這個用法和ReadWriteLock的用法沒有區別，寫操作和寫操作也是互斥的。

關鍵在於讀的時候，用了一個「樂觀讀」sl.tryOptimisticRead()，相當於在讀之前給數據的狀態做了一個「快照」。然後，把數據拷貝到記憶體裡面，在用之前，再比對一次版本號。如果版本號變了，則說明在讀的期間有其他執行緒修改了數據。讀出來的數據廢棄，重新獲取讀鎖。關鍵程式碼就是下面這三行：

要說明的是，這三行關鍵程式碼對順序非常敏感，不能有重排序。因為 state 變數已經是volatile，所以可以禁止重排序，但stamp並不是volatile的。為此，在validate(stamp)方法裡面插入記憶體屏障。

8.4.3 “樂觀讀”的實現原理

首先，StampedLock是一個讀寫鎖，因此也會像讀寫鎖那樣，把一個state變數分成兩半，分別表示讀鎖和寫鎖的狀態。同時，它還需要一個數據的version。但是，一次CAS沒有辦法操作兩個變數，所以這個state變數本身同時也表示了數據的version。下面先分析state變數。

如下圖：用最低的8位表示讀和寫的狀態，其中第8位表示寫鎖的狀態，最低的7位表示讀鎖的狀態。因為寫鎖只有一個bit位，所以寫鎖是不可重入的。

初始值不為0，而是把WBIT 向左移動了一位，也就是上面的ORIGIN 常量，構造方法如下所示。

為什麼state的初始值不設為0呢？看樂觀鎖的實現：

上面兩個方法必須結合起來看：當state&WBIT != 0的時候，說明有執行緒持有寫鎖，上面的tryOptimisticRead會永遠返回0。這樣，再調用validate（stamp），也就是validate（0）也會永遠返回false。這正是我們想要的邏輯：當有執行緒持有寫鎖的時候，validate永遠返回false，無論寫執行緒是否釋放了寫鎖。因為無論是否釋放了（state回到初始值）寫鎖，state值都不為0，所以validate（0）永遠為false。

為什麼上面的validate(…)方法不直接比較stamp=state，而要比較state&SBITS=state&SBITS 呢？

因為讀鎖和讀鎖是不互斥的！

所以，即使在「樂觀讀」的時候，state 值被修改了，但如果它改的是第7位，validate(…)還是會返回true。

另外要說明的一點是，上面使用了記憶體屏障VarHandle.acquireFence();，是因為在這行程式碼的下一行裡面的stamp、SBITS變數不是volatile的，由此可以禁止其和前面的currentX=X，currentY=Y進行重排序。

通過上面的分析，可以發現state的設計非常巧妙。只通過一個變數，既實現了讀鎖、寫鎖的狀態記錄，還實現了數據的版本號的記錄。

8.4.4 悲觀讀/寫：「阻塞」與「自旋」策略實現差異

同ReadWriteLock一樣，StampedLock也要進行悲觀的讀鎖和寫鎖操作。不過，它不是基於AQS實現的，而是內部重新實現了一個阻塞隊列。如下所示。

這個阻塞隊列和 AQS 裡面的很像。

剛開始的時候，whead=wtail=NULL，然後初始化，建一個空節點，whead和wtail都指向這個空節點，之後往裡面加入一個個讀執行緒或寫執行緒節點。

但基於這個阻塞隊列實現的鎖的調度策略和AQS很不一樣，也就是「自旋」。

在AQS裡面，當一個執行緒CAS state失敗之後，會立即加入阻塞隊列，並且進入阻塞狀態。

但在StampedLock中，CAS state失敗之後，會不斷自旋，自旋足夠多的次數之後，如果還拿不到鎖，才進入阻塞狀態。

為此，根據CPU的核數，定義了自旋次數的常量值。如果是單核的CPU，肯定不能自旋，在多核情況下，才採用自旋策略。

下面以寫鎖的加鎖，也就是StampedLock的writeLock()方法為例，來看一下自旋的實現。

如上面程式碼所示，當state&ABITS==0的時候，說明既沒有執行緒持有讀鎖，也沒有執行緒持有寫鎖，此時當前執行緒才有資格通過CAS操作state。若操作不成功，則調用acquireWrite()方法進入阻塞隊列，並進行自旋，這個方法是整個加鎖操作的核心，程式碼如下：

整個acquireWrite(…)方法是兩個大的for循環，內部實現了非常複雜的自旋策略。在第一個大的for循環裡面，目的就是把該Node加入隊列的尾部，一邊加入，一邊通過CAS操作嘗試獲得鎖。如果獲得了，整個方法就會返回；如果不能獲得鎖，會一直自旋，直到加入隊列尾部。

在第二個大的for循環里，也就是該Node已經在隊列尾部了。這個時候，如果發現自己剛好也在隊列頭部，說明隊列中除了空的Head節點，就是當前執行緒了。此時，再進行新一輪的自旋，直到達到MAX_HEAD_SPINS次數，然後進入阻塞。這裡有一個關鍵點要說明：當release(…)方法被調用之後，會喚醒隊列頭部的第1個元素，此時會執行第二個大的for循環裡面的邏輯，也就是接著for循環裡面park()方法後面的程式碼往下執行。

另外一個不同於AQS的阻塞隊列的地方是，在每個WNode裡面有一個cowait指針，用於串聯起所有的讀執行緒。例如，隊列尾部阻塞的是一個讀執行緒 1，現在又來了讀執行緒 2、3，那麼會通過cowait指針，把1、2、3串聯起來。1被喚醒之後，2、3也隨之一起被喚醒，因為讀和讀之間不互斥。

明白加鎖的自旋策略後，下面來看鎖的釋放操作。和讀寫鎖的實現類似，也是做了兩件事情：一是把state變數置回原位，二是喚醒阻塞隊列中的第一個節點。

Tags: juc 並發編程