並發編程從零開始（十二）-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變量置回原位，二是喚醒阻塞隊列中的第一個節點。
