深入淺出 Java Concurrency (7): 鎖機制 part 2 AQS

dtl樂學(xué)館 2015-06-10

展開全文

在理解J.U.C原理以及鎖機制之前，我們來介紹J.U.C框架最核心也是最復(fù)雜的一個基礎(chǔ)類：java.util.concurrent.locks.AbstractQueuedSynchronizer。

AQS

AbstractQueuedSynchronizer，簡稱AQS，是J.U.C最復(fù)雜的一個類，導(dǎo)致絕大多數(shù)講解并發(fā)原理或者實戰(zhàn)的時候都不會提到此類。但是虛心的作者愿意借助自己有限的能力和精力來探討一二（參考資源中也有一些作者做了部分的分析。）。

首先從理論知識開始，在了解了相關(guān)原理后會針對源碼進行一些分析，最后加上一些實戰(zhàn)來描述。

上面的繼承體系中，AbstractQueuedSynchronizer是CountDownLatch/FutureTask/ReentrantLock/RenntrantReadWriteLock/Semaphore的基礎(chǔ)，因此AbstractQueuedSynchronizer是Lock/Executor實現(xiàn)的前提。公平鎖、不公平鎖、Condition、CountDownLatch、Semaphore等放到后面的篇幅中說明。

完整的設(shè)計原理可以參考Doug Lea的論文 The java.util.concurrent Synchronizer Framework ，這里做一些簡要的分析。

基本的思想是表現(xiàn)為一個同步器，支持下面兩個操作：

獲取鎖：首先判斷當前狀態(tài)是否允許獲取鎖，如果是就獲取鎖，否則就阻塞操作或者獲取失敗，也就是說如果是獨占鎖就可能阻塞，如果是共享鎖就可能失敗。另外如果是阻塞線程，那么線程就需要進入阻塞隊列。當狀態(tài)位允許獲取鎖時就修改狀態(tài)，并且如果進了隊列就從隊列中移除。

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)位
阻塞和喚醒線程
一個有序的隊列

目標明確，要解決的問題也清晰了，那么剩下的就是解決上面三個問題。

狀態(tài)位的原子操作

這里使用一個32位的整數(shù)來描述狀態(tài)位，前面章節(jié)的原子操作的理論知識整好派上用場，在這里依然使用CAS操作來解決這個問題。事實上這里還有一個64位版本的同步器（AbstractQueuedLongSynchronizer），這里暫且不談。

阻塞和喚醒線程

標準的JAVA API里面是無法掛起（阻塞）一個線程，然后在將來某個時刻再喚醒它的。JDK 1.0的API里面有Thread.suspend和Thread.resume，并且一直延續(xù)了下來。但是這些都是過時的API，而且也是不推薦的做法。

在JDK 5.0以后利用JNI在LockSupport類中實現(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()是在當前線程中調(diào)用，導(dǎo)致線程阻塞，帶參數(shù)的Object是掛起的對象，這樣監(jiān)視的時候就能夠知道此線程是因為什么資源而阻塞的。由于park()立即返回，所以通常情況下需要在循環(huán)中去檢測競爭資源來決定是否進行下一次阻塞。park()返回的原因有三：

其他某個線程調(diào)用將當前線程作為目標調(diào)用 unpark；
其他某個線程中斷當前線程；
該調(diào)用不合邏輯地（即毫無理由地）返回。

其實第三條就決定了需要循環(huán)檢測了，類似于通常寫的while(checkCondition()){Thread.sleep(time);}類似的功能。

有序隊列

在AQS中采用CHL列表來解決有序的隊列的問題。

AQS采用的CHL模型采用下面的算法完成FIFO的入隊列和出隊列過程。

對于入隊列(enqueue)：采用CAS操作，每次比較尾結(jié)點是否一致，然后插入的到尾結(jié)點中。

do {

pred = tail;

}while ( !compareAndSet(pred,tail,node) );

對于出隊列(dequeue):由于每一個節(jié)點也緩存了一個狀態(tài)，決定是否出隊列，因此當不滿足條件時就需要自旋等待，一旦滿足條件就將頭結(jié)點設(shè)置為下一個節(jié)點。

while (pred.status != RELEASED) ;

head = node;

實際上這里自旋等待也是使用LockSupport.park()來實現(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é)點的屬性。

volatile int waitStatus; 節(jié)點的等待狀態(tài)，一個節(jié)點可能位于以下幾種狀態(tài)：

CANCELLED = 1：節(jié)點操作因為超時或者對應(yīng)的線程被interrupt。節(jié)點不應(yīng)該留在此狀態(tài)，一旦達到此狀態(tài)將從CHL隊列中踢出。
SIGNAL = -1：節(jié)點的繼任節(jié)點是（或者將要成為）BLOCKED狀態(tài)（例如通過LockSupport.park()操作），因此一個節(jié)點一旦被釋放（解鎖）或者取消就需要喚醒（LockSupport.unpack()）它的繼任節(jié)點。
CONDITION = -2：表明節(jié)點對應(yīng)的線程因為不滿足一個條件（Condition）而被阻塞。
0：正常狀態(tài)，新生的非CONDITION節(jié)點都是此狀態(tài)。
非負值標識節(jié)點不需要被通知（喚醒）。

volatile Node prev;此節(jié)點的前一個節(jié)點。節(jié)點的waitStatus依賴于前一個節(jié)點的狀態(tài)。

volatile Node next;此節(jié)點的后一個節(jié)點。后一個節(jié)點是否被喚醒（uppark()）依賴于當前節(jié)點是否被釋放。

volatile Thread thread;節(jié)點綁定的線程。

Node nextWaiter;下一個等待條件（Condition）的節(jié)點，由于Condition是獨占模式，因此這里有一個簡單的隊列來描述Condition上的線程節(jié)點。