java condition_死磕Java并发：J.U.C之Condition
在沒有Lock之前，我們使用synchronized來控制同步，配合Object的wait()、notify()系列方法可以實現等待/通知模式。在Java SE5后，Java提供了Lock接口，相對于Synchronized而言，Lock提供了條件Condition，對線程的等待、喚醒操作更加詳細和靈活。下圖是Condition與Object的監視器方法的對比(摘自《Java并發編程的藝術》)：
Condition提供了一系列的方法來對阻塞和喚醒線程：
Condition是一種廣義上的條件隊列。他為線程提供了一種更為靈活的等待/通知模式，線程在調用await方法后執行掛起操作，直到線程等待的某個條件為真時才會被喚醒。Condition必須要配合鎖一起使用，因為對共享狀態變量的訪問發生在多線程環境下。一個Condition的實例必須與一個Lock綁定，因此Condition一般都是作為Lock的內部實現。
1、Condtion的實現
獲取一個Condition必須要通過Lock的newCondition()方法。該方法定義在接口Lock下面，返回的結果是綁定到此 Lock 實例的新 Condition 實例。
Condition為一個接口，其下僅有一個實現類ConditionObject，由于Condition的操作需要獲取相關的鎖，而AQS則是同步鎖的實現基礎，所以ConditionObject則定義為AQS的內部類。定義如下：
- 等待隊列
每個Condition對象都包含著一個FIFO隊列，該隊列是Condition對象通知/等待功能的關鍵。在隊列中每一個節點都包含著一個線程引用，該線程就是在該Condition對象上等待的線程。我們看Condition的定義就明白了：
從上面代碼可以看出Condition擁有首節點(firstWaiter)，尾節點(lastWaiter)。當前線程調用await()方法，將會以當前線程構造成一個節點(Node)，并將節點加入到該隊列的尾部。結構如下：
Node里面包含了當前線程的引用。Node定義與AQS的CLH同步隊列的節點使用的都是同一個類(AbstractQueuedSynchronized.Node靜態內部類)。
Condition的隊列結構比CLH同步隊列的結構簡單些，新增過程較為簡單只需要將原尾節點的nextWaiter指向新增節點，然后更新lastWaiter即可。
- 等待
調用Condition的await()方法會使當前線程進入等待狀態，同時會加入到Condition等待隊列同時釋放鎖。當從await()方法返回時，當前線程一定是獲取了Condition相關連的鎖。
public final void await() throws InterruptedException {
// 當前線程中斷
if (Thread.interrupted())
throw new InterruptedException();
//當前線程加入等待隊列
Node node = addConditionWaiter();
//釋放鎖
long 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);
}
此段代碼的邏輯是：首先將當前線程新建一個節點同時加入到條件隊列中，然后釋放當前線程持有的同步狀態。然后則是不斷檢測該節點代表的線程釋放出現在CLH同步隊列中(收到signal信號之后就會在AQS隊列中檢測到)，如果不存在則一直掛起，否則參與競爭同步狀態。
加入條件隊列(addConditionWaiter())源碼如下：
private Node addConditionWaiter() {
Node t = lastWaiter; //尾節點
//Node的節點狀態如果不為CONDITION，則表示該節點不處于等待狀態，需要清除節點
if (t != null && t.waitStatus != Node.CONDITION) {
//清除條件隊列中所有狀態不為Condition的節點
unlinkCancelledWaiters();
t = lastWaiter;
}
//當前線程新建節點，狀態CONDITION
Node node = new Node(Thread.currentThread(), Node.CONDITION);
/**
* 將該節點加入到條件隊列中最后一個位置
*/
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}
該方法主要是將當前線程加入到Condition條件隊列中。當然在加入到尾節點之前會清楚所有狀態不為Condition的節點。
fullyRelease(Node node)，負責釋放該線程持有的鎖。
final long fullyRelease(Node node) {
boolean failed = true;
try {
//節點狀態--其實就是持有鎖的數量
long savedState = getState();
//釋放鎖
if (release(savedState)) {
failed = false;
return savedState;
} else {
throw new IllegalMonitorStateException();
}
} finally {
if (failed)
node.waitStatus = Node.CANCELLED;
}
}
isOnSyncQueue(Node node)：如果一個節點剛開始在條件隊列上，現在在同步隊列上獲取鎖則返回true。
final boolean isOnSyncQueue(Node node) {
//狀態為Condition，獲取前驅節點為null，返回false
if (node.waitStatus == Node.CONDITION || node.prev == null)
return false;
//后繼節點不為null，肯定在CLH同步隊列中
if (node.next != null)
return true;
return findNodeFromTail(node);
}
unlinkCancelledWaiters()：負責將條件隊列中狀態不為Condition的節點刪除。
private void unlinkCancelledWaiters() {
Node t = firstWaiter;
Node trail = null;
while (t != null) {
Node next = t.nextWaiter;
if (t.waitStatus != Node.CONDITION) {
t.nextWaiter = null;
if (trail == null)
firstWaiter = next;
else
trail.nextWaiter = next;
if (next == null)
lastWaiter = trail;
}
else
trail = t;
t = next;
}
}
- 通知
調用Condition的signal()方法，將會喚醒在等待隊列中等待最長時間的節點(條件隊列里的首節點)，在喚醒節點前，會將節點移到CLH同步隊列中。
public final void signal() {
//檢測當前線程是否為擁有鎖的獨
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
//頭節點，喚醒條件隊列中的第一個節點
Node first = firstWaiter;
if (first != null)
doSignal(first); //喚醒
}
該方法首先會判斷當前線程是否已經獲得了鎖，這是前置條件。然后喚醒條件隊列中的頭節點。
doSignal(Node first)：喚醒頭節點。
private void doSignal(Node first) {
do {
//修改頭結點，完成舊頭結點的移出工作
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
doSignal(Node first)主要是做兩件事：
transferForSignal(Node first)源碼如下：
final boolean transferForSignal(Node node) {
//將該節點從狀態CONDITION改變為初始狀態0,
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
//將節點加入到syn隊列中去，返回的是syn隊列中node節點前面的一個節點
Node p = enq(node);
int ws = p.waitStatus;
//如果結點p的狀態為cancel 或者修改waitStatus失敗，則直接喚醒
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}
整個通知的流程如下：
- 總結
一個線程獲取鎖后，通過調用Condition的await()方法，會將當前線程先加入到條件隊列中，然后釋放鎖，最后通過isOnSyncQueue(Node node)方法不斷自檢看節點是否已經在CLH同步隊列了，如果是則嘗試獲取鎖，否則一直掛起。
當線程調用signal()方法后，程序首先檢查當前線程是否獲取了鎖，然后通過doSignal(Node first)方法喚醒CLH同步隊列的首節點。被喚醒的線程，將從await()方法中的while循環中退出來，然后調用acquireQueued()方法競爭同步狀態。
2、Condition的應用
只知道原理，如果不知道使用那就坑爹了，下面是用Condition實現的生產者消費者問題：
public class Condition{
private LinkedList buffer; //容器
private int maxSize ;
private Lock lock;
private Condition fullCondition;
private Condition notFullCondition;
ConditionTest(int maxSize){
this.maxSize = maxSize;
buffer = new LinkedList();
lock = new ReentrantLock();
fullCondition = lock.newCondition();
notFullCondition = lock.newCondition();
}
public void set(String string) throws InterruptedException {
lock.lock(); //獲取鎖
try {
while (maxSize == buffer.size()){
notFullCondition.await(); //滿了，添加的線程進入等待狀態
}
buffer.add(string);
fullCondition.signal();
} finally {
lock.unlock(); //記得釋放鎖
}
}
public String get() throws InterruptedException {
String string;
lock.lock();
try {
while (buffer.size() == 0){
fullCondition.await();
}
string = buffer.poll();
notFullCondition.signal();
} finally {
lock.unlock();
}
return string;
}
}
總結
以上是生活随笔為你收集整理的java condition_死磕Java并发：J.U.C之Condition的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： python的property用法_Py
- 下一篇： java最好的开发工具_IntelliJ