Java多线程(五) —— 线程并发库之锁机制
參考文獻(xiàn)：
http://www.blogjava.net/xylz/archive/2010/07/08/325587.html
一、Lock與ReentrantLock
前面的章節(jié)主要談?wù)勗硬僮?#xff0c;至于與原子操作一些相關(guān)的問題或者說陷阱就放到最后的總結(jié)篇來整體說明。從這一章開始花少量的篇幅談?wù)勬i機(jī)制。
上一個章節(jié)中談到了鎖機(jī)制，并且針對于原子操作談了一些相關(guān)的概念和設(shè)計思想。接下來的文章中，盡可能的深入研究鎖機(jī)制，并且理解里面的原理和實(shí)際應(yīng)用場合。
盡管synchronized在語法上已經(jīng)足夠簡單了，在JDK 5之前只能借助此實(shí)現(xiàn)，但是由于是獨(dú)占鎖，性能卻不高，因此JDK 5以后就開始借助于JNI來完成更高級的鎖實(shí)現(xiàn)。
JDK 5中的鎖是接口java.util.concurrent.locks.Lock。另外java.util.concurrent.locks.ReadWriteLock提供了一對可供讀寫并發(fā)的鎖。根據(jù)前面的規(guī)則，我們從java.util.concurrent.locks.Lock的API開始。
?
void lock();
獲取鎖。
如果鎖不可用，出于線程調(diào)度目的，將禁用當(dāng)前線程，并且在獲得鎖之前，該線程將一直處于休眠狀態(tài)。
void lockInterruptibly() throws InterruptedException;
如果當(dāng)前線程未被中斷，則獲取鎖。
如果鎖可用，則獲取鎖，并立即返回。
如果鎖不可用，出于線程調(diào)度目的，將禁用當(dāng)前線程，并且在發(fā)生以下兩種情況之一以前，該線程將一直處于休眠狀態(tài)：
- 鎖由當(dāng)前線程獲得；或者
- 其他某個線程中斷當(dāng)前線程，并且支持對鎖獲取的中斷。
如果當(dāng)前線程：
- 在進(jìn)入此方法時已經(jīng)設(shè)置了該線程的中斷狀態(tài)；或者
- 在獲取鎖時被中斷，并且支持對鎖獲取的中斷，
則將拋出?InterruptedException，并清除當(dāng)前線程的已中斷狀態(tài)。
Condition newCondition();
返回綁定到此?Lock?實(shí)例的新?Condition?實(shí)例。下一小節(jié)中會重點(diǎn)談Condition，此處不做過多的介紹。
boolean tryLock();
僅在調(diào)用時鎖為空閑狀態(tài)才獲取該鎖。
如果鎖可用，則獲取鎖，并立即返回值?true。如果鎖不可用，則此方法將立即返回值?false。
通常對于那些不是必須獲取鎖的操作可能有用。
Lock lock = ...;if (lock.tryLock()) {try {// manipulate protected state} finally {lock.unlock();}} else {// perform alternative actions}?
boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
如果鎖在給定的等待時間內(nèi)空閑，并且當(dāng)前線程未被中斷，則獲取鎖。
如果鎖可用，則此方法將立即返回值?true。如果鎖不可用，出于線程調(diào)度目的，將禁用當(dāng)前線程，并且在發(fā)生以下三種情況之一前，該線程將一直處于休眠狀態(tài)：
- 鎖由當(dāng)前線程獲得；或者
- 其他某個線程中斷當(dāng)前線程，并且支持對鎖獲取的中斷；或者
- 已超過指定的等待時間
如果獲得了鎖，則返回值?true。
如果當(dāng)前線程：
- 在進(jìn)入此方法時已經(jīng)設(shè)置了該線程的中斷狀態(tài)；或者
- 在獲取鎖時被中斷，并且支持對鎖獲取的中斷，
則將拋出?InterruptedException，并會清除當(dāng)前線程的已中斷狀態(tài)。
如果超過了指定的等待時間，則將返回值?false。如果 time 小于等于 0，該方法將完全不等待。
void unlock();
釋放鎖。對應(yīng)于lock()、tryLock()、tryLock(xx)、lockInterruptibly()等操作，如果成功的話應(yīng)該對應(yīng)著一個unlock()，這樣可以避免死鎖或者資源浪費(fèi)。
?
相對于比較空洞的API，來看一個實(shí)際的例子。下面的代碼實(shí)現(xiàn)了一個類似于AtomicInteger的操作。
package xylz.study.concurrency.lock;import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock;public class AtomicIntegerWithLock {private int value;private Lock lock = new ReentrantLock();public AtomicIntegerWithLock() {super();}public AtomicIntegerWithLock(int value) {this.value = value;}public final int get() {lock.lock();try {return value;} finally {lock.unlock();}}public final void set(int newValue) {lock.lock();try {value = newValue;} finally {lock.unlock();}}public final int getAndSet(int newValue) {lock.lock();try {int ret = value;value = newValue;return ret;} finally {lock.unlock();}}public final boolean compareAndSet(int expect, int update) {lock.lock();try {if (value == expect) {value = update;return true;}return false;} finally {lock.unlock();}}public final int getAndIncrement() {lock.lock();try {return value++;} finally {lock.unlock();}}public final int getAndDecrement() {lock.lock();try {return value--;} finally {lock.unlock();}}public final int incrementAndGet() {lock.lock();try {return ++value;} finally {lock.unlock();}}public final int decrementAndGet() {lock.lock();try {return --value;} finally {lock.unlock();}}public String toString() {return Integer.toString(get());} }?
類AtomicIntegerWithLock是線程安全的，此結(jié)構(gòu)中大量使用了Lock對象的lock/unlock方法對。同樣可以看到的是對于自增和自減操作使用了++/--。之所以能夠保證線程安全，是因為Lock對象的lock()方法保證了只有一個線程能夠只有此鎖。需要說明的是對于任何一個lock()方法，都需要一個unlock()方法與之對于，通常情況下為了保證unlock方法總是能夠得到執(zhí)行，unlock方法被置于finally塊中。另外這里使用了java.util.concurrent.locks.ReentrantLock.ReentrantLock對象，下一個小節(jié)中會具體描述此類作為Lock的唯一實(shí)現(xiàn)是如何設(shè)計和實(shí)現(xiàn)的。
盡管synchronized實(shí)現(xiàn)Lock的相同語義，并且在語法上比Lock要簡單多，但是前者卻比后者的開銷要大得多。做一個簡單的測試。
public static void main(String[] args) throws Exception{final int max = 10;final int loopCount = 100000;long costTime = 0;for (int m = 0; m < max; m++) {long start1 = System.nanoTime();final AtomicIntegerWithLock value1 = new AtomicIntegerWithLock(0);Thread[] ts = new Thread[max];for(int i=0;i<max;i++) {ts[i] = new Thread() {public void run() {for (int i = 0; i < loopCount; i++) {value1.incrementAndGet();}}};}for(Thread t:ts) {t.start();}for(Thread t:ts) {t.join();}long end1 = System.nanoTime();costTime += (end1-start1);}System.out.println("cost1: " + (costTime));//System.out.println();costTime = 0;//final Object lock = new Object();for (int m = 0; m < max; m++) {staticValue=0;long start1 = System.nanoTime();Thread[] ts = new Thread[max];for(int i=0;i<max;i++) {ts[i] = new Thread() {public void run() {for (int i = 0; i < loopCount; i++) {synchronized(lock) {++staticValue;}}}};}for(Thread t:ts) {t.start();}for(Thread t:ts) {t.join();}long end1 = System.nanoTime();costTime += (end1-start1);}//System.out.println("cost2: " + (costTime)); }static int staticValue = 0;?
?
在這個例子中每次啟動10個線程，每個線程計算100000次自增操作，重復(fù)測試10次，下面是某此測試的結(jié)果：
cost1: 624071136
cost2: 2057847833
盡管上面的例子不是非常正式的測試案例，但上面的例子在于說明，Lock的性能比synchronized的要好得多。如果可以的話總是使用Lock替代synchronized是一個明智的選擇。
二、AQS
AbstractQueuedSynchronizer，簡稱AQS，是J.U.C最復(fù)雜的一個類，導(dǎo)致絕大多數(shù)講解并發(fā)原理或者實(shí)戰(zhàn)的時候都不會提到此類。但是虛心的作者愿意借助自己有限的能力和精力來探討一二（參考資源中也有一些作者做了部分的分析。）。
首先從理論知識開始，在了解了相關(guān)原理后會針對源碼進(jìn)行一些分析，最后加上一些實(shí)戰(zhàn)來描述。
上面的繼承體系中，AbstractQueuedSynchronizer是CountDownLatch/FutureTask/ReentrantLock/RenntrantReadWriteLock/Semaphore的基礎(chǔ)，因此AbstractQueuedSynchronizer是Lock/Executor實(shí)現(xiàn)的前提。公平鎖、不公平鎖、Condition、CountDownLatch、Semaphore等放到后面的篇幅中說明。
完整的設(shè)計原理可以參考Doug Lea的論文?The java.util.concurrent Synchronizer Framework?，這里做一些簡要的分析。
基本的思想是表現(xiàn)為一個同步器，支持下面兩個操作：
獲取鎖：首先判斷當(dāng)前狀態(tài)是否允許獲取鎖，如果允許就獲取鎖，否則就阻塞操作或者獲取失敗，也就是說如果是獨(dú)占鎖就可能阻塞，如果是共享鎖就可能失敗。另外如果是阻塞線程，那么線程就需要進(jìn)入阻塞隊列。當(dāng)狀態(tài)位允許獲取鎖時就修改狀態(tài)，并且如果進(jìn)了隊列就從隊列中移除。
while(synchronization state does not allow acquire){
??? enqueue current thread if not already queued;
??? possibly block current thread;
}
dequeue current thread if it was queued;
釋放鎖:這個過程就是修改狀態(tài)位，如果有線程因為狀態(tài)位阻塞的話就喚醒隊列中的一個或者更多線程。
update synchronization state;
if(state may permit a blocked thread to acquire)
??? unlock one or more queued threads;
要支持上面兩個操作就必須有下面的條件：
- 原子性操作同步器的狀態(tài)位
- 阻塞和喚醒線程
- 一個有序的隊列
目標(biāo)明確，要解決的問題也清晰了，那么剩下的就是解決上面三個問題。
狀態(tài)位的原子操作
這里使用一個32位的整數(shù)來描述狀態(tài)位，前面章節(jié)的原子操作的理論知識整好派上用場，在這里依然使用CAS操作來解決這個問題。事實(shí)上這里還有一個64位版本的同步器（AbstractQueuedLongSynchronizer），這里暫且不談。
阻塞和喚醒線程
標(biāo)準(zhǔn)的JAVA API里面是無法掛起（阻塞）一個線程，然后在將來某個時刻再喚醒它的。JDK 1.0的API里面有Thread.suspend和Thread.resume，并且一直延續(xù)了下來。但是這些都是過時的API，而且也是不推薦的做法。
在JDK 5.0以后利用JNI在LockSupport類中實(shí)現(xiàn)了此特性。
LockSupport.park()
LockSupport.park(Object)
LockSupport.parkNanos(Object, long)
LockSupport.parkNanos(long)
LockSupport.parkUntil(Object, long)
LockSupport.parkUntil(long)
LockSupport.unpark(Thread)
上面的API中park()是在當(dāng)前線程中調(diào)用，導(dǎo)致線程阻塞，帶參數(shù)的Object是掛起的對象，這樣監(jiān)視的時候就能夠知道此線程是因為什么資源而阻塞的。由于park()立即返回，所以通常情況下需要在循環(huán)中去檢測競爭資源來決定是否進(jìn)行下一次阻塞。park()返回的原因有三：
- 其他某個線程調(diào)用將當(dāng)前線程作為目標(biāo)調(diào)用?unpark；
- 其他某個線程中斷當(dāng)前線程；
- 該調(diào)用不合邏輯地（即毫無理由地）返回。
其實(shí)第三條就決定了需要循環(huán)檢測了，類似于通常寫的while(checkCondition()){Thread.sleep(time);}類似的功能。
有序隊列
在AQS中采用CHL列表來解決有序的隊列的問題。
AQS采用的CHL模型采用下面的算法完成FIFO的入隊列和出隊列過程。
對于入隊列(enqueue)：采用CAS操作，每次比較尾結(jié)點(diǎn)是否一致，然后插入的到尾結(jié)點(diǎn)中。
do {
??????? pred = tail;
}while ( !compareAndSet(pred,tail,node) );
對于出隊列(dequeue):由于每一個節(jié)點(diǎn)也緩存了一個狀態(tài)，決定是否出隊列，因此當(dāng)不滿足條件時就需要自旋等待，一旦滿足條件就將頭結(jié)點(diǎn)設(shè)置為下一個節(jié)點(diǎn)。
while (pred.status != RELEASED) ;
head? = node;
實(shí)際上這里自旋等待也是使用LockSupport.park()來實(shí)現(xiàn)的。
AQS里面有三個核心字段：
private volatile int state;
private transient volatile Node head;
private transient volatile Node tail;
其中state描述的有多少個線程取得了鎖，對于互斥鎖來說state<=1。head/tail加上CAS操作就構(gòu)成了一個CHL的FIFO隊列。下面是Node節(jié)點(diǎn)的屬性。
volatile int waitStatus;?節(jié)點(diǎn)的等待狀態(tài)，一個節(jié)點(diǎn)可能位于以下幾種狀態(tài)：
- CANCELLED = 1： 節(jié)點(diǎn)操作因為超時或者對應(yīng)的線程被interrupt。節(jié)點(diǎn)不應(yīng)該留在此狀態(tài)，一旦達(dá)到此狀態(tài)將從CHL隊列中踢出。
- SIGNAL = -1： 節(jié)點(diǎn)的繼任節(jié)點(diǎn)是（或者將要成為）BLOCKED狀態(tài)（例如通過LockSupport.park()操作），因此一個節(jié)點(diǎn)一旦被釋放（解鎖）或者取消就需要喚醒（LockSupport.unpack()）它的繼任節(jié)點(diǎn)。
- CONDITION = -2：表明節(jié)點(diǎn)對應(yīng)的線程因為不滿足一個條件（Condition）而被阻塞。
- 0： 正常狀態(tài)，新生的非CONDITION節(jié)點(diǎn)都是此狀態(tài)。
- 非負(fù)值標(biāo)識節(jié)點(diǎn)不需要被通知（喚醒）。
volatile Node prev;此節(jié)點(diǎn)的前一個節(jié)點(diǎn)。節(jié)點(diǎn)的waitStatus依賴于前一個節(jié)點(diǎn)的狀態(tài)。
volatile Node next;此節(jié)點(diǎn)的后一個節(jié)點(diǎn)。后一個節(jié)點(diǎn)是否被喚醒（uppark()）依賴于當(dāng)前節(jié)點(diǎn)是否被釋放。
volatile Thread thread;節(jié)點(diǎn)綁定的線程。
Node nextWaiter;下一個等待條件（Condition）的節(jié)點(diǎn)，由于Condition是獨(dú)占模式，因此這里有一個簡單的隊列來描述Condition上的線程節(jié)點(diǎn)。
?
AQS 在J.U.C里面是一個非常核心的工具，而且也非常復(fù)雜，里面考慮到了非常多的邏輯實(shí)現(xiàn)，所以在后面的章節(jié)中總是不斷的嘗試介紹AQS的特性和實(shí)現(xiàn)。
這一個小節(jié)主要介紹了一些理論背景和相關(guān)的數(shù)據(jù)結(jié)構(gòu)，在下一個小節(jié)中將根據(jù)以上知識來了解Lock.lock/unlock是如何實(shí)現(xiàn)的。
三、加鎖的原理lock? unlock
接上篇，這篇從Lock.lock/unlock開始。特別說明在沒有特殊情況下所有程序、API、文檔都是基于JDK 6.0的。
public void java.util.concurrent.locks.ReentrantLock.lock()
獲取鎖。
如果該鎖沒有被另一個線程保持，則獲取該鎖并立即返回，將鎖的保持計數(shù)設(shè)置為 1。
如果當(dāng)前線程已經(jīng)保持該鎖，則將保持計數(shù)加 1，并且該方法立即返回。
如果該鎖被另一個線程保持，則出于線程調(diào)度的目的，禁用當(dāng)前線程，并且在獲得鎖之前，該線程將一直處于休眠狀態(tài)，此時鎖保持計數(shù)被設(shè)置為 1。
從上面的文檔可以看出ReentrantLock是可重入鎖的實(shí)現(xiàn)。而內(nèi)部是委托java.util.concurrent.locks.ReentrantLock.Sync.lock()實(shí)現(xiàn)的。java.util.concurrent.locks.ReentrantLock.Sync是抽象類，有java.util.concurrent.locks.ReentrantLock.FairSync和java.util.concurrent.locks.ReentrantLock.NonfairSync兩個實(shí)現(xiàn)，也就是常說的公平鎖和不公平鎖。
1.公平鎖和非公平鎖
如果獲取一個鎖是按照請求的順序得到的，那么就是公平鎖，否則就是非公平鎖。
在沒有深入了解內(nèi)部機(jī)制及實(shí)現(xiàn)之前，先了解下為什么會存在公平鎖和非公平鎖。公平鎖保證一個阻塞的線程最終能夠獲得鎖，因為是有序的，所以總是可以按照請求的順序獲得鎖。不公平鎖意味著后請求鎖的線程可能在其前面排列的休眠線程恢復(fù)前拿到鎖，這樣就有可能提高并發(fā)的性能。這是因為通常情況下掛起的線程重新開始與它真正開始運(yùn)行，二者之間會產(chǎn)生嚴(yán)重的延時。因此非公平鎖就可以利用這段時間完成操作。這是非公平鎖在某些時候比公平鎖性能要好的原因之一。
二者在實(shí)現(xiàn)上的區(qū)別會在后面介紹，我們先從公平鎖（FairSync）開始。
前面說過java.util.concurrent.locks.AbstractQueuedSynchronizer （AQS)是Lock的基礎(chǔ)，對于一個FairSync而言，lock()就直接調(diào)用AQS的acquire(int arg);
public final void acquire(int arg)?以獨(dú)占模式獲取對象，忽略中斷。通過至少調(diào)用一次?tryAcquire(int)?來實(shí)現(xiàn)此方法，并在成功時返回。否則在成功之前，一直調(diào)用?tryAcquire(int)?將線程加入隊列，線程可能重復(fù)被阻塞或不被阻塞。
在介紹實(shí)現(xiàn)之前先要補(bǔ)充上一節(jié)的知識，對于一個AQS的實(shí)現(xiàn)而言，通常情況下需要實(shí)現(xiàn)以下方法來描述如何鎖定線程。
- tryAcquire(int)?試圖在獨(dú)占模式下獲取對象狀態(tài)。此方法應(yīng)該查詢是否允許它在獨(dú)占模式下獲取對象狀態(tài)，如果允許，則獲取它。
此方法總是由執(zhí)行 acquire 的線程來調(diào)用。如果此方法報告失敗，則 acquire 方法可以將線程加入隊列（如果還沒有將它加入隊列），直到獲得其他某個線程釋放了該線程的信號。也就是說此方法是一種嘗試性方法，如果成功獲取鎖那最好，如果沒有成功也沒有關(guān)系，直接返回false。
- tryRelease(int)?試圖設(shè)置狀態(tài)來反映獨(dú)占模式下的一個釋放。 此方法總是由正在執(zhí)行釋放的線程調(diào)用。釋放鎖可能失敗或者拋出異常，這個在后面會具體分析。
- tryAcquireShared(int)?試圖在共享模式下獲取對象狀態(tài)。
- tryReleaseShared(int)?試圖設(shè)置狀態(tài)來反映共享模式下的一個釋放。
- isHeldExclusively()?如果對于當(dāng)前（正調(diào)用的）線程，同步是以獨(dú)占方式進(jìn)行的，則返回?true。
除了tryAcquire(int)外，其它方法會在后面具體介紹。首先對于ReentrantLock而言，不管是公平鎖還是非公平鎖，都是獨(dú)占鎖，也就是說同時能夠有一個線程持有鎖。因此對于acquire(int arg)而言，arg==1。在AQS中acquire的實(shí)現(xiàn)如下：
public final void acquire(int arg) {
??? if (!tryAcquire(arg) &&
??????? acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
??????? selfInterrupt();
}
這個看起來比較復(fù)雜，我們分解以下4個步驟。
這是一個比較復(fù)雜的過程，我們按部就班一個一個分析。
tryAcquire(acquires)
對于公平鎖而言，它的實(shí)現(xiàn)方式如下：
??? protected final boolean tryAcquire(int acquires) {
??????? final Thread current = Thread.currentThread();
??????? int c = getState();
??????? if (c == 0) {
??????????? if (isFirst(current) &&
??????????????? compareAndSetState(0, acquires)) {
??????????????? setExclusiveOwnerThread(current);
??????????????? return true;
??????????? }
??????? }
??????? else if (current == getExclusiveOwnerThread()) {
??????????? int nextc = c + acquires;
??????????? if (nextc < 0)
??????????????? throw new Error("Maximum lock count exceeded");
??????????? setState(nextc);
??????????? return true;
??????? }
??????? return false;
??? }
}
在這段代碼中，前面說明對于AQS存在一個state來描述當(dāng)前有多少線程持有鎖。由于AQS支持共享鎖（例如讀寫鎖，后面會繼續(xù)講），所以這里state>=0，但是由于ReentrantLock是獨(dú)占鎖，所以這里不妨理解為0<=state，acquires=1。isFirst(current)是一個很復(fù)雜的邏輯，包括踢出無用的節(jié)點(diǎn)等復(fù)雜過程，這里暫且不提，大體上的意思是說判斷AQS是否為空或者當(dāng)前線程是否在隊列頭（為了區(qū)分公平與非公平鎖）。
比較非公平鎖的tryAcquire實(shí)現(xiàn)java.util.concurrent.locks.ReentrantLock.Sync.nonfairTryAcquire(int)，公平鎖多了一個判斷當(dāng)前節(jié)點(diǎn)是否在隊列頭，這個就保證了是否按照請求鎖的順序來決定獲取鎖的順序（同一個線程的多次獲取鎖除外）。
現(xiàn)在再回頭看公平鎖和非公平鎖的lock()方法。公平鎖只有一句acquire(1)；而非公平鎖的調(diào)用如下：
final void lock() {
??? if (compareAndSetState(0, 1))
??????? setExclusiveOwnerThread(Thread.currentThread());
??? else
??????? acquire(1);
}
很顯然，非公平鎖在第一次獲取鎖，或者其它線程釋放鎖后（可能等待），優(yōu)先采用compareAndSetState(0,1)然后設(shè)置AQS獨(dú)占線程而持有鎖，這樣有時候比acquire(1)順序檢查鎖持有而要高效。即使在重入鎖上，也就是compareAndSetState(0,1)失敗，但是是當(dāng)前線程持有鎖上，非公平鎖也沒有問題。
addWaiter(mode)
tryAcquire失敗就意味著入隊列了。此時AQS的隊列中節(jié)點(diǎn)Node就開始發(fā)揮作用了。一般情況下AQS支持獨(dú)占鎖和共享鎖，而獨(dú)占鎖在Node中就意味著條件（Condition）隊列為空（上一篇中介紹過相關(guān)概念）。在java.util.concurrent.locks.AbstractQueuedSynchronizer.Node中有兩個常量，
static final Node EXCLUSIVE = null; //獨(dú)占節(jié)點(diǎn)模式
static final Node SHARED = new Node(); //共享節(jié)點(diǎn)模式
addWaiter(mode)中的mode就是節(jié)點(diǎn)模式，也就是共享鎖還是獨(dú)占鎖模式。
前面一再強(qiáng)調(diào)ReentrantLock是獨(dú)占鎖模式。
private Node addWaiter(Node mode) {
???? Node node = new Node(Thread.currentThread(), mode);
???? // Try the fast path of enq; backup to full enq on failure
???? Node pred = tail;
???? if (pred != null) {
???????? node.prev = pred;
???????? if (compareAndSetTail(pred, node)) {
???????????? pred.next = node;
???????????? return node;
???????? }
???? }
???? enq(node);
???? return node;
}
上面是節(jié)點(diǎn)如隊列的一部分。當(dāng)前僅當(dāng)隊列不為空并且將新節(jié)點(diǎn)插入尾部成功后直接返回新節(jié)點(diǎn)。否則進(jìn)入enq(Node)進(jìn)行操作。
private Node enq(final Node node) {
??? for (;;) {
??????? Node t = tail;
??????? if (t == null) { // Must initialize
??????????? Node h = new Node(); // Dummy header
??????????? h.next = node;
??????????? node.prev = h;
??????????? if (compareAndSetHead(h)) {
??????????????? tail = node;
??????????????? return h;
??????????? }
??????? }
??????? else {
??????????? node.prev = t;
??????????? if (compareAndSetTail(t, node)) {
??????????????? t.next = node;
??????????????? return t;
??????????? }
??????? }
??? }
}
enq(Node)去隊列操作實(shí)現(xiàn)了CHL隊列的算法，如果為空就創(chuàng)建頭結(jié)點(diǎn)，然后同時比較節(jié)點(diǎn)尾部是否是改變來決定CAS操作是否成功，當(dāng)且僅當(dāng)成功后才將為不節(jié)點(diǎn)的下一個節(jié)點(diǎn)指向為新節(jié)點(diǎn)。可以看到這里仍然是CAS操作。
acquireQueued(node,arg)
自旋請求鎖，如果可能的話掛起線程，直到得到鎖，返回當(dāng)前線程是否中斷過（如果park()過并且中斷過的話有一個interrupted中斷位）。
final boolean acquireQueued(final Node node, int arg) {
??? try {
??????? boolean interrupted = false;
??????? for (;;) {
??????????? final Node p = node.predecessor();
??????????? if (p == head && tryAcquire(arg)) {
??????????????? setHead(node);
??????????????? p.next = null; // help GC
??????????????? return interrupted;
??????????? }
??????????? if (shouldParkAfterFailedAcquire(p, node) &&
??????????????? parkAndCheckInterrupt())
??????????????? interrupted = true;
??????? }
??? } catch (RuntimeException ex) {
??????? cancelAcquire(node);
??????? throw ex;
??? }
}
下面的分析就需要用到上節(jié)節(jié)點(diǎn)的狀態(tài)描述了。acquireQueued過程是這樣的：
一個節(jié)點(diǎn)是否該park()是關(guān)鍵，這是由方法java.util.concurrent.locks.AbstractQueuedSynchronizer.shouldParkAfterFailedAcquire(Node, Node)實(shí)現(xiàn)的。
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
??? int s = pred.waitStatus;
??? if (s < 0) return true;
??? if (s > 0) {
??????? do {
??????????? node.prev = pred = pred.prev;
??????? } while (pred.waitStatus > 0);
??????? pred.next = node;
??? } else compareAndSetWaitStatus(pred, 0, Node.SIGNAL);
??? return false;
}
selfInterrupt()
private static void selfInterrupt() {
??? Thread.currentThread().interrupt();
}
如果線程曾經(jīng)中斷過（或者阻塞過）（比如手動interrupt()或者超時等等，那么就再中斷一次，中斷兩次的意思就是清除中斷位）。
大體上整個Lock.lock()就這樣一個流程。除了lock()方法外，還有l(wèi)ockInterruptibly()/tryLock()/unlock()/newCondition()等，在接下來的章節(jié)中會一一介紹。
四、鎖釋放與條件變量
?
本小節(jié)介紹鎖釋放Lock.unlock()。
Release/TryRelease
unlock操作實(shí)際上就調(diào)用了AQS的release操作，釋放持有的鎖。
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)操作，此操作里面總是嘗試去釋放鎖，如果成功，說明鎖確實(shí)被當(dāng)前線程持有，那么就看AQS隊列中的頭結(jié)點(diǎn)是否為空并且能否被喚醒，如果可以的話就喚醒繼任節(jié)點(diǎn)（下一個非CANCELLED節(jié)點(diǎn)，下面會具體分析）。
對于獨(dú)占鎖而言，java.util.concurrent.locks.ReentrantLock.Sync.tryRelease(int)展示了如何嘗試釋放鎖(tryRelease)操作。
protected final boolean tryRelease(int releases) {
??? int c = getState() - releases;
??? if (Thread.currentThread() != getExclusiveOwnerThread())
??????? throw new IllegalMonitorStateException();
??? boolean free = false;
??? if (c == 0) {
??????? free = true;
??????? setExclusiveOwnerThread(null);
??? }
??? setState(c);
??? return free;
}
整個tryRelease操作是這樣的：
參考上一節(jié)的分析就可以知道，這里c==0決定了是否完全釋放了鎖。由于ReentrantLock是可重入鎖，因此同一個線程可能多重持有鎖，那么當(dāng)且僅當(dāng)最后一個持有鎖的線程釋放鎖是才能將AQS中持有鎖的獨(dú)占線程清空，這樣接下來的操作才需要喚醒下一個需要鎖的AQS節(jié)點(diǎn)（Node），否則就只是減少鎖持有的計數(shù)器，并不能改變其他操作。
當(dāng)tryRelease操作成功后（也就是完全釋放了鎖），release操作才能檢查是否需要喚醒下一個繼任節(jié)點(diǎn)。這里的前提是AQS隊列的頭結(jié)點(diǎn)需要鎖(waitStatus!=0)，如果頭結(jié)點(diǎn)需要鎖，就開始檢測下一個繼任節(jié)點(diǎn)是否需要鎖操作。
在上一節(jié)中說道acquireQueued操作完成后（拿到了鎖），會將當(dāng)前持有鎖的節(jié)點(diǎn)設(shè)為頭結(jié)點(diǎn)，所以一旦頭結(jié)點(diǎn)釋放鎖，那么就需要尋找頭結(jié)點(diǎn)的下一個需要鎖的繼任節(jié)點(diǎn)，并喚醒它。
private void unparkSuccessor(Node node) {
??????? //此時node是需要是需要釋放鎖的頭結(jié)點(diǎn)
??????? //清空頭結(jié)點(diǎn)的waitStatus，也就是不再需要鎖了
??????? compareAndSetWaitStatus(node, Node.SIGNAL, 0);
??????? //從頭結(jié)點(diǎn)的下一個節(jié)點(diǎn)開始尋找繼任節(jié)點(diǎn)，當(dāng)且僅當(dāng)繼任節(jié)點(diǎn)的waitStatus<=0才是有效繼任節(jié)點(diǎn)，否則將這些waitStatus>0（也就是CANCELLED的節(jié)點(diǎn)）從AQS隊列中剔除??
? ? ? ?//這里并沒有從head->tail開始尋找，而是從tail->head尋找最后一個有效節(jié)點(diǎn)。
? ? ? ?//解釋在這里 http://www.blogjava.net/xylz/archive/2010/07/08/325540.html#377512
??????? Node s = node.next;
??????? if (s == null || s.waitStatus > 0) {
??????????? s = null;
??????????? for (Node t = tail; t != null && t != node; t = t.prev)
??????????????? if (t.waitStatus <= 0)
??????????????????? s = t;
??????? }
??????? //如果找到一個有效的繼任節(jié)點(diǎn)，就喚醒此節(jié)點(diǎn)線程
??????? if (s != null)
??????????? LockSupport.unpark(s.thread);
??? }
這里再一次把acquireQueued的過程找出來。對比unparkSuccessor，一旦頭節(jié)點(diǎn)的繼任節(jié)點(diǎn)被喚醒，那么繼任節(jié)點(diǎn)就會嘗試去獲取鎖（在acquireQueued中node就是有效的繼任節(jié)點(diǎn)，p就是喚醒它的頭結(jié)點(diǎn)），如果成功就會將頭結(jié)點(diǎn)設(shè)置為自身，并且將頭結(jié)點(diǎn)的前任節(jié)點(diǎn)清空，這樣前任節(jié)點(diǎn)（已經(jīng)過時了）就可以被GC釋放了。
final boolean acquireQueued(final Node node, int arg) {
??? try {
??????? boolean interrupted = false;
??????? for (;;) {
??????????? final Node p = node.predecessor();
??????????? if (p == head && tryAcquire(arg)) {
??????????????? setHead(node);
??????????????? p.next = null; // help GC
??????????????? return interrupted;
??????????? }
??????????? if (shouldParkAfterFailedAcquire(p, node) &&
??????????????? parkAndCheckInterrupt())
??????????????? interrupted = true;
??????? }
??? } catch (RuntimeException ex) {
??????? cancelAcquire(node);
??????? throw ex;
??? }
}
在setHead中，將頭結(jié)點(diǎn)的前任節(jié)點(diǎn)清空并且將頭結(jié)點(diǎn)的線程清空就是為了更好的GC，防止內(nèi)存泄露。
private void setHead(Node node) {
??? head = node;
??? node.thread = null;
??? node.prev = null;
}
對比lock()操作，unlock()操作還是比較簡單的，主要就是釋放響應(yīng)的資源，并且喚醒AQS隊列中有效的繼任節(jié)點(diǎn)。這樣所就按照請求的順序去嘗試獲取鎖了。
整個lock()/unlock()過程完成了，我們再回頭看公平鎖(FairSync)和非公平鎖(NonfairSync)。
公平鎖和非公平鎖只是在獲取鎖的時候有差別，其它都是一樣的。
final void lock() {
??? if (compareAndSetState(0, 1))
??????? setExclusiveOwnerThread(Thread.currentThread());
??? else
??????? acquire(1);
}
在上面非公平鎖的代碼中總是優(yōu)先嘗試當(dāng)前是否有線程持有鎖，一旦沒有任何線程持有鎖，那么非公平鎖就霸道的嘗試將鎖“占為己有”。如果在搶占鎖的時候失敗就和公平鎖一樣老老實(shí)實(shí)的去排隊。
也即是說公平鎖和非公平鎖只是在入AQS的CLH隊列之前有所差別，一旦進(jìn)入了隊列，所有線程都是按照隊列中先來后到的順序請求鎖。
Condition
條件變量很大一個程度上是為了解決Object.wait/notify/notifyAll難以使用的問題。
條件（也稱為條件隊列?或條件變量）為線程提供了一個含義，以便在某個狀態(tài)條件現(xiàn)在可能為 true 的另一個線程通知它之前，一直掛起該線程（即讓其“等待”）。因為訪問此共享狀態(tài)信息發(fā)生在不同的線程中，所以它必須受保護(hù)，因此要將某種形式的鎖與該條件相關(guān)聯(lián)。等待提供一個條件的主要屬性是：以原子方式?釋放相關(guān)的鎖，并掛起當(dāng)前線程，就像?Object.wait?做的那樣。
上述API說明表明條件變量需要與鎖綁定，而且多個Condition需要綁定到同一鎖上。前面的Lock中提到，獲取一個條件變量的方法是Lock.newCondition()。
void await() throws InterruptedException;
void awaitUninterruptibly();
long awaitNanos(long nanosTimeout) throws InterruptedException;
boolean await(long time, TimeUnit unit) throws InterruptedException;
boolean awaitUntil(Date deadline) throws InterruptedException;
void signal();
void signalAll();
以上是Condition接口定義的方法，await*對應(yīng)于Object.wait，signal對應(yīng)于Object.notify，signalAll對應(yīng)于Object.notifyAll。特別說明的是Condition的接口改變名稱就是為了避免與Object中的wait/notify/notifyAll的語義和使用上混淆，因為Condition同樣有wait/notify/notifyAll方法。
每一個Lock可以有任意數(shù)據(jù)的Condition對象，Condition是與Lock綁定的，所以就有Lock的公平性特性：如果是公平鎖，線程為按照FIFO的順序從Condition.await中釋放，如果是非公平鎖，那么后續(xù)的鎖競爭就不保證FIFO順序了。
一個使用Condition實(shí)現(xiàn)生產(chǎn)者消費(fèi)者的模型例子如下。
package xylz.study.concurrency.lock;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class ProductQueue<T> {
??? private final T[] items;
??? private final Lock lock = new ReentrantLock();
??? private Condition notFull = lock.newCondition();
??? private Condition notEmpty = lock.newCondition();
??? //
??? private int head, tail, count;
??? public ProductQueue(int maxSize) {
??????? items = (T[]) new Object[maxSize];
??? }
??? public ProductQueue() {
??????? this(10);
??? }
??? public void put(T t) throws InterruptedException {
??????? lock.lock();
??????? try {
??????????? while (count == getCapacity()) {
??????????????? notFull.await();
??????????? }
??????????? items[tail] = t;
??????????? if (++tail == getCapacity()) {
??????????????? tail = 0;
??????????? }
??????????? ++count;
??????????? notEmpty.signalAll();
??????? } finally {
??????????? lock.unlock();
??????? }
??? }
??? public T take() throws InterruptedException {
??????? lock.lock();
??????? try {
??????????? while (count == 0) {
??????????????? notEmpty.await();
??????????? }
??????????? T ret = items[head];
??????????? items[head] = null;//GC
??????????? //
??????????? if (++head == getCapacity()) {
??????????????? head = 0;
??????????? }
??????????? --count;
??????????? notFull.signalAll();
??????????? return ret;
??????? } finally {
??????????? lock.unlock();
??????? }
??? }
??? public int getCapacity() {
??????? return items.length;
??? }
??? public int size() {
??????? lock.lock();
??????? try {
??????????? return count;
??????? } finally {
??????????? lock.unlock();
??????? }
??? }
}
在這個例子中消費(fèi)take()需要 隊列不為空，如果為空就掛起（await()），直到收到notEmpty的信號；生產(chǎn)put()需要隊列不滿，如果滿了就掛起（await()），直到收到notFull的信號。
可能有人會問題，如果一個線程lock()對象后被掛起還沒有unlock，那么另外一個線程就拿不到鎖了（lock()操作會掛起），那么就無法通知(notify)前一個線程，這樣豈不是“死鎖”了？
?
await* 操作
上一節(jié)中說過多次ReentrantLock是獨(dú)占鎖，一個線程拿到鎖后如果不釋放，那么另外一個線程肯定是拿不到鎖，所以在lock.lock()和lock.unlock()之間可能有一次釋放鎖的操作（同樣也必然還有一次獲取鎖的操作）。我們再回頭看代碼，不管take()還是put()，在進(jìn)入lock.lock()后唯一可能釋放鎖的操作就是await()了。也就是說await()操作實(shí)際上就是釋放鎖，然后掛起線程，一旦條件滿足就被喚醒，再次獲取鎖！
public final void await() throws InterruptedException {
??? if (Thread.interrupted())
??????? throw new InterruptedException();
??? Node node = addConditionWaiter();
??? int savedState = fullyRelease(node);
??? int interruptMode = 0;
??? while (!isOnSyncQueue(node)) {
??????? LockSupport.park(this);
??????? if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
??????????? break;
??? }
??? if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
??????? interruptMode = REINTERRUPT;
??? if (node.nextWaiter != null)
??????? unlinkCancelledWaiters();
??? if (interruptMode != 0)
??????? reportInterruptAfterWait(interruptMode);
}
上面是await()的代碼片段。上一節(jié)中說過，AQS在獲取鎖的時候需要有一個CHL的FIFO隊列，所以對于一個Condition.await()而言，如果釋放了鎖，要想再一次獲取鎖那么就需要進(jìn)入隊列，等待被通知獲取鎖。完整的await()操作是安裝如下步驟進(jìn)行的：
這里再回頭介紹Condition的數(shù)據(jù)結(jié)構(gòu)。我們知道一個Condition可以在多個地方被await*()，那么就需要一個FIFO的結(jié)構(gòu)將這些Condition串聯(lián)起來，然后根據(jù)需要喚醒一個或者多個（通常是所有）。所以在Condition內(nèi)部就需要一個FIFO的隊列。
private transient Node firstWaiter;
private transient Node lastWaiter;
上面的兩個節(jié)點(diǎn)就是描述一個FIFO的隊列。我們再結(jié)合前面提到的節(jié)點(diǎn)（Node）數(shù)據(jù)結(jié)構(gòu)。我們就發(fā)現(xiàn)Node.nextWaiter就派上用場了！nextWaiter就是將一系列的Condition.await*串聯(lián)起來組成一個FIFO的隊列。
?
signal/signalAll 操作
await*()清楚了，現(xiàn)在再來看signal/signalAll就容易多了。按照signal/signalAll的需求，就是要將Condition.await*()中FIFO隊列中第一個Node喚醒（或者全部Node）喚醒。盡管所有Node可能都被喚醒，但是要知道的是仍然只有一個線程能夠拿到鎖，其它沒有拿到鎖的線程仍然需要自旋等待，就上上面提到的第4步(acquireQueued)。
private void doSignal(Node first) {
??? do {
??????? if ( (firstWaiter = first.nextWaiter) == null)
??????????? lastWaiter = null;
??????? first.nextWaiter = null;
??? } while (!transferForSignal(first) &&
???????????? (first = firstWaiter) != null);
}
private void doSignalAll(Node first) {
??? lastWaiter = firstWaiter? = null;
??? do {
??????? Node next = first.nextWaiter;
??????? first.nextWaiter = null;
??????? transferForSignal(first);
??????? first = next;
??? } while (first != null);
}
上面的代碼很容易看出來，signal就是喚醒Condition隊列中的第一個非CANCELLED節(jié)點(diǎn)線程，而signalAll就是喚醒所有非CANCELLED節(jié)點(diǎn)線程。當(dāng)然了遇到CANCELLED線程就需要將其從FIFO隊列中剔除。
final boolean transferForSignal(Node node) {
??? if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
??????? return false;
??? Node p = enq(node);
??? int c = p.waitStatus;
??? if (c > 0 || !compareAndSetWaitStatus(p, c, Node.SIGNAL))
??????? LockSupport.unpark(node.thread);
??? return true;
}
上面就是喚醒一個await*()線程的過程，根據(jù)前面的小節(jié)介紹的，如果要unpark線程，并使線程拿到鎖，那么就需要線程節(jié)點(diǎn)進(jìn)入AQS的隊列。所以可以看到在LockSupport.unpark之前調(diào)用了enq(node)操作，將當(dāng)前節(jié)點(diǎn)加入到AQS隊列。
整個鎖機(jī)制的原理就介紹完了，從下一節(jié)開始就進(jìn)入了鎖機(jī)制的應(yīng)用了。
五、閉鎖
閉鎖（Latch）：一種同步方法，可以延遲線程的進(jìn)度直到線程到達(dá)某個終點(diǎn)狀態(tài)。通俗的講就是，一個閉鎖相當(dāng)于一扇大門，在大門打開之前所有線程都被阻斷，一旦大門打開所有線程都將通過，但是一旦大門打開，所有線程都通過了，那么這個閉鎖的狀態(tài)就失效了，門的狀態(tài)也就不能變了，只能是打開狀態(tài)。也就是說閉鎖的狀態(tài)是一次性的，它確保在閉鎖打開之前所有特定的活動都需要在閉鎖打開之后才能完成。
CountDownLatch是JDK 5+里面閉鎖的一個實(shí)現(xiàn)，允許一個或者多個線程等待某個事件的發(fā)生。CountDownLatch有一個正數(shù)計數(shù)器，countDown方法對計數(shù)器做減操作，await方法等待計數(shù)器達(dá)到0。所有await的線程都會阻塞直到計數(shù)器為0或者等待線程中斷或者超時。
CountDownLatch的API如下。
- public void await() throws InterruptedException
- public boolean await(long timeout, TimeUnit unit) throws InterruptedException
- public void countDown()
- public long getCount()
其中g(shù)etCount()描述的是當(dāng)前計數(shù)，通常用于調(diào)試目的。
下面的例子中描述了閉鎖的兩種常見的用法。
package xylz.study.concurrency.lock;
import java.util.concurrent.CountDownLatch;
public class PerformanceTestTool {
??? public long timecost(final int times, final Runnable task) throws InterruptedException {
??????? if (times <= 0) throw new IllegalArgumentException();
??????? final CountDownLatch startLatch = new CountDownLatch(1);
??????? final CountDownLatch overLatch = new CountDownLatch(times);
??????? for (int i = 0; i < times; i++) {
??????????? new Thread(new Runnable() {
??????????????? public void run() {
??????????????????? try {
??????????????????????? startLatch.await();
??????????????????????? //
??????????????????????? task.run();
??????????????????? } catch (InterruptedException ex) {
??????????????????????? Thread.currentThread().interrupt();
??????????????????? } finally {
??????????????????????? overLatch.countDown();
??????????????????? }
??????????????? }
??????????? }).start();
??????? }
??????? //
??????? long start = System.nanoTime();
// 準(zhǔn)備工作完成，調(diào)用下面的方法，打開閉鎖
??????? startLatch.countDown();
// 主線程在此等候，所有其他線程完成后，主線程開始工作
??????? overLatch.await();
??????? return System.nanoTime() - start;
??? }
}
在上面的例子中使用了兩個閉鎖，第一個閉鎖確保在所有線程開始執(zhí)行任務(wù)前，所有準(zhǔn)備工作都已經(jīng)完成，一旦準(zhǔn)備工作完成了就調(diào)用startLatch.countDown()打開閉鎖，所有線程開始執(zhí)行。第二個閉鎖在于確保所有任務(wù)執(zhí)行完成后主線程才能繼續(xù)進(jìn)行，這樣保證了主線程等待所有任務(wù)線程執(zhí)行完成后才能得到需要的結(jié)果。在第二個閉鎖當(dāng)中，初始化了一個N次的計數(shù)器，每個任務(wù)執(zhí)行完成后都會將計數(shù)器減一，所有任務(wù)完成后計數(shù)器就變?yōu)榱?，這樣主線程閉鎖overLatch拿到此信號后就可以繼續(xù)往下執(zhí)行了。
根據(jù)前面的happend-before法則可以知道閉鎖有以下特性：
內(nèi)存一致性效果：線程中調(diào)用?countDown()?之前的操作?happen-before?緊跟在從另一個線程中對應(yīng)?await()?成功返回的操作。
在上面的例子中第二個閉鎖相當(dāng)于把一個任務(wù)拆分成N份，每一份獨(dú)立完成任務(wù)，主線程等待所有任務(wù)完成后才能繼續(xù)執(zhí)行。這個特性在后面的線程池框架中會用到，其實(shí)FutureTask就可以看成一個閉鎖。后面的章節(jié)還會具體分析FutureTask的。
?
同樣基于探索精神，仍然需要“窺探”下CountDownLatch里面到底是如何實(shí)現(xiàn)await*和countDown的。
首先，研究下await()方法。內(nèi)部直接調(diào)用了AQS的acquireSharedInterruptibly(1)。
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
??? if (Thread.interrupted())
??????? throw new InterruptedException();
??? if (tryAcquireShared(arg) < 0)
??????? doAcquireSharedInterruptibly(arg);
}
前面一直提到的都是獨(dú)占鎖（排它鎖、互斥鎖），現(xiàn)在就用到了另外一種鎖，共享鎖。
所謂共享鎖是說所有共享鎖的線程共享同一個資源，一旦任意一個線程拿到共享資源，那么所有線程就都擁有的同一份資源。也就是通常情況下共享鎖只是一個標(biāo)志，所有線程都等待這個標(biāo)識是否滿足，一旦滿足所有線程都被激活（相當(dāng)于所有線程都拿到鎖一樣）。這里的閉鎖CountDownLatch就是基于共享鎖的實(shí)現(xiàn)。
閉鎖中關(guān)于AQS的tryAcquireShared的實(shí)現(xiàn)是如下代碼（java.util.concurrent.CountDownLatch.Sync.tryAcquireShared）：
public int tryAcquireShared(int acquires) {
??? return getState() == 0? 1 : -1;
}
在這份邏輯中，對于閉鎖而言第一次await時tryAcquireShared應(yīng)該總是-1，因為對于閉鎖CountDownLatch而言state的值就是初始化的count值。這也就解釋了為什么在countDown調(diào)用之前閉鎖的count總是>0。
private void doAcquireSharedInterruptibly(int arg)
??? throws InterruptedException {
??? final Node node = addWaiter(Node.SHARED);
??? try {
??????? for (;;) {
??????????? final Node p = node.predecessor();
??????????? if (p == head) {
??????????????? int r = tryAcquireShared(arg);
??????????????? if (r >= 0) {
??????????????????? setHeadAndPropagate(node, r);
??????????????????? p.next = null; // help GC
??????????????????? return;
??????????????? }
??????????? }
??????????? if (shouldParkAfterFailedAcquire(p, node) &&
??????????????? parkAndCheckInterrupt())
??????????????? break;
??????? }
??? } catch (RuntimeException ex) {
??????? cancelAcquire(node);
??????? throw ex;
??? }
??? // Arrive here only if interrupted
??? cancelAcquire(node);
??? throw new InterruptedException();
}
上面的邏輯展示了如何通過await將所有線程串聯(lián)并掛起，直到被喚醒或者條件滿足或者被中斷。整個過程是這樣的：
這里有一點(diǎn)值得說明下，設(shè)置頭結(jié)點(diǎn)并喚醒繼任節(jié)點(diǎn)setHeadAndPropagate。由于前面tryAcquireShared總是返回1或者-1，而進(jìn)入setHeadAndPropagate時總是propagate>=0，所以這里propagate==1。后面喚醒繼任節(jié)點(diǎn)操作就非常熟悉了。
private void setHeadAndPropagate(Node node, int propagate) {
??? setHead(node);
??? if (propagate > 0 && node.waitStatus != 0) {
??????? Node s = node.next;
??????? if (s == null || s.isShared())
??????????? unparkSuccessor(node);
??? }
}
從上面的所有邏輯可以看出countDown應(yīng)該就是在條件滿足（計數(shù)為0）時喚醒頭結(jié)點(diǎn)（時間最長的一個節(jié)點(diǎn)），然后頭結(jié)點(diǎn)就會根據(jù)FIFO隊列喚醒整個節(jié)點(diǎn)列表（如果有的話）。
從CountDownLatch的countDown代碼中看到，直接調(diào)用的是AQS的releaseShared(1)，參考前面的知識，這就印證了上面的說法。
tryReleaseShared中正是采用CAS操作減少計數(shù)（每次減-1）。
public boolean tryReleaseShared(int releases) {
??? for (;;) {
??????? int c = getState();
??????? if (c == 0)
??????????? return false;
??????? int nextc = c-1;
??????? if (compareAndSetState(c, nextc))
??????????? return nextc == 0;
??? }
}
整個CountDownLatch就是這個樣子的。其實(shí)有了前面原子操作和AQS的原理及實(shí)現(xiàn)，分析CountDownLatch還是比較容易的。
?
六、CyclicBarrier
如果說CountDownLatch是一次性的，那么CyclicBarrier正好可以循環(huán)使用。它允許一組線程互相等待，直到到達(dá)某個公共屏障點(diǎn) (common barrier point)。所謂屏障點(diǎn)就是一組任務(wù)執(zhí)行完畢的時刻。
?
清單1 一個使用CyclicBarrier的例子
package xylz.study.concurrency.lock;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierDemo {
??? final CyclicBarrier barrier;
??? final int MAX_TASK;
??? public CyclicBarrierDemo(int cnt) {
??????? barrier = new CyclicBarrier(cnt + 1);
??????? MAX_TASK = cnt;
??? }
??? public void doWork(final Runnable work) {
??????? new Thread() {
??????????? public void run() {
??????????????? work.run();
??????????????? try {
??????????????????? int index = barrier.await();
??????????????????? doWithIndex(index);
??????????????? } catch (InterruptedException e) {
??????????????????? return;
??????????????? } catch (BrokenBarrierException e) {
??????????????????? return;
??????????????? }
??????????? }
??????? }.start();
??? }
??? private void doWithIndex(int index) {
??????? if (index == MAX_TASK / 3) {
??????????? System.out.println("Left 30%.");
??????? } else if (index == MAX_TASK / 2) {
??????????? System.out.println("Left 50%");
??????? } else if (index == 0) {
??????????? System.out.println("run over");
??????? }
??? }
??? public void waitForNext() {
??????? try {
??????????? doWithIndex(barrier.await());
??????? } catch (InterruptedException e) {
??????????? return;
??????? } catch (BrokenBarrierException e) {
??????????? return;
??????? }
??? }
??? public static void main(String[] args) {
??????? final int count = 10;
??????? CyclicBarrierDemo demo = new CyclicBarrierDemo(count);
??????? for (int i = 0; i < 100; i++) {
??????????? demo.doWork(new Runnable() {
??????????????? public void run() {
??????????????????? //do something
??????????????????? try {
??????????????????????? Thread.sleep(1000L);
??????????????????? } catch (Exception e) {
??????????????????????? return;
??????????????????? }
??????????????? }
??????????? });
??????????? if ((i + 1) % count == 0) {
??????????????? demo.waitForNext();
??????????? }
??????? }
??? }
}
清單1描述的是一個周期性處理任務(wù)的例子，在這個例子中有一對的任務(wù)（100個），希望每10個為一組進(jìn)行處理，當(dāng)前僅當(dāng)上一組任務(wù)處理完成后才能進(jìn)行下一組，另外在每一組任務(wù)中，當(dāng)任務(wù)剩下50%，30%以及所有任務(wù)執(zhí)行完成時向觀察者發(fā)出通知。
在這個例子中，CyclicBarrierDemo 構(gòu)建了一個count+1的任務(wù)組（其中一個任務(wù)時為了外界方便掛起主線程）。每一個子任務(wù)里，人物本身執(zhí)行完畢后都需要等待同組內(nèi)其它任務(wù)執(zhí)行完成后才能繼續(xù)。同時在剩下任務(wù)50%、30%已經(jīng)0時執(zhí)行特殊的其他任務(wù)（發(fā)通知）。
很顯然CyclicBarrier有以下幾個特點(diǎn)：
- await()方法將掛起線程，直到同組的其它線程執(zhí)行完畢才能繼續(xù)
- await()方法返回線程執(zhí)行完畢的索引，注意，索引時從任務(wù)數(shù)-1開始的，也就是第一個執(zhí)行完成的任務(wù)索引為parties-1,最后一個為0，這個parties為總?cè)蝿?wù)數(shù)，清單中是cnt+1
- CyclicBarrier 是可循環(huán)的，顯然名稱說明了這點(diǎn)。在清單1中，每一組任務(wù)執(zhí)行完畢就能夠執(zhí)行下一組任務(wù)。
另外除了CyclicBarrier除了以上特點(diǎn)外，還有以下幾個特點(diǎn)：
- 如果屏障操作不依賴于掛起的線程，那么任何線程都可以執(zhí)行屏障操作。在清單1中可以看到并沒有指定那個線程執(zhí)行50%、30%、0%的操作，而是一組線程（cnt+1）個中任何一個線程只要到達(dá)了屏障點(diǎn)都可以執(zhí)行相應(yīng)的操作
- CyclicBarrier 的構(gòu)造函數(shù)允許攜帶一個任務(wù)，這個任務(wù)將在0%屏障點(diǎn)執(zhí)行，它將在await()==0后執(zhí)行。
- CyclicBarrier 如果在await時因為中斷、失敗、超時等原因提前離開了屏障點(diǎn)，那么任務(wù)組中的其他任務(wù)將立即被中斷，以InterruptedException異常離開線程。
- 所有await()之前的操作都將在屏障點(diǎn)之前運(yùn)行，也就是CyclicBarrier 的內(nèi)存一致性效果
?
CyclicBarrier 的所有API如下：
- public CyclicBarrier(int parties)?創(chuàng)建一個新的?CyclicBarrier，它將在給定數(shù)量的參與者（線程）處于等待狀態(tài)時啟動，但它不會在啟動 barrier 時執(zhí)行預(yù)定義的操作。
- public CyclicBarrier(int parties, Runnable barrierAction)?創(chuàng)建一個新的?CyclicBarrier，它將在給定數(shù)量的參與者（線程）處于等待狀態(tài)時啟動，并在啟動 barrier 時執(zhí)行給定的屏障操作，該操作由最后一個進(jìn)入 barrier 的線程執(zhí)行。
- public int await() throws InterruptedException, BrokenBarrierException?在所有參與者都已經(jīng)在此 barrier 上調(diào)用?await?方法之前，將一直等待。
- public int await(long timeout,TimeUnit unit) throws InterruptedException, BrokenBarrierException,TimeoutException?在所有參與者都已經(jīng)在此屏障上調(diào)用 await 方法之前將一直等待,或者超出了指定的等待時間。
- public int getNumberWaiting()?返回當(dāng)前在屏障處等待的參與者數(shù)目。此方法主要用于調(diào)試和斷言。
- public int getParties()?返回要求啟動此 barrier 的參與者數(shù)目。
- public boolean isBroken()?查詢此屏障是否處于損壞狀態(tài)。
- public void reset()?將屏障重置為其初始狀態(tài)。
針對以上API，下面來探討下CyclicBarrier 的實(shí)現(xiàn)原理，以及為什么有這樣的API。
清單2 CyclicBarrier.await*()的實(shí)現(xiàn)片段
??? private int dowait(boolean timed, long nanos)
??? throws InterruptedException, BrokenBarrierException,
?????????? TimeoutException {
??? final ReentrantLock lock = this.lock;
??? lock.lock();
??? try {
??????? final Generation g = generation;
??????? if (g.broken)
??????????? throw new BrokenBarrierException();
??????? if (Thread.interrupted()) {
??????????? breakBarrier();
??????????? throw new InterruptedException();
??????? }
?????? int index = --count;
?????? if (index == 0) {? // tripped
?????????? boolean ranAction = false;
?????????? try {
?????????????? final Runnable command = barrierCommand;
?????????????? if (command != null)
?????????????????? command.run();
?????????????? ranAction = true;
?????????????? nextGeneration();
?????????????? return 0;
?????????? } finally {
?????????????? if (!ranAction)
?????????????????? breakBarrier();
?????????? }
?????? }
??????? // loop until tripped, broken, interrupted, or timed out
??????? for (;;) {
??????????? try {
??????????????? if (!timed)
??????????????????? trip.await();
??????????????? else if (nanos > 0L)
??????????????????? nanos = trip.awaitNanos(nanos);
??????????? } catch (InterruptedException ie) {
??????????????? if (g == generation && ! g.broken) {
??????????????????? breakBarrier();
??????????????????? throw ie;
??????????????? } else {
??????????????????? Thread.currentThread().interrupt();
??????????????? }
??????????? }
??????????? if (g.broken)
??????????????? throw new BrokenBarrierException();
??????????? if (g != generation)
??????????????? return index;
??????????? if (timed && nanos <= 0L) {
??????????????? breakBarrier();
??????????????? throw new TimeoutException();
??????????? }
??????? }
??? } finally {
??????? lock.unlock();
??? }
}
清單2有點(diǎn)復(fù)雜，這里一點(diǎn)一點(diǎn)的剖析，并且還原到最原始的狀態(tài)。
利用前面學(xué)到的知識，我們知道要想讓線程等待其他線程執(zhí)行完畢，那么已經(jīng)執(zhí)行完畢的線程（進(jìn)入await*()方法）就需要park()，直到超時或者被中斷，或者被其它線程喚醒。
前面說過CyclicBarrier 的特點(diǎn)是要么大家都正常執(zhí)行完畢，要么大家都異常被中斷，不會其中有一個被中斷而其它正常執(zhí)行完畢的現(xiàn)象存在。這種特點(diǎn)叫all-or-none。類似的概念是原子操作中的要么大家都執(zhí)行完，要么一個操作都不執(zhí)行完。當(dāng)前這其實(shí)是兩個概念了。要完成這樣的特點(diǎn)就必須有一個狀態(tài)來描述曾經(jīng)是否有過線程被中斷（broken)了，這樣后面執(zhí)行完的線程就該知道是否需要繼續(xù)等待了。而在CyclicBarrier 中Generation 就是為了完成這件事情的。Generation的定義非常簡單，整個結(jié)構(gòu)就只有一個變量boolean broken = false;，定義是否發(fā)生了broken操作。
由于有競爭資源的存在（broken/index），所以毫無疑問需要一把鎖lock。拿到鎖后整個過程是這樣的：
上面4個步驟其實(shí)只是描述主體結(jié)構(gòu)，事實(shí)上整個過程中有非常多的邏輯來處理異常引發(fā)的問題，比如執(zhí)行屏障點(diǎn)任務(wù)引發(fā)的異常，park線程超時引發(fā)的中斷異常和超時異常等等。所以對于await()而言，異常的處理比業(yè)務(wù)邏輯的處理更復(fù)雜，這就解釋了為什么await()的時候可能引發(fā)InterruptedException,BrokenBarrierException,TimeoutException?三種異常。
清單3 生成下一個循環(huán)周期并喚醒其它線程
private void nextGeneration() {
???? trip.signalAll();
???? count = parties;
???? generation = new Generation();
}
清單3 描述了如何生成下一個循環(huán)周期的過程，在這個過程中當(dāng)然需要使用Condition.signalAll()喚醒所有已經(jīng)執(zhí)行完成并且正在等待的線程。另外這里count描述的是還有多少線程需要執(zhí)行，是為了線程執(zhí)行完畢索引計數(shù)。
isBroken() 方法描述的就是generation.broken，也即線程組是否發(fā)生了異常。這里再一次解釋下為什么要有這個狀態(tài)的存在。
如果一個將要位于屏障點(diǎn)或者已經(jīng)位于屏障點(diǎn)的而執(zhí)行屏障點(diǎn)任務(wù)的線程發(fā)生了異常，那么即使喚醒了其它等待的線程，其它等待的線程也會因為循環(huán)等待而“死去”，因為再也沒有一個線程來喚醒這些第二次進(jìn)行park的線程了。還有一個意圖是，如果屏障點(diǎn)都已經(jīng)損壞了，那么其它將要等待屏障點(diǎn)的再線程掛起就沒有意義了。
寫到這里的時候非常不幸，用了4年多了臺燈終于“壽終正寢了”。
其實(shí)CyclicBarrier 還有一個reset方法，描述的是手動立即將所有線程中斷，恢復(fù)屏障點(diǎn)，進(jìn)行下一組任務(wù)的執(zhí)行。也就是與重新創(chuàng)建一個新的屏障點(diǎn)相比，可能維護(hù)的代價要小一些（減少同步，減少上一個CyclicBarrier 的管理等等）。
本來是想和Semaphore 一起將的，最后發(fā)現(xiàn)鋪開后就有點(diǎn)長了，而且也不利于理解和吸收，所以放到下一篇吧。
七、信號量（Semaphore）
Semaphore 是一個計數(shù)信號量。從概念上講，信號量維護(hù)了一個許可集。如有必要，在許可可用前會阻塞每一個?acquire()，然后再獲取該許可。每個?release()?添加一個許可，從而可能釋放一個正在阻塞的獲取者。但是，不使用實(shí)際的許可對象，Semaphore?只對可用許可的號碼進(jìn)行計數(shù)，并采取相應(yīng)的行動。
說白了，Semaphore是一個計數(shù)器，在計數(shù)器不為0的時候?qū)€程就放行，一旦達(dá)到0，那么所有請求資源的新線程都會被阻塞，包括增加請求到許可的線程，也就是說Semaphore不是可重入的。每一次請求一個許可都會導(dǎo)致計數(shù)器減少1，同樣每次釋放一個許可都會導(dǎo)致計數(shù)器增加1，一旦達(dá)到了0，新的許可請求線程將被掛起。
緩存池整好使用此思想來實(shí)現(xiàn)的，比如鏈接池、對象池等。
清單1 對象池
package xylz.study.concurrency.lock;
import java.util.concurrent.Semaphore;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class ObjectCache<T> {
??? public interface ObjectFactory<T> {
??????? T makeObject();
??? }
??? class Node {
??????? T obj;
??????? Node next;
??? }
??? final int capacity;
??? final ObjectFactory<T> factory;
??? final Lock lock = new ReentrantLock();
??? final Semaphore semaphore;
??? private Node head;
??? private Node tail;
??? public ObjectCache(int capacity, ObjectFactory<T> factory) {
??????? this.capacity = capacity;
??????? this.factory = factory;
??????? this.semaphore = new Semaphore(this.capacity);
??????? this.head = null;
??????? this.tail = null;
??? }
??? public T getObject() throws InterruptedException {
//獲取一個信號量
??????? semaphore.acquire();
??????? return getNextObject();
??? }
??? private T getNextObject() {
??????? lock.lock();
??????? try {
??????????? if (head == null) {
??????????????? return factory.makeObject();
??????????? } else {
??????????????? Node ret = head;
??????????????? head = head.next;
??????????????? if (head == null) tail = null;
??????????????? ret.next = null;//help GC
??????????????? return ret.obj;
??????????? }
??????? } finally {
??????????? lock.unlock();
??????? }
??? }
??? private void returnObjectToPool(T t) {
??????? lock.lock();
??????? try {
??????????? Node node = new Node();
??????????? node.obj = t;
??????????? if (tail == null) {
??????????????? head = tail = node;
??????????? } else {
??????????????? tail.next = node;
??????????????? tail = node;
??????????? }
??????? } finally {
??????????? lock.unlock();
??????? }
??? }
??? public void returnObject(T t) {
??????? returnObjectToPool(t);
//返回一個信號量
??????? semaphore.release();
??? }
}
清單1描述了一個基于信號量Semaphore的對象池實(shí)現(xiàn)。此對象池最多支持capacity個對象，這在構(gòu)造函數(shù)中傳入。對象池有一個基于FIFO的隊列，每次從對象池的頭結(jié)點(diǎn)開始取對象，如果頭結(jié)點(diǎn)為空就直接構(gòu)造一個新的對象返回。否則將頭結(jié)點(diǎn)對象取出，并且頭結(jié)點(diǎn)往后移動。特別要說明的如果對象的個數(shù)用完了，那么新的線程將被阻塞，直到有對象被返回回來。返還對象時將對象加入FIFO的尾節(jié)點(diǎn)并且釋放一個空閑的信號量，表示對象池中增加一個可用對象。
實(shí)際上對象池、線程池的原理大致上就是這樣的，只不過真正的對象池、線程池要處理比較復(fù)雜的邏輯，所以實(shí)現(xiàn)起來還需要做很多的工作，例如超時機(jī)制，自動回收機(jī)制，對象的有效期等等問題。
這里特別說明的是信號量只是在信號不夠的時候掛起線程，但是并不能保證信號量足夠的時候獲取對象和返還對象是線程安全的，所以在清單1中仍然需要鎖Lock來保證并發(fā)的正確性。
將信號量初始化為 1，使得它在使用時最多只有一個可用的許可，從而可用作一個相互排斥的鎖。這通常也稱為二進(jìn)制信號量，因為它只能有兩種狀態(tài)：一個可用的許可，或零個可用的許可。按此方式使用時，二進(jìn)制信號量具有某種屬性（與很多?Lock?實(shí)現(xiàn)不同），即可以由線程釋放“鎖”，而不是由所有者（因為信號量沒有所有權(quán)的概念）。在某些專門的上下文（如死鎖恢復(fù)）中這會很有用。
上面這段話的意思是說當(dāng)某個線程A持有信號量數(shù)為1的信號量時，其它線程只能等待此線程釋放資源才能繼續(xù)，這時候持有信號量的線程A就相當(dāng)于持有了“鎖”，其它線程的繼續(xù)就需要這把鎖，于是線程A的釋放才能決定其它線程的運(yùn)行，相當(dāng)于扮演了“鎖”的角色。
?
另外同公平鎖非公平鎖一樣，信號量也有公平性。如果一個信號量是公平的表示線程在獲取信號量時按FIFO的順序得到許可，也就是按照請求的順序得到釋放。這里特別說明的是：所謂請求的順序是指在請求信號量而進(jìn)入FIFO隊列的順序，有可能某個線程先請求信號而后進(jìn)去請求隊列，那么次線程獲取信號量的順序就會晚于其后請求但是先進(jìn)入請求隊列的線程。這個在公平鎖和非公平鎖中談過很多。
?
除了acquire以外，Semaphore還有幾種類似的acquire方法，這些方法可以更好的處理中斷和超時或者異步等特性，可以參考JDK API。
按照同樣的學(xué)習(xí)原則，下面對主要的實(shí)現(xiàn)進(jìn)行分析。Semaphore的acquire方法實(shí)際上訪問的是AQS的acquireSharedInterruptibly(arg)方法。這個可以參考CountDownLatch一節(jié)或者AQS一節(jié)。
所以Semaphore的await實(shí)現(xiàn)也是比較簡單的。與CountDownLatch不同的是，Semaphore區(qū)分公平信號和非公平信號。
清單2 公平信號獲取方法
protected int tryAcquireShared(int acquires) {
??? Thread current = Thread.currentThread();
??? for (;;) {
??????? Thread first = getFirstQueuedThread();
??????? if (first != null && first != current)
??????????? return -1;
??????? int available = getState();
??????? int remaining = available - acquires;
??????? if (remaining < 0 ||
??????????? compareAndSetState(available, remaining))
??????????? return remaining;
??? }
}
清單3 非公平信號獲取方法
protected int tryAcquireShared(int acquires) {
??? return nonfairTryAcquireShared(acquires);
}
final int nonfairTryAcquireShared(int acquires) {
??? for (;;) {
??????? int available = getState();
??????? int remaining = available - acquires;
??????? if (remaining < 0 ||
??????????? compareAndSetState(available, remaining))
??????????? return remaining;
??? }
}
對比清單2和清單3可以看到，公平信號和非公平信號在于第一次嘗試能否獲取信號時，公平信號量總是將當(dāng)前線程進(jìn)入AQS的CLH隊列進(jìn)行排隊（因為第一次嘗試時隊列的頭結(jié)點(diǎn)線程很有可能不是當(dāng)前線程，當(dāng)然不排除同一個線程第二次進(jìn)入信號量），從而根據(jù)AQS的CLH隊列的順序FIFO依次獲取信號量；而對于非公平信號量，第一次立即嘗試能否拿到信號量，一旦信號量的剩余數(shù)available大于請求數(shù)（acquires通常為1），那么線程就立即得到了釋放，而不需要進(jìn)行AQS隊列進(jìn)行排隊。只有remaining<0的時候（也就是信號量不夠的時候）才會進(jìn)入AQS隊列。
所以非公平信號量的吞吐量總是要比公平信號量的吞吐量要大，但是需要強(qiáng)調(diào)的是非公平信號量和非公平鎖一樣存在“饑渴死”的現(xiàn)象，也就是說活躍線程可能總是拿到信號量，而非活躍線程可能難以拿到信號量。而對于公平信號量由于總是靠請求的線程的順序來獲取信號量，所以不存在此問題。
八、讀寫鎖
從這一節(jié)開始介紹鎖里面的最后一個工具：讀寫鎖(ReadWriteLock)。
ReentrantLock 實(shí)現(xiàn)了標(biāo)準(zhǔn)的互斥操作，也就是一次只能有一個線程持有鎖，也即所謂獨(dú)占鎖的概念。前面的章節(jié)中一直在強(qiáng)調(diào)這個特點(diǎn)。顯然這個特點(diǎn)在一定程度上面減低了吞吐量，實(shí)際上獨(dú)占鎖是一種保守的鎖策略，在這種情況下任何“讀/讀”，“寫/讀”，“寫/寫”操作都不能同時發(fā)生。但是同樣需要強(qiáng)調(diào)的一個概念是，鎖是有一定的開銷的，當(dāng)并發(fā)比較大的時候，鎖的開銷就比較客觀了。所以如果可能的話就盡量少用鎖，非要用鎖的話就嘗試看能否改造為讀寫鎖。
ReadWriteLock描述的是：一個資源能夠被多個讀線程訪問，或者被一個寫線程訪問，但是不能同時存在讀寫線程。也就是說讀寫鎖使用的場合是一個共享資源被大量讀取操作，而只有少量的寫操作（修改數(shù)據(jù)）。清單1描述了ReadWriteLock的API。
?清單1 ReadWriteLock 接口
public interface ReadWriteLock {
??? Lock readLock();
??? Lock writeLock();
}
清單1描述的ReadWriteLock結(jié)構(gòu)，這里需要說明的是ReadWriteLock并不是Lock的子接口，只不過ReadWriteLock借助Lock來實(shí)現(xiàn)讀寫兩個視角。在ReadWriteLock中每次讀取共享數(shù)據(jù)就需要讀取鎖，當(dāng)需要修改共享數(shù)據(jù)時就需要寫入鎖。看起來好像是兩個鎖，但其實(shí)不盡然，在下一節(jié)中的分析中會解釋這點(diǎn)奧秘。
在JDK 6里面ReadWriteLock的實(shí)現(xiàn)是ReentrantReadWriteLock。
清單2 SimpleConcurrentMap
package xylz.study.concurrency.lock;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
public class SimpleConcurrentMap<K, V> implements Map<K, V> {
??? final ReadWriteLock lock = new ReentrantReadWriteLock();
??? final Lock r = lock.readLock();
??? final Lock w = lock.writeLock();
??? final Map<K, V> map;
??? public SimpleConcurrentMap(Map<K, V> map) {
??????? this.map = map;
??????? if (map == null) throw new NullPointerException();
??? }
??? public void clear() {
??????? w.lock();
??????? try {
??????????? map.clear();
??????? } finally {
??????????? w.unlock();
??????? }
??? }
??? public boolean containsKey(Object key) {
??????? r.lock();
??????? try {
??????????? return map.containsKey(key);
??????? } finally {
??????????? r.unlock();
??????? }
??? }
??? public boolean containsValue(Object value) {
??????? r.lock();
??????? try {
??????????? return map.containsValue(value);
??????? } finally {
??????????? r.unlock();
??????? }
??? }
??? public Set<java.util.Map.Entry<K, V>> entrySet() {
??????? throw new UnsupportedOperationException();
??? }
??? public V get(Object key) {
??????? r.lock();
??????? try {
??????????? return map.get(key);
??????? } finally {
??????????? r.unlock();
??????? }
??? }
??? public boolean isEmpty() {
??????? r.lock();
??????? try {
??????????? return map.isEmpty();
??????? } finally {
??????????? r.unlock();
??????? }
??? }
??? public Set<K> keySet() {
??????? r.lock();
??????? try {
??????????? return new HashSet<K>(map.keySet());
??????? } finally {
??????????? r.unlock();
??????? }
??? }
??? public V put(K key, V value) {
??????? w.lock();
??????? try {
??????????? return map.put(key, value);
??????? } finally {
??????????? w.unlock();
??????? }
??? }
??? public void putAll(Map<? extends K, ? extends V> m) {
??????? w.lock();
??????? try {
??????????? map.putAll(m);
??????? } finally {
??????????? w.unlock();
??????? }
??? }
??? public V remove(Object key) {
??????? w.lock();
??????? try {
??????????? return map.remove(key);
??????? } finally {
??????????? w.unlock();
??????? }
??? }
??? public int size() {
??????? r.lock();
??????? try {
??????????? return map.size();
??????? } finally {
??????????? r.unlock();
??????? }
??? }
??? public Collection<V> values() {
??????? r.lock();
??????? try {
??????????? return new ArrayList<V>(map.values());
??????? } finally {
??????????? r.unlock();
??????? }
??? }
}
清單2描述的是用讀寫鎖實(shí)現(xiàn)的一個線程安全的Map。其中需要特別說明的是并沒有實(shí)現(xiàn)entrySet()方法，這是因為實(shí)現(xiàn)這個方法比較復(fù)雜，在后面章節(jié)中講到ConcurrentHashMap的時候會具體談這些細(xì)節(jié)。另外這里keySet()和values()也沒有直接返回Map的視圖，而是一個映射原有元素的新視圖，其實(shí)這個entrySet()一樣，是為了保護(hù)原始Map的數(shù)據(jù)邏輯，防止不正確的修改導(dǎo)致原始Map發(fā)生數(shù)據(jù)錯誤。特別說明的是在沒有特別需求的情況下沒有必要按照清單2寫一個線程安全的Map實(shí)現(xiàn)，因為ConcurrentHashMap已經(jīng)完成了此操作。
?
ReadWriteLock需要嚴(yán)格區(qū)分讀寫操作，如果讀操作使用了寫入鎖，那么降低讀操作的吞吐量，如果寫操作使用了讀取鎖，那么就可能發(fā)生數(shù)據(jù)錯誤。
另外ReentrantReadWriteLock還有以下幾個特性：
- 公平性
- 非公平鎖（默認(rèn)） 這個和獨(dú)占鎖的非公平性一樣，由于讀線程之間沒有鎖競爭，所以讀操作沒有公平性和非公平性，寫操作時，由于寫操作可能立即獲取到鎖，所以會推遲一個或多個讀操作或者寫操作。因此非公平鎖的吞吐量要高于公平鎖。
- 公平鎖 利用AQS的CLH隊列，釋放當(dāng)前保持的鎖（讀鎖或者寫鎖）時，優(yōu)先為等待時間最長的那個寫線程分配寫入鎖，當(dāng)前前提是寫線程的等待時間要比所有讀線程的等待時間要長。同樣一個線程持有寫入鎖或者有一個寫線程已經(jīng)在等待了，那么試圖獲取公平鎖的（非重入）所有線程（包括讀寫線程）都將被阻塞，直到最先的寫線程釋放鎖。如果讀線程的等待時間比寫線程的等待時間還有長，那么一旦上一個寫線程釋放鎖，這一組讀線程將獲取鎖。
- 重入性
- 讀寫鎖允許讀線程和寫線程按照請求鎖的順序重新獲取讀取鎖或者寫入鎖。當(dāng)然了只有寫線程釋放了鎖，讀線程才能獲取重入鎖。
- 寫線程獲取寫入鎖后可以再次獲取讀取鎖，但是讀線程獲取讀取鎖后卻不能獲取寫入鎖。
- 另外讀寫鎖最多支持65535個遞歸寫入鎖和65535個遞歸讀取鎖。
- 鎖降級
- 寫線程獲取寫入鎖后可以獲取讀取鎖，然后釋放寫入鎖，這樣就從寫入鎖變成了讀取鎖，從而實(shí)現(xiàn)鎖降級的特性。
- 鎖升級
- 讀取鎖是不能直接升級為寫入鎖的。因為獲取一個寫入鎖需要釋放所有讀取鎖，所以如果有兩個讀取鎖視圖獲取寫入鎖而都不釋放讀取鎖時就會發(fā)生死鎖。
- 鎖獲取中斷
- 讀取鎖和寫入鎖都支持獲取鎖期間被中斷。這個和獨(dú)占鎖一致。
- 條件變量
- 寫入鎖提供了條件變量(Condition)的支持，這個和獨(dú)占鎖一致，但是讀取鎖卻不允許獲取條件變量，將得到一個UnsupportedOperationException異常。
- 重入數(shù)
- 讀取鎖和寫入鎖的數(shù)量最大分別只能是65535（包括重入數(shù)）。這在下節(jié)中有介紹。
上面幾個特性對讀寫鎖的理解很有幫助，而且也是必要的，另外在下一節(jié)中講ReadWriteLock的實(shí)現(xiàn)會用到這些知識的。
九、讀寫鎖的實(shí)現(xiàn)
這一節(jié)主要是談?wù)勛x寫鎖的實(shí)現(xiàn)。
上一節(jié)中提到，ReadWriteLock看起來有兩個鎖：readLock/writeLock。如果真的是兩個鎖的話，它們之間又是如何相互影響的呢？
事實(shí)上在ReentrantReadWriteLock里鎖的實(shí)現(xiàn)是靠java.util.concurrent.locks.ReentrantReadWriteLock.Sync完成的。這個類看起來比較眼熟，實(shí)際上它是AQS的一個子類，這中類似的結(jié)構(gòu)在CountDownLatch、ReentrantLock、Semaphore里面都存在。同樣它也有兩種實(shí)現(xiàn)：公平鎖和非公平鎖，也就是java.util.concurrent.locks.ReentrantReadWriteLock.FairSync和java.util.concurrent.locks.ReentrantReadWriteLock.NonfairSync。這里暫且不提。
在ReentrantReadWriteLock里面的鎖主體就是一個Sync，也就是上面提到的FairSync或者NonfairSync，所以說實(shí)際上只有一個鎖，只是在獲取讀取鎖和寫入鎖的方式上不一樣，所以前面才有讀寫鎖是獨(dú)占鎖的兩個不同視圖一說。
ReentrantReadWriteLock里面有兩個類：ReadLock/WriteLock，這兩個類都是Lock的實(shí)現(xiàn)。
清單1 ReadLock 片段
public static class ReadLock implements Lock, java.io.Serializable? {
??? private final Sync sync;
??? protected ReadLock(ReentrantReadWriteLock lock) {
??????? sync = lock.sync;
??? }
??? public void lock() {
??????? sync.acquireShared(1);
??? }
??? public void lockInterruptibly() throws InterruptedException {
??????? sync.acquireSharedInterruptibly(1);
??? }
??? public? boolean tryLock() {
??????? return sync.tryReadLock();
??? }
??? public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
??????? return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
??? }
??? public? void unlock() {
??????? sync.releaseShared(1);
??? }
??? public Condition newCondition() {
??????? throw new UnsupportedOperationException();
??? }
}
清單2 WriteLock 片段
public static class WriteLock implements Lock, java.io.Serializable? {
??? private final Sync sync;
??? protected WriteLock(ReentrantReadWriteLock lock) {
??????? sync = lock.sync;
??? }
??? public void lock() {
??????? sync.acquire(1);
??? }
??? public void lockInterruptibly() throws InterruptedException {
??????? sync.acquireInterruptibly(1);
??? }
??? public boolean tryLock( ) {
??????? return sync.tryWriteLock();
??? }
??? public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
??????? return sync.tryAcquireNanos(1, unit.toNanos(timeout));
??? }
??? public void unlock() {
??????? sync.release(1);
??? }
??? public Condition newCondition() {
??????? return sync.newCondition();
??? }
??? public boolean isHeldByCurrentThread() {
??????? return sync.isHeldExclusively();
??? }
??? public int getHoldCount() {
??????? return sync.getWriteHoldCount();
??? }
}
清單1描述的是讀鎖的實(shí)現(xiàn)，清單2描述的是寫鎖的實(shí)現(xiàn)。顯然WriteLock就是一個獨(dú)占鎖，這和ReentrantLock里面的實(shí)現(xiàn)幾乎相同，都是使用了AQS的acquire/release操作。當(dāng)然了在內(nèi)部處理方式上與ReentrantLock還是有一點(diǎn)不同的。對比清單1和清單2可以看到，ReadLock獲取的是共享鎖，WriteLock獲取的是獨(dú)占鎖。
在AQS章節(jié)中介紹到AQS中有一個state字段（int類型，32位）用來描述有多少線程獲持有鎖。在獨(dú)占鎖的時代這個值通常是0或者1（如果是重入的就是重入的次數(shù)），在共享鎖的時代就是持有鎖的數(shù)量。在上一節(jié)中談到，ReadWriteLock的讀、寫鎖是相關(guān)但是又不一致的，所以需要兩個數(shù)來描述讀鎖（共享鎖）和寫鎖（獨(dú)占鎖）的數(shù)量。顯然現(xiàn)在一個state就不夠用了。于是在ReentrantReadWrilteLock里面將這個字段一分為二，高位16位表示共享鎖的數(shù)量，低位16位表示獨(dú)占鎖的數(shù)量（或者重入數(shù)量）。2^16-1=65536，這就是上節(jié)中提到的為什么共享鎖和獨(dú)占鎖的數(shù)量最大只能是65535的原因了。
有了上面的知識后再來分析讀寫鎖的獲取和釋放就容易多了。
清單3 寫入鎖獲取片段
protected final boolean tryAcquire(int acquires) {
??? Thread current = Thread.currentThread();
??? int c = getState();
??? int w = exclusiveCount(c);
??? if (c != 0) {
??????? if (w == 0 || current != getExclusiveOwnerThread())
??????????? return false;
??????? if (w + exclusiveCount(acquires) > MAX_COUNT)
??????????? throw new Error("Maximum lock count exceeded");
??? }
??? if ((w == 0 && writerShouldBlock(current)) ||
??????? !compareAndSetState(c, c + acquires))
??????? return false;
??? setExclusiveOwnerThread(current);
??? return true;
}
清單3 是寫入鎖獲取的邏輯片段，整個工作流程是這樣的：
清單3 中 exclusiveCount(c)就是獲取寫線程數(shù)（包括重入數(shù)），也就是state的低16位值。另外這里有一段邏輯是當(dāng)前寫線程是否需要阻塞writerShouldBlock(current)。清單4 和清單5 就是公平鎖和非公平鎖中是否需要阻塞的片段。很顯然對于非公平鎖而言總是不阻塞當(dāng)前線程，而對于公平鎖而言如果AQS隊列不為空或者當(dāng)前線程不是在AQS的隊列頭那么就阻塞線程，直到隊列前面的線程處理完鎖邏輯。
清單4 公平讀寫鎖寫線程是否阻塞
final boolean writerShouldBlock(Thread current) {
??? return !isFirst(current);
}
清單5 非公平讀寫鎖寫線程是否阻塞
final boolean writerShouldBlock(Thread current) {
??? return false;
}
寫入鎖的獲取邏輯清楚后，釋放鎖就比較簡單了。清單6 描述的寫入鎖釋放邏輯片段，其實(shí)就是檢測下剩下的寫入鎖數(shù)量，如果是0就將獨(dú)占鎖線程清空（意味著沒有線程獲取鎖），否則就是說當(dāng)前是重入鎖的一次釋放，所以不能將獨(dú)占鎖線程清空。然后將剩余線程狀態(tài)數(shù)寫回AQS。
清單6 寫入鎖釋放邏輯片段
protected final boolean tryRelease(int releases) {
??? int nextc = getState() - releases;
??? if (Thread.currentThread() != getExclusiveOwnerThread())
??????? throw new IllegalMonitorStateException();
??? if (exclusiveCount(nextc) == 0) {
??????? setExclusiveOwnerThread(null);
??????? setState(nextc);
??????? return true;
??? } else {
??????? setState(nextc);
??????? return false;
??? }
}
清單3~6 描述的寫入鎖的獲取釋放過程。讀取鎖的獲取和釋放過程要稍微復(fù)雜些。 清單7描述的是讀取鎖的獲取過程。
清單7 讀取鎖獲取過程片段
protected final int tryAcquireShared(int unused) {
??? Thread current = Thread.currentThread();
??? int c = getState();
??? if (exclusiveCount(c) != 0 &&
??????? getExclusiveOwnerThread() != current)
??????? return -1;
??? if (sharedCount(c) == MAX_COUNT)
??????? throw new Error("Maximum lock count exceeded");
??? if (!readerShouldBlock(current) &&
??????? compareAndSetState(c, c + SHARED_UNIT)) {
??????? HoldCounter rh = cachedHoldCounter;
??????? if (rh == null || rh.tid != current.getId())
??????????? cachedHoldCounter = rh = readHolds.get();
??????? rh.count++;
??????? return 1;
??? }
??? return fullTryAcquireShared(current);
}
final int fullTryAcquireShared(Thread current) {
??? HoldCounter rh = cachedHoldCounter;
??? if (rh == null || rh.tid != current.getId())
??????? rh = readHolds.get();
??? for (;;) {
??????? int c = getState();
??????? int w = exclusiveCount(c);
??????? if ((w != 0 && getExclusiveOwnerThread() != current) ||
??????????? ((rh.count | w) == 0 && readerShouldBlock(current)))
??????????? return -1;
??????? if (sharedCount(c) == MAX_COUNT)
??????????? throw new Error("Maximum lock count exceeded");
??????? if (compareAndSetState(c, c + SHARED_UNIT)) {
??????????? cachedHoldCounter = rh; // cache for release
??????????? rh.count++;
??????????? return 1;
??????? }
??? }
}
讀取鎖獲取的過程是這樣的：
在清單7 中有一個對象HoldCounter，這里暫且不提這是什么結(jié)構(gòu)和為什么存在這樣一個結(jié)構(gòu)。
接下來根據(jù)清單8 我們來看如何釋放一個讀取鎖。同樣先不理HoldCounter，關(guān)鍵的在于for循環(huán)里面，其實(shí)就是一個不斷嘗試的CAS操作，直到修改狀態(tài)成功。前面說過state的高16位描述的共享鎖（讀取鎖）的數(shù)量，所以每次都需要減去2^16，這樣就相當(dāng)于讀取鎖數(shù)量減1。實(shí)際上SHARED_UNIT=1<<16。
清單8 讀取鎖釋放過程
protected final boolean tryReleaseShared(int unused) {
??? HoldCounter rh = cachedHoldCounter;
??? Thread current = Thread.currentThread();
??? if (rh == null || rh.tid != current.getId())
??????? rh = readHolds.get();
??? if (rh.tryDecrement() <= 0)
??????? throw new IllegalMonitorStateException();
??? for (;;) {
??????? int c = getState();
??????? int nextc = c - SHARED_UNIT;
??????? if (compareAndSetState(c, nextc))
??????????? return nextc == 0;
??? }
}
好了，現(xiàn)在回頭看HoldCounter到底是一個什么東西。首先我們可以看到只有在獲取共享鎖（讀取鎖）的時候加1，也只有在釋放共享鎖的時候減1有作用，并且在釋放鎖的時候拋出了一個IllegalMonitorStateException異常。而我們知道IllegalMonitorStateException通常描述的是一個線程操作一個不屬于自己的監(jiān)視器對象的引發(fā)的異常。也就是說這里的意思是一個線程釋放了一個不屬于自己或者不存在的共享鎖。
前面的章節(jié)中一再強(qiáng)調(diào)，對于共享鎖，其實(shí)并不是鎖的概念，更像是計數(shù)器的概念。一個共享鎖就相對于一次計數(shù)器操作，一次獲取共享鎖相當(dāng)于計數(shù)器加1，釋放一個共享鎖就相當(dāng)于計數(shù)器減1。顯然只有線程持有了共享鎖（也就是當(dāng)前線程攜帶一個計數(shù)器，描述自己持有多少個共享鎖或者多重共享鎖），才能釋放一個共享鎖。否則一個沒有獲取共享鎖的線程調(diào)用一次釋放操作就會導(dǎo)致讀寫鎖的state（持有鎖的線程數(shù)，包括重入數(shù)）錯誤。
明白了HoldCounter的作用后我們就可以猜到它的作用其實(shí)就是當(dāng)前線程持有共享鎖（讀取鎖）的數(shù)量，包括重入的數(shù)量。那么這個數(shù)量就必須和線程綁定在一起。
在Java里面將一個對象和線程綁定在一起，就只有ThreadLocal才能實(shí)現(xiàn)了。所以毫無疑問HoldCounter就應(yīng)該是綁定到線程上的一個計數(shù)器。
清單9 線程持有讀取鎖數(shù)量的計數(shù)器
static final class HoldCounter {
??? int count;
??? final long tid = Thread.currentThread().getId();
??? int tryDecrement() {
??????? int c = count;
??????? if (c > 0)
??????????? count = c - 1;
??????? return c;
??? }
}
static final class ThreadLocalHoldCounter
??? extends ThreadLocal<HoldCounter> {
??? public HoldCounter initialValue() {
??????? return new HoldCounter();
??? }
}
清單9 描述的是線程持有讀取鎖數(shù)量的計數(shù)器。可以看到這里使用ThreadLocal將HoldCounter綁定到當(dāng)前線程上，同時HoldCounter也持有線程Id，這樣在釋放鎖的時候才能知道ReadWriteLock里面緩存的上一個讀取線程（cachedHoldCounter）是否是當(dāng)前線程。這樣做的好處是可以減少ThreadLocal.get()的次數(shù)，因為這也是一個耗時操作。需要說明的是這樣HoldCounter綁定線程id而不綁定線程對象的原因是避免HoldCounter和ThreadLocal互相綁定而GC難以釋放它們（盡管GC能夠智能的發(fā)現(xiàn)這種引用而回收它們，但是這需要一定的代價），所以其實(shí)這樣做只是為了幫助GC快速回收對象而已。
除了readLock()和writeLock()外，Lock對象還允許tryLock()，那么ReadLock和WriteLock的tryLock()不一樣。清單10 和清單11 分別描述了讀取鎖的tryLock()和寫入鎖的tryLock()。
讀取鎖tryLock()也就是tryReadLock()成功的條件是：沒有寫入鎖或者寫入鎖是當(dāng)前線程，并且讀線程共享鎖數(shù)量沒有超過65535個。
寫入鎖tryLock()也就是tryWriteLock()成功的條件是: 沒有寫入鎖或者寫入鎖是當(dāng)前線程，并且嘗試一次修改state成功。
清單10 讀取鎖的tryLock()
final boolean tryReadLock() {
??? Thread current = Thread.currentThread();
??? for (;;) {
??????? int c = getState();
??????? if (exclusiveCount(c) != 0 &&
??????????? getExclusiveOwnerThread() != current)
??????????? return false;
??????? if (sharedCount(c) == MAX_COUNT)
??????????? throw new Error("Maximum lock count exceeded");
??????? if (compareAndSetState(c, c + SHARED_UNIT)) {
??????????? HoldCounter rh = cachedHoldCounter;
??????????? if (rh == null || rh.tid != current.getId())
??????????????? cachedHoldCounter = rh = readHolds.get();
??????????? rh.count++;
??????????? return true;
??????? }
??? }
}
清單11 寫入鎖的tryLock()
final boolean tryWriteLock() {
??? Thread current = Thread.currentThread();
??? int c = getState();
??? if (c != 0) {
??????? int w = exclusiveCount(c);
??????? if (w == 0 ||current != getExclusiveOwnerThread())
??????????? return false;
??????? if (w == MAX_COUNT)
??????????? throw new Error("Maximum lock count exceeded");
??? }
??? if (!compareAndSetState(c, c + 1))
??????? return false;
??? setExclusiveOwnerThread(current);
??? return true;
}
?
整個讀寫鎖的邏輯大概就這么多，其實(shí)真正研究起來也不是很復(fù)雜，真正復(fù)雜的東西都在AQS里面。
十、鎖的一些其他問題
主要談?wù)勬i的性能以及其它一些理論知識，內(nèi)容主要的出處是《Java Concurrency in Practice》，結(jié)合自己的理解和實(shí)際應(yīng)用對鎖機(jī)制進(jìn)行一個小小的總結(jié)。
?
首先需要強(qiáng)調(diào)的一點(diǎn)是：所有鎖（包括內(nèi)置鎖和高級鎖）都是有性能消耗的，也就是說在高并發(fā)的情況下，由于鎖機(jī)制帶來的上下文切換、資源同步等消耗是非常可觀的。在某些極端情況下，線程在鎖上的消耗可能比線程本身的消耗還要多。所以如果可能的話，在任何情況下都盡量少用鎖，如果不可避免那么采用非阻塞算法是一個不錯的解決方案，但是卻也不是絕對的。
?
內(nèi)部鎖
Java語言通過synchronized關(guān)鍵字來保證原子性。這是因為每一個Object都有一個隱含的鎖，這個也稱作監(jiān)視器對象。在進(jìn)入synchronized之前自動獲取此內(nèi)部鎖，而一旦離開此方式（不管通過和中方式離開此方法）都會自動釋放鎖。顯然這是一個獨(dú)占鎖，每個鎖請求之間是互斥的。相對于前面介紹的眾多高級鎖（Lock/ReadWriteLock等），synchronized的代價都比后者要高。但是synchronized的語法比較簡單，而且也比較容易使用和理解，不容易寫法上的錯誤。而我們知道Lock一旦調(diào)用了lock()方法獲取到鎖而未正確釋放的話很有可能就死鎖了。所以Lock的釋放操作總是跟在finally代碼塊里面，這在代碼結(jié)構(gòu)上也是一次調(diào)整和冗余。另外前面介紹中說過Lock的實(shí)現(xiàn)已經(jīng)將硬件資源用到了極致，所以未來可優(yōu)化的空間不大，除非硬件有了更高的性能。但是synchronized只是規(guī)范的一種實(shí)現(xiàn)，這在不同的平臺不同的硬件還有很高的提升空間，未來Java在鎖上的優(yōu)化也會主要在這上面。
?
性能
由于鎖總是帶了性能影響，所以是否使用鎖和使用鎖的場合就變得尤為重要。如果在一個高并發(fā)的Web請求中使用了強(qiáng)制的獨(dú)占鎖，那么就可以發(fā)現(xiàn)Web的吞吐量將急劇下降。
為了利用并發(fā)來提高性能，出發(fā)點(diǎn)就是：更有效的利用現(xiàn)有的資源，同時讓程序盡可能的開拓更多可用的資源。這意味著機(jī)器盡可能的處于忙碌的狀態(tài)，通常意義是說CPU忙于計算，而不是等待。當(dāng)然CPU要做有用的事情，而不是進(jìn)行無謂的循環(huán)。當(dāng)然在實(shí)踐中通常會預(yù)留一些資源出來以便應(yīng)急特殊情況，這在以后的線程池并發(fā)中可以看到很多例子。
?
線程阻塞
鎖機(jī)制的實(shí)現(xiàn)通常需要操作系統(tǒng)提供支持，顯然這會增加開銷。當(dāng)鎖競爭的時候，失敗的線程必然會發(fā)生阻塞。JVM既能自旋等待（不斷嘗試，知道成功，很多CAS就是這樣實(shí)現(xiàn)的），也能夠在操作系統(tǒng)中掛起阻塞的線程，直到超時或者被喚醒。通常情況下這取決于上下文切換的開銷以及與獲取鎖需要等待的時間二者之間的關(guān)系。自旋等待適合于比較短的等待，而掛起線程比較適合那些比較耗時的等待。
掛起一個線程可能是因為無法獲取到鎖，或者需要某個特定的條件，或者耗時的I/O操作。掛起一個線程需要兩次額外的上下文切換以及操作系統(tǒng)、緩存等多資源的配合：如果線程被提前換出，那么一旦拿到鎖或者條件滿足，那么又需要將線程換回執(zhí)行隊列，這對線程而言，兩次上下文切換可能比較耗時。
?
鎖競爭
影響鎖競爭性的條件有兩個：鎖被請求的頻率和每次持有鎖的時間。顯然當(dāng)而這二者都很小的時候，鎖競爭不會成為主要的瓶頸。但是如果鎖使用不當(dāng)，導(dǎo)致二者都比較大，那么很有可能CPU不能有效的處理任務(wù)，任務(wù)被大量堆積。
所以減少鎖競爭的方式有下面三種：
?
死鎖
如果一個線程永遠(yuǎn)不釋放另外一個線程需要的資源那么就會導(dǎo)致死鎖。這有兩種情況：一種情況是線程A永遠(yuǎn)不釋放鎖，結(jié)果B一直拿不到鎖，所以線程B就“死掉”了；第二種情況下，線程A擁有線程B需要的鎖Y，同時線程B擁有線程A需要的鎖X，那么這時候線程A/B互相依賴對方釋放鎖，于是二者都“死掉”了。
還有一種情況為發(fā)生死鎖，如果一個線程總是不能被調(diào)度，那么等待此線程結(jié)果的線程可能就死鎖了。這種情況叫做線程饑餓死鎖。比如說在前面介紹的非公平鎖中，如果某些線程非常活躍，在高并發(fā)情況下這類線程可能總是拿到鎖，那么那些活躍度低的線程可能就一直拿不到鎖，這樣就發(fā)生了“饑餓死”。
避免死鎖的解決方案是：盡可能的按照鎖的使用規(guī)范請求鎖，另外鎖的請求粒度要小（不要在不需要鎖的地方占用鎖，鎖不用了盡快釋放）；在高級鎖里面總是使用tryLock或者定時機(jī)制（這個以后會講，就是指定獲取鎖超時的時間，如果時間到了還沒有獲取到鎖那么就放棄）。高級鎖（Lock）里面的這兩種方式可以有效的避免死鎖。
?
活鎖
活鎖描述的是線程總是嘗試某項操作卻總是失敗的情況。這種情況下盡管線程沒有被阻塞，但是人物卻總是不能被執(zhí)行。比如在一個死循環(huán)里面總是嘗試做某件事，結(jié)果卻總是失敗，現(xiàn)在線程將永遠(yuǎn)不能跳出這個循環(huán)。另外一種情況是在一個隊列中每次從隊列頭取出一個任務(wù)來執(zhí)行，每次都失敗，然后將任務(wù)放入隊列頭，接下來再一次從隊列頭取出任務(wù)執(zhí)行，仍然失敗。
還有一種活鎖方式發(fā)生在“碰撞協(xié)讓”情況下：兩個人過獨(dú)木橋，如果在半路相撞，雙方禮貌退出去然后再試一次。如果總是失敗，那么這兩個任務(wù)將一直無法得到執(zhí)行。
?
總之解決鎖問題的關(guān)鍵就是：從簡單的開始，先保證正確，然后再開始優(yōu)化。
?
轉(zhuǎn)載于:https://www.cnblogs.com/bopo/p/9242846.html
總結(jié)
以上是生活随笔為你收集整理的Java多线程(五) —— 线程并发库之锁机制的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 基于java的房屋销售管理系统设计(含源
- 下一篇： 友盟分享Title设置