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