在上一篇博客,簡單的說下了AQS的基本概念,核心源碼解析,可是還有一部份內容沒有涉及到,就是AQS對條件變量的支持,這篇博客將着重介紹這方面的內容。node
咱們先經過模擬一個消費者/生產者模型來看下條件變量的基本應用:bash
public class CommonResource {
private boolean isHaveData = false;
Lock lock = new ReentrantLock();
Condition producer_con = lock.newCondition();
Condition consumer_con = lock.newCondition();
public void product() {
lock.lock();
try {
while (isHaveData) {
try {
System.out.println("還有數據,等待消費數據");
producer_con.await();
} catch (InterruptedException e) {
}
}
System.out.println("生產者生產數據了");
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
isHaveData = true;
consumer_con.signal();
} finally {
lock.unlock();
}
}
public void consume() {
lock.lock();
try {
while (!isHaveData) {
try {
System.out.println("沒有數據了,等待生產者消費數據");
consumer_con.await();
} catch (InterruptedException e) {
}
}
System.out.println("消費者消費數據");
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
isHaveData = false;
producer_con.signal();
} finally {
lock.unlock();
}
}
}
複製代碼
public class Main {
public static void main(String[] args) {
CommonResource resource = new CommonResource();
new Thread(() -> {
while (true) {
resource.product();
}
}).start();
new Thread(() -> {
while (true) {
resource.consume();
}
}).start();
}
}
複製代碼
運行結果: less
這就是條件變量的應用,第一反應是否是和object中的wait/nofity很像,wait/nofity是配合synchronized工做的,而條件變量的await/signal是配合使用AQS實現的鎖 來完成工做的,固然也要看用AQS實現的鎖是否支持了條件變量。synchronized只能與一個共享變量進行工做,而AQS實現的鎖支持多個條件變量。ui
咱們試着分析下上面的代碼:this
首先建立了兩個條件變量,一個條件變量用來阻塞/喚醒消費者線程,一個條件變量用來阻塞/喚醒生產者線程。spa
生產者,首先獲取了獨佔鎖,判斷是否有數據:線程
最終釋放鎖。code
消費者,首先獲取了獨佔鎖,判斷是否有數據:cdn
最終釋放鎖。對象
這裏有兩點須要特別注意:
爲了加深對條件變量的理解,咱們再來看一個例子,兩個線程交替打印奇偶數:
public class Test {
private int num = 0;
private Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();
public void add() {
while(num<100) {
try {
lock.lock();
System.out.println(Thread.currentThread().getName() + ":" + num++);
condition.signal();
condition.await();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
}
複製代碼
public class Main {
public static void main(String[] args) {
Test test=new Test();
new Thread(() -> {
test.add();
}).start();
new Thread(() -> {
test.add();
}).start();
}
}
複製代碼
運行結果:
翻閱網上的大多數案例是分兩個線程方法交替打印,同時開兩個條件變量,其中一個條件變量負責阻塞/喚醒打印奇數的線程,一個變量負責阻塞/喚醒打印偶數的線程,可是我的以爲沒什麼必要,兩個線程共用一個線程方法,共用一個條件變量也能夠。不知道各位看官是什麼想的?
當咱們點開lock.newCondition,發現它有好幾個實現類:
public Condition newCondition() {
return sync.newCondition();
}
複製代碼
繼續往下點:
final ConditionObject newCondition() {
return new ConditionObject();
}
複製代碼
能夠看到,當咱們調用lock.newnewCondition,最終會new出一個ConditionObject對象,而ConditionObject類是AbstractQueuedSynchronizer的內部類,咱們先看下ConditionObject的UML圖:
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) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
複製代碼
在這裏,咱們就搞清楚三個問題便可:
第一個問題在addConditionWaiter方法能夠獲得答案:
private Node addConditionWaiter() {
Node t = lastWaiter;
// If lastWaiter is cancelled, clean out.
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
Node node = new Node(Thread.currentThread(), Node.CONDITION);
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}
複製代碼
首先是判斷條件隊列中的尾節點是否被取消了,若是被取消了,執行unlinkCancelledWaiters方法。咱們這裏確定沒有被取消,事實上,若是是第一次調用await方法,lastWaiter是爲空的,因此確定不會進入第一個if。隨後,新建一個Node,這個Node類就是上一篇博客中大量介紹過的,也是AbstractQueuedSynchronizer的內部類,也就是新建了一個Node節點,其中保存了當前線程和Node的類型,這裏Node的類型是CONDITION,若是t==null,則說明新建的Node是第一個節點,因此賦值給firstWaiter ,不然將尾節點的nextWaiter設置爲新Node,造成一個單向鏈表,這個nextWaiter在哪裏呢,它是經過node點出來的,也就是它也屬於node類的一個字段:
其實,AQS中的條件隊列也是一個阻塞隊列,只是爲了方便,因此在本篇博客中出現的AQS中的條件隊列特指在被條件變量await的,而阻塞隊列特指FIFO雙向鏈表隊列。
第一個問題解決了,咱們再來看第二個問題,第二個問題答案在await的第二個方法:
final int fullyRelease(Node node) {
boolean failed = true;
try {
int savedState = getState();
if (release(savedState)) {
failed = false;
return savedState;
} else {
throw new IllegalMonitorStateException();
}
} finally {
if (failed)
node.waitStatus = Node.CANCELLED;
}
}
複製代碼
首先調用getState方法,這個state是什麼,不知你們是否還有印象,對於ReentrantLock來講,state就是重入次數,隨後調用release方法,傳入state。也就是無論重入了多少次,這裏是一次性把鎖徹底釋放掉。
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方法,這個方法正是須要被重寫的。
當完成了前兩個方法的調用後,就會進行一個判斷isOnSyncQueue,通常來講會進入這個if,park這個線程,等待喚醒,這就解決了第三個問題。
下面咱們再來看看signal方法,一樣的,咱們須要解決幾個問題:
public final void signal() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignal(first);
}
複製代碼
重點在於doSignal中的transferForSignal方法:
final boolean transferForSignal(Node node) {
/*
* If cannot change waitStatus, the node has been cancelled.
*/
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
/*
* Splice onto queue and try to set waitStatus of predecessor to
* indicate that thread is (probably) waiting. If cancelled or
* attempt to set waitStatus fails, wake up to resync (in which
* case the waitStatus can be transiently and harmlessly wrong).
*/
Node p = enq(node);
int ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
複製代碼
在這個方法中,咱們能夠會調用enq方法,把條件隊列的線程放入阻塞隊列中,而後調用unpark方法,喚醒線程。
本篇博客到這裏也結束了。
通過上下兩篇博客,相信你們對AQS必定有了一個比較淺顯的理解。聰明的你,能夠看出來,其實這兩篇博客有不少內容都沒有講透,甚至有點模棱兩可,只是「走馬觀花」,因此這也符合了個人標題:難以理解的AQS。的確,AQS要深刻研究的話,不比線程池簡單多少。看,我又再給本身找理由了。但願通過從此的沉澱,我能夠把這兩篇博客重寫下,而後換個標題「完全理解AQS」,嘿嘿。