并發(fā)編程中的等待通知模型
為避免輪詢條件為真的開銷，并發(fā)編程中常用等待通知模型來優(yōu)化這一點，而本文將針對等待通知模型這一知識點進(jìn)行深入剖析，希望對你有所啟發(fā)。
一、同步鎖下的等待通知模型
1. 狀態(tài)依賴性的管理
在經(jīng)典的生產(chǎn)者和消費者模式中，我們經(jīng)常用到ArrayBlockingQueue作為并發(fā)安全的有界緩存，而該有界緩解進(jìn)行讀寫操作時都必須嚴(yán)格按照如下兩個條件謂語時機(jī)執(zhí)行，即：
- 針對阻塞隊列進(jìn)行元素存操作時，有界緩存必須有空閑空間，即可非滿
- 針對阻塞隊列進(jìn)行取操作時，有界隊列必須有元素，即非空
基于上述的說法，我們基于同步鎖synchronized實現(xiàn)了一個數(shù)組形式的環(huán)形阻塞隊列的核心方法模板，大體思路為：
- 當(dāng)進(jìn)行元素存操作時，互斥調(diào)用doPut函數(shù)，判斷是否到達(dá)數(shù)組末端，若到達(dá)則直接將元素存到索引0，并累加count
- 進(jìn)行元素取操作時，互斥上鎖執(zhí)行doTake，同樣執(zhí)行步驟1的邊界判斷，完成后扣減count
- 基于count判斷非空和非滿
我們的環(huán)形有界隊列是用數(shù)組實現(xiàn)的，所以筆者也用數(shù)組直觀的展現(xiàn)這個流程，當(dāng)然讀者可以在邏輯上將數(shù)組首位相接，即可構(gòu)成一個環(huán)形隊列：
對應(yīng)的筆者也給出這個環(huán)形隊列的抽象模板，核心函數(shù)思路和上述基本一致，讀者可結(jié)合圖文注釋了解大體流程，后文將基于該模板落地一個支持阻塞等待空閑通知線程存取元素的緩存隊列：
public abstract class BaseBoundedBuffer<V> {
private final V[] items;
private int head;
private int tail;
private int count;
/**
* 初始化環(huán)形有界隊列
*
* @param capacity 容量
*/
protected BaseBoundedBuffer(int capacity) {
items = (V[]) new Object[capacity];
}
protected synchronized final void doPut(V v) throws InterruptedException {
//尾節(jié)點添加元素
items[tail] = v;
//如果到達(dá)數(shù)組末端，則重新從0開始
if (++tail == items.length) {
tail = 0;
}
//累加元素個數(shù)
count++;
}
protected synchronized final V doTake() throws InterruptedException {
//頭節(jié)點取元素
V v = items[head];
//頭節(jié)點置空實現(xiàn)刪除
items[head] = null;
if (++head == items.length) {//如果到達(dá)邊界，則循環(huán)從0開始
head = 0;
}
//減元素個數(shù)
count--;
return v;
}
public synchronized final boolean isFull() {
return count == items.length;
}
public synchronized final boolean isEmpty() {
return count == 0;
}
}2. 基于異常式的隊列模型
我們先來看看第一個有界緩存的基本實現(xiàn)，一旦觸發(fā)如下兩個條件時，該緩存就會拋出異常：
- 獲取元素時隊列空
- 插入元素時隊列滿
對應(yīng)落地代碼如下，直接繼承有界隊列后落地落采用異常通知方式實現(xiàn)元素存取的緩存隊列：
public class GrumpyBoundedBuffer extends BaseBoundedBuffer<Integer> {
protected GrumpyBoundedBuffer(int capacity) {
super(capacity);
}
public synchronized void put(int value) throws Exception {
//隊列滿了，直接拋出異常
if (isFull()) {
throw new RuntimeException("queue is full");
}
//隊列沒滿，正常入隊
doPut(value);
}
public synchronized int take() throws Exception {
//隊列為空，直接拋出異常
if (isEmpty()) {
throw new RuntimeException("queue is empty");
}
//隊列不為空，正常出隊
return doTake();
}
}雖然這種方式使得緩存在實現(xiàn)非常的簡單，但是這種方案對于使用者來說非常的不友好，在業(yè)務(wù)正常的情況下，即使存取消費的緩存在單位時間滿即直接拋出異常告知線程不可存取，讓使用者手動捕獲異常進(jìn)行重試：
public static void main(String[] args) {
GrumpyBoundedBuffer grumpyBoundedBuffer = new GrumpyBoundedBuffer(1);
ThreadUtil.execAsync(() -> {
while (true) {
try {
grumpyBoundedBuffer.put(1);
} catch (Exception e) {
Console.error("隊列已滿，1s后重試");
ThreadUtil.sleep(1000);
}
}
});
}輸出結(jié)果如下所示，非常的不方便：
3. 輪詢檢測式的等待喚醒
于是我們就考慮在隊列存儲上在一個重試的的機(jī)制，即當(dāng)隊列存取失敗時，進(jìn)行休眠重試，直到成功后返回。
但是對于程序的性能表現(xiàn)而言，也是一種災(zāi)難，這種做法設(shè)計釋放鎖之后的休眠和循環(huán)重試，這就使得設(shè)計者需要在CPU使用率和響應(yīng)性之間做好權(quán)衡：
- 如果設(shè)置休眠時間相對短，那么重試就會盡可能快，響應(yīng)性就會越高，但是循環(huán)帶來的CPU資源的開銷卻急劇增加。
- 如果休眠時間設(shè)置過長，有概率完成任務(wù)處理，但是卻來響應(yīng)的延遲。
public class SleepyBoundedBuffer extends BaseBoundedBuffer<Integer> {
protected SleepyBoundedBuffer(int capacity) {
super(capacity);
}
/**
* 輪詢重試，直到成功
*
* @param value
* @throws InterruptedException
*/
public synchronized void put(int value) throws InterruptedException {
while (true) {
synchronized (this) {
if (!isFull()) {
doPut(value);
}
}
Console.log("隊列已滿，500ms后重試");
ThreadUtil.sleep(500);
}
}
public synchronized int take() throws InterruptedException {
while (true) {
synchronized (this) {
if (!isEmpty()) {
return doTake();
}
}
Console.log("隊列已空，500ms后重試");
ThreadUtil.sleep(500);
}
}
}這種方案一定程度解決用戶手動捕獲異常重試的繁瑣，但也存在著如下缺點：
- 重試時休眠間隔500ms可能太長也可能太短，固定值等待非常不合理
- 頻繁循環(huán)重試使得線程大量時間得到CPU時間片做一些無用功
- 重試多次無果后無法中斷
4. 基于條件等待的有界緩存
所以我們需要進(jìn)行進(jìn)一步的優(yōu)化即通過如下兩個列條件謂語避免線程無用的輪詢開銷：
- 當(dāng)隊列滿的時候，當(dāng)前存線程阻塞等待，直到隊列非空時被喚醒
- 當(dāng)隊列空的時候，取線程阻塞等待，知道隊列有元素時將其喚醒
總結(jié)起來就是一句話，非滿的時候喚醒存線程嘗試存元素，非空的時候通知取線程取元素，由此得出如下兩個條件謂語isNotFull和isNotEmpty:
所以我們需要以object中對應(yīng)的wait、notify和notifyAll構(gòu)成內(nèi)部條件隊列的交互通知，當(dāng)然要調(diào)用這些通知方法的前提也就是需要獲取當(dāng)前這個對象的鎖。
以我們有界緩存存元素操作為例，我們執(zhí)行添加操作時執(zhí)行步驟為：
- 獲得這個對象的鎖
- 當(dāng)發(fā)現(xiàn)緩存空間已滿即不符合檢測條件時，則調(diào)用當(dāng)前對象(有界緩存)的wait方法將當(dāng)前線程掛起
- 與此同時，線程也會釋放這把鎖，等待隊列非滿時通過notify或者notifyAll嘗試將當(dāng)前線程喚醒。
對應(yīng)我們給出代碼示例，這種方式相比于休眠的方案，改進(jìn)了響應(yīng)的效率和CPU使用率的開銷，避免了非必要的檢測步驟:
public class BoundedBuffer extends BaseBoundedBuffer<Integer> {
protected BoundedBuffer(int capacity) {
super(capacity);
}
public synchronized void put(int value) throws InterruptedException {
if (isFull()) {
Console.log("隊列已滿，等待");
wait();
}
Console.log("隊列非滿，開始寫入");
doPut(value);
//通知阻塞線程消費
notifyAll();
}
public synchronized int take() throws InterruptedException {
if (isEmpty()) {
Console.log("隊列已空，等待");
wait();
}
int value = doTake();
//通知阻塞線程寫入
notifyAll();
return value;
}
}對應(yīng)的筆者以線程調(diào)試模式給出下面這段代碼，在首先讓線程1執(zhí)行兩次寫操作,查看是否在第二次阻塞是否會在消費者線程消費后存入，所以筆者也會在兩個線程執(zhí)行完畢后，判斷隊列非空來查看是否實現(xiàn)這一點：
//創(chuàng)建一個容量為1的緩沖區(qū)
BoundedBuffer boundedBuffer = new BoundedBuffer(1);
CountDownLatch countDownLatch = new CountDownLatch(2);
//啟動寫入線程第一次寫入成功，第二次寫入阻塞，直到消費者線程完成消費
new Thread(() -> {
try {
boundedBuffer.put(1);
boundedBuffer.put(2);
countDownLatch.countDown();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}).start();
new Thread(() -> {
try {
ThreadUtil.sleep(1000);
Console.log("take:{}", boundedBuffer.take());
countDownLatch.countDown();
} catch (InterruptedException e) {
throw new RuntimeException();
}
}).start();
try {
countDownLatch.await();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
//通過非空函數(shù)判斷線程1第二個元素是否存成功
Console.log("main線程結(jié)束：{}", boundedBuffer.isEmpty());對應(yīng)輸出結(jié)果如下，可以看到第二次寫入因為隊列滿而阻塞，一旦消費者完成消費后，生產(chǎn)者就立刻被喚醒寫入：
二、關(guān)于條件謂詞的一些探討
1. 條件謂詞的使用方式
要想正確的使用條件隊列，就需要正確的抓住線程與條件謂語之間的關(guān)聯(lián)，保證合適的條件下當(dāng)線程添加至條件隊列，并在合適的時機(jī)將其喚醒，以我們的本文一直在強(qiáng)調(diào)的有界隊列：
- 對于put方法來說：只有條件非滿的情況下，才能添加元素至隊列
- 對于take方法來說，只有條件非空的情況下，才能取出元素
同時，每一次wait的調(diào)用都會將調(diào)用者隱式的和條件隊列加以關(guān)聯(lián)，例如:
- 調(diào)用有界緩存的take方法時，若沒有元素，當(dāng)前線程調(diào)用wait阻塞存入監(jiān)視鎖底層的waitSet
- 調(diào)用有界緩存put方法時，若空間已滿，當(dāng)前線程調(diào)用wait存入監(jiān)視鎖底層的waitset
當(dāng)然這一切的都有一個前提，即調(diào)用者已經(jīng)獲取到的當(dāng)前wait方法對應(yīng)的對象的監(jiān)視鎖，這是并發(fā)互斥中等待通知模型有序協(xié)調(diào)的一個必要條件：
2. 過早的喚醒或錯誤喚醒
對條件隊列有了基本的概念之后，我們再來更進(jìn)一步的探討這套設(shè)計理念，實際上按照目前的設(shè)計來看，這套等待喚醒模型還是存在一定的缺陷，即多條件關(guān)聯(lián)單監(jiān)視鎖導(dǎo)致的錯誤喚醒問題。
舉個例子，假設(shè)基于我們要上述的有界緩存隊列，我們打算增加一個關(guān)閉有界緩存的操作，即直接起一個線程查看shutdownFlag如果為false則掛起等待，當(dāng)其他線程將shutdownFlag設(shè)置為true的時候?qū)⑵鋯拘?對應(yīng)的我們也給出下面這樣一段代碼：
public synchronized void shutdown() {
isShuttingDown = true;
notifyAll();
}
private volatile boolean isShuttingDown = false;
public synchronized void shutdownIfInNeed() throws InterruptedException {
if (isShuttingDown == false) {
wait();
Console.log("關(guān)閉線程被喚醒");
}
//執(zhí)行阻塞隊列中斷和關(guān)閉所有線程的操作
//......
}對此我們試想這樣一個情況，我們現(xiàn)在有一個上界為1的有界隊列，對應(yīng)3個線程按如下順序執(zhí)行：
- 消費者線程嘗試從有界緩存獲取元素，阻塞等待喚醒
- 停止線程發(fā)現(xiàn)停止標(biāo)識為false，阻塞等待喚醒
- 生產(chǎn)者線程存入元素，隊列有新元素，調(diào)用notifyall通知消費者消費
重點來了，停止線程和消費者線程都處于當(dāng)前監(jiān)視鎖的等待隊列中，所以notifyall操作可能會誤喚醒停止線程將隊列消費和所有線程中斷造成系統(tǒng)崩潰。
除此之外處于wait的線程還可能會被錯誤的喚醒即沒有任何征兆的情況下蘇醒被CPU時間片執(zhí)行，引用《java并發(fā)編程實戰(zhàn)中》的說法：
以 “早餐” 烤面包機(jī)烤面包完成后通知人們食用為例 ， 這就好?烤?包機(jī)的線 路 連 接 有 問 題 ， 有時候當(dāng)?包還未烤 時 ， 鈴聲 就 響起來了
對應(yīng)的我們也給出這個案例的代碼：
public static void main(String[] args) {
//創(chuàng)建一個容量為1的緩沖區(qū)
BoundedBuffer boundedBuffer = new BoundedBuffer(1);
CountDownLatch countDownLatch = new CountDownLatch(2);
new Thread(() -> {
try {
//線程0取元素阻塞
Console.log("take:{}", boundedBuffer.take());
countDownLatch.countDown();
} catch (InterruptedException e) {
throw new RuntimeException();
}
}, "t0").start();
new Thread(() -> {
try {
//線程1查看停止信號為false阻塞
boundedBuffer.shutdownIfInNeed();
countDownLatch.countDown();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}, "t1").start();
new Thread(() -> {
try {
//線程2put操作隊列非空執(zhí)行通知操作，導(dǎo)致停止線程被錯誤的喚醒
boundedBuffer.put(1);
countDownLatch.countDown();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}, "t2").start();
try {
countDownLatch.await();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
Console.log("main線程結(jié)束：{}", boundedBuffer.size());
}輸出結(jié)果如下，可以看到在生產(chǎn)者生產(chǎn)元素后的通知動作，把關(guān)閉線程給喚醒了，這就是經(jīng)典的錯誤喚醒：
隊列已空，take線程:t0等待
隊列非滿，開始寫入
關(guān)閉線程被喚醒
take線程 t0被喚醒
take:-1
main線程結(jié)束：1本質(zhì)原因就是一個監(jiān)視鎖中的隊列關(guān)聯(lián)多個條件，使得在多條件的等待通知場景下存在錯誤通知的情況，考慮到這一點，無論是對于put、take還是shutdown方法，我們都需要進(jìn)行改進(jìn)，確保：
- 生產(chǎn)者被喚醒后，進(jìn)行必要的非滿檢查，且只有將空隊列存入元素后通知消費者
- 消費者被喚醒后，進(jìn)行必要的非空檢查，只有將非空隊列消費空之后，通知生產(chǎn)者
- shutdown線程被喚醒后，進(jìn)行必要的狀態(tài)標(biāo)識檢查，只有狀態(tài)標(biāo)識為true才能停止線程
改進(jìn)后的代碼如下所示，可以看到筆者將if條件判斷后wait的操作改為while+wait操作確保喚醒后的再確認(rèn)：
public synchronized void put(int value) throws InterruptedException {
while (isFull()) {//條件觸發(fā)時循環(huán)檢測一下
wait();
}
//空變?yōu)榉强? boolean wasEmpty = isEmpty();
doPut(value);
if (wasEmpty) {//僅當(dāng)空變?yōu)榉强諘r才通知
notifyAll();
}
}
public synchronized int take() throws InterruptedException {
while (isEmpty()) {
wait();
}
//滿變?yōu)榉菨M才通知
boolean wasFull = isFull();
int value = doTake();
if (wasFull) {
notifyAll();
}
return value;
}
public synchronized void shutdownIfInNeed() throws InterruptedException {
while (isShuttingDown == false) {
wait();
Console.log("關(guān)閉線程被喚醒");
}
//執(zhí)行阻塞隊列中斷和關(guān)閉所有線程的操作
//......
}3. notify下的信號丟失問題
我們再來說說通知的哲學(xué)，剛接觸java這門語言的時候，都會了解到notify和notifyAll的區(qū)別，這一點我們也可以直接從源碼的注釋上了解這一點，即前者僅僅通知監(jiān)視鎖下的單個線程而后者則是所有線程：
1. notify:Wakes up a single thread that is waiting on this object's monitor.
2. notifyAll:Wakes up all threads that are waiting on this object's monitor. A thread waits on an object's monitor by calling one of the wait methods.所以這也就是為什么筆者在實現(xiàn)上述通知這個動作的時候，使用的是notifyAll而非notify,即notify存在信號丟失問題，還是用我上述的生產(chǎn)者-消費者和異步關(guān)閉線程的例子，試想下述場景：
- 有界隊列元素空間為1
- 線程1取元素為空，阻塞
- 線程2查看停止標(biāo)識為false，阻塞
- 線程0添加元素，元素非空，notify選中了線程2
- 本該處理元素的線程1因為沒收到通知，造成了一種信號丟失的情況
這本質(zhì)就是同步鎖和wait以及條件謂語上一種設(shè)計缺陷，即一個同步鎖只能關(guān)聯(lián)一組條件隊列，而條件隊列無法做區(qū)分。
所以基于上述條件隊列的案例，我們通過條件通知的方式進(jìn)行比對保證更高效的準(zhǔn)確的通知，避免每次操作之后都非常激進(jìn)的通知所有線程造成非必要的上下文切換開銷，當(dāng)然讀者在進(jìn)行這樣的優(yōu)化時務(wù)必記得，只有保證程序可以使用的情況下，在進(jìn)行優(yōu)化的哲學(xué)：
4. 基于條件變量下的等待通知模型
內(nèi)置隊列存在一個內(nèi)置鎖關(guān)聯(lián)多個條件隊列的情況，這使得很多線程被錯誤的喚醒，導(dǎo)致非必要的CPU時鐘消耗和上下文切換和并發(fā)競爭鎖的開銷。針對上述的問題，我們必須到借由一種工具保證同一把鎖下的各個條件的線程都放置到不同的隊列中，從而保證正確的喚醒，即：
等待隊列非滿的生產(chǎn)者線程存到一個隊列，待消費者完成元素消費后通知這個隊列
- 等待隊列非空的消費者線程存到一個等待隊列，待生產(chǎn)者完成元素投遞后通知這個隊列
所以，通過juc包下的鎖即可實現(xiàn)將條件放到不同的條件隊列中，同時它還能可以實現(xiàn)隊列內(nèi)部公平的喚醒，以保證等待喚醒的是需要的線程從而從而等到通知的高效，以及減小非必要的上下文切換的開銷：
public class ConditionBoundedBuffer<V> {
private final V[] items;
private int head;
private int tail;
private int count;
//下述兩個條件隊列關(guān)聯(lián)同一把鎖，線程按照各自條件與隊列關(guān)聯(lián)
private final ReentrantLock lock = new ReentrantLock();
//生產(chǎn)者等待隊列非滿的等待隊列
private final Condition notFull = lock.newCondition();
//消費者等待隊列非空的等待隊列
private final Condition notEmpty = lock.newCondition();
public ConditionBoundedBuffer(int capacity) {
this.items = (V[]) new Object[capacity];
}
public boolean isFull() {
return count == items.length;
}
public boolean isEmpty() {
return count == 0;
}
public void put(V v) throws InterruptedException {
lock.lock();
try {
while (isFull()) {//輪詢檢測非滿
notFull.await();
}
//添加元素
items[tail++] = v;
count++;
if (tail == items.length) {
tail = 0;
}
notEmpty.signal();
} finally {
lock.unlock();
}
}
public V take() throws InterruptedException {
lock.lock();
try {
while (isEmpty()) {//輪詢檢測非空
notEmpty.await();
}
//消費元素
V v = items[head];
items[head] = null;
head++;
count--;
if (head == items.length) {
head = 0;
}
notFull.signal();
return v;
} finally {
lock.unlock();
}
}
}對應(yīng)的我們也給出壓測代碼，最終斷言也是正確的：
ConditionBoundedBuffer<Integer> conditionBoundedBuffer = new ConditionBoundedBuffer<>(1);
ExecutorService threadPool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() + 1);
for (int i = 0; i < 100_0000; i++) {
//提交1一個元素
threadPool.execute(() -> {
try {
conditionBoundedBuffer.put(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
//消費一個元素
threadPool.execute(() -> {
try {
conditionBoundedBuffer.take();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
threadPool.shutdown();
while (!threadPool.isTerminated()) {
}
//判斷并發(fā)下線程是否正確的對等生產(chǎn)和消費
Assert.equals(conditionBoundedBuffer.count, 0);三、小結(jié)
自此我們針對并發(fā)編程中的等待通知模型中的狀態(tài)管理，等待通知原則和技巧進(jìn)行了深入的分析和演示，希望對你有幫助。
