C并發(fā)系列寫到第四篇終于輪到 condition_variable 這個在生產(chǎn)者消費者、任務(wù)隊列、線程池里出鏡率極高的同步原語。前面幾篇我們把線程創(chuàng)建、mutex 互斥、原子操作都過了一遍但很多人寫完這些還是會卡殼鎖能保證“同一時間只有一個人進”卻沒法解決“我什么時候可以干活”這種問題。你總不能讓線程死循環(huán)去搶鎖吧那 CPU 早就燒沒了。condition_variable 就是來解決這個“等待-通知”問題的。這篇文章我盡量不繞彎子先把機制講透再給一個可以完整編譯運行的案例最后用 Java 的 wait/notify、Lock 和 BlockingQueue 做對照幫你在面試和實際開發(fā)里都能把兩邊的知識串起來。1. 從本質(zhì)講起condition_variable 到底解決了什么問題1.1 沒有通知機制的世界輪詢與延遲陷阱先看一個非常常見的場景。假設(shè)有一個全局的任務(wù)隊列生產(chǎn)線程往里塞任務(wù)消費線程從里面取任務(wù)。在沒有 condition_variable 之前最原始的寫法是讓消費線程不停地循環(huán)std::mutex mtx; std::dequeint tasks; void consumer() { while (true) { std::lock_guardstd::mutex lock(mtx); if (!tasks.empty()) { int task tasks.front(); tasks.pop_front(); // 處理任務(wù) } } }這個代碼邏輯上沒錯但有兩個無法接受的問題。第一CPU 會被這個空轉(zhuǎn)的循環(huán)占滿一個消費者線程就能把一個核燒到 100%而且它什么都沒干純粹在檢查隊列。第二如果你為了讓它在沒任務(wù)時休息一下在空隊列分支里加一個std::this_thread::sleep_for(10ms)那任務(wù)的延遲就完全取決于這個 sleep 的間隔。想去掉延遲就得縮短 sleep想省 CPU 就得加長 sleep兩頭都難受。我在實際項目里見過類似的代碼一個消費者線程占滿了一整顆核排查了半天才發(fā)現(xiàn)是這種輪詢寫法改成條件變量之后 CPU 直接降到接近 0。1.2 條件變量解決的兩個核心問題condition_variable 要做的事情其實就兩件。第一把“檢查條件”和“進入等待”變成一個不可分割的整體。如果先檢查條件再等待中間可能被別人插一腳導(dǎo)致通知丟在“檢查完”和“開始等”的縫隙里這也就是所謂的 lost wakeup丟失喚醒。條件變量通過和 mutex 配合把這兩個動作綁定到 wait 函數(shù)內(nèi)部從根源上堵住了這個縫隙。第二等待的時候真正掛起線程讓出 CPU直到別人通知它再醒來。這就像你去餐廳吃飯沒有叫號系統(tǒng)的時候你只能每隔幾分鐘跑去問服務(wù)員“有空位了嗎”既累又煩有了叫號系統(tǒng)你可以在等候區(qū)安心休息服務(wù)員喊到你的時候再起身進去。mutex 是門鎖condition_variable 就是那個叫號器兩者協(xié)同工作數(shù)據(jù)本身由 mutex 保護數(shù)據(jù)狀態(tài)的變化由 condition_variable 通知。2. 核心 API 與機制剖析wait / notify 到底怎么配合2.1 wait 系列自動釋放鎖與被喚醒后的鎖回先看最簡單的wait(lock)這個接口的調(diào)用前提是當(dāng)前線程已經(jīng)持有了傳入的鎖。進入 wait 之后它會原子地完成兩件事把當(dāng)前線程放進等待隊列然后釋放掉這個鎖。為什么要釋放因為當(dāng)前線程要睡覺了不可能攥著鎖不放手否則其他線程無法修改共享數(shù)據(jù)也就沒人能來叫醒它。當(dāng)收到 notify 通知并且當(dāng)前線程成功搶到鎖之后wait 才會返回。這里有一個特別重要的點wait 返回的時候條件并不一定成立?？赡苡袃蓚€消費者同時被喚醒其中一個先搶到鎖把隊列里的任務(wù)取走了另一個搶到鎖后發(fā)現(xiàn)隊列又空了。所以標(biāo)準(zhǔn)的用法永遠是循環(huán)檢查條件而不是用 ifstd::unique_lockstd::mutex lock(mtx); while (tasks.empty()) { cv.wait(lock); } // 到這里 tasks 一定不為空這個 while 循環(huán)寫多了之后C 直接在 overload 版本里幫你封裝了它。下面這兩種寫法是完全等價的// 寫法一手動 while wait while (tasks.empty()) { cv.wait(lock); } // 寫法二帶 predicate 的 wait cv.wait(lock, []() { return !tasks.empty(); });帶 predicate 的版本是我日常用得最多的因為它把“條件不滿足就繼續(xù)等”這個意圖表達得非常清楚也避免了忘記寫 while 的低級錯誤。它內(nèi)部就是 while (!pred()) wait(lock) 的簡寫。超時接口需要單獨說。wait_for和wait_until都有兩個版本帶 predicate 和不帶 predicate 的// 不帶 predicate 版本返回 std::cv_status std::cv_status status cv.wait_for(lock, std::chrono::milliseconds(100)); if (status std::cv_status::timeout) { // 超時 } else { // 被喚醒 } // 帶 predicate 版本返回 bool bool ready cv.wait_for(lock, std::chrono::milliseconds(100), []() { return flag; }); if (ready) { // 條件成立 } else { // 超時 }帶 predicate 的版本返回值就是條件是否成立用起來最省心。手動版本要注意返回timeout不能等同于“條件不成立”因為完全有可能在超時的那一瞬間恰好有人 notify這時候你去讀條件可能已經(jīng)成立了。所以無論哪種寫法超時返回之后都必須再檢查一次實際條件這是很多 bug 的源頭。2.2 notify_one 與 notify_all 的選型和細節(jié)notify_one負(fù)責(zé)喚醒等待隊列里的一個線程notify_all負(fù)責(zé)喚醒所有線程。選型不能只看等待線程數(shù)量更要看這次通知之后被喚醒的線程是不是都能繼續(xù)干活。如果隊列里只有一條任務(wù)你喚醒所有消費者它們會同時醒來搶鎖最后只有一個能拿到任務(wù)其他幾個只能繼續(xù)回去睡。這個現(xiàn)象有點像驚群效應(yīng)白白增加調(diào)度開銷。所以單任務(wù)場景用notify_one就夠了。反過來如果是廣播型事件比如服務(wù)器要關(guān)閉了、任務(wù)隊列要清空了這種所有等待線程都需要感知的狀態(tài)變化必須用notify_all否則只喚醒一個其他線程永遠不知道發(fā)生了改變。還有一個經(jīng)驗是 notify 的時機問題。調(diào)用 notify 的時候不一定非要持有鎖相反我建議先把鎖釋放掉再通知。我給你看一個細節(jié)對比// 持鎖通知 { std::lock_guardstd::mutex lock(mtx); tasks.push_back(1); cv.notify_one(); // 被喚醒的線程想去搶鎖但鎖還沒釋放 } // 釋放后再通知 { std::unique_lockstd::mutex lock(mtx); tasks.push_back(1); lock.unlock(); cv.notify_one(); // 被喚醒的線程一醒來就能拿到鎖 }第二種寫法能讓被喚醒的線程立刻拿到鎖減少“喚醒了又阻塞在鎖上”的來回抖動。有些場景里這個細節(jié)對吞吐量有明顯影響后面工程實踐部分我再細說。2.3 為什么 wait 必須和同一把 mutex 配合條件變量和 mutex 的關(guān)系經(jīng)常讓人困惑我當(dāng)年也迷過一陣。其實核心是為了保證“檢查-等待”的原子性。設(shè)想一種錯誤的寫法std::mutex mtx; std::condition_variable cv; bool ready false; // 線程 A生產(chǎn)者 { std::lock_guardstd::mutex lock(mtx); ready true; } cv.notify_one(); // 線程 B消費者 if (!ready) { // 在鎖外檢查 cv.wait(lock); // 這里會一直等下去 }問題出在線程 B 檢查 ready 和調(diào)用 wait 之間不是原子的。如果線程 A 在 B 檢查完 ready此時還是 false之后、B 調(diào)用 wait 之前把 ready 改成 true 并且調(diào)用了 notify那么這次通知就發(fā)生在 B 正式進入等待之前直接丟了。從此 B 就永遠等在那里即使 ready 已經(jīng)是 true。正確做法是讓條件檢查和 wait 在同一個鎖的保護下并且直接使用帶 predicate 的重載。因為 wait 內(nèi)部會在持有鎖的情況下檢查 predicate不滿足才原子地釋放鎖并進入等待這樣線程 A 即使先改了狀態(tài)再通知B 進入 wait 時也會先看一眼 predicate發(fā)現(xiàn)條件已經(jīng)滿足就不會真的睡過去。這就是 condition_variable 為什么必須和同一把 mutex 綁在一起的根本原因。3. 完整案例用 condition_variable 實現(xiàn)一個帶緩沖的任務(wù)隊列3.1 場景與設(shè)計哪些條件需要等待這次寫一個經(jīng)典但足夠完整的例子一個容量有限的任務(wù)隊列多個生產(chǎn)者往里放任務(wù)多個消費者從里面取任務(wù)。這個隊列有兩個需要等待的場景隊列為空時消費者不能取需要等“隊列非空”的條件。隊列滿時生產(chǎn)者不能放需要等“隊列有空位”的條件。所以最自然的設(shè)計是使用兩個 condition_variable一個管“非空”一個管“非滿”。Java 的ArrayBlockingQueue內(nèi)部就是這種雙 Condition 結(jié)構(gòu)我們這里用 C 手寫一遍你會發(fā)現(xiàn)兩邊幾乎是一一對應(yīng)的。3.2 完整代碼與逐段解釋完整代碼如下可以直接編譯運行#include condition_variable #include deque #include iostream #include mutex #include thread #include chrono class BlockingQueue { public: explicit BlockingQueue(size_t maxSize) : maxSize_(maxSize) {} void push(int val) { std::unique_lockstd::mutex lock(mtx_); notFull_.wait(lock, []() { return queue_.size() maxSize_; }); queue_.push_back(val); std::cout push val , queue size queue_.size() std::endl; // 釋放鎖之后再通知消費者減少鎖競爭 lock.unlock(); notEmpty_.notify_one(); } int pop() { std::unique_lockstd::mutex lock(mtx_); notEmpty_.wait(lock, []() { return !queue_.empty(); }); int val queue_.front(); queue_.pop_front(); std::cout pop val , queue size queue_.size() std::endl; lock.unlock(); notFull_.notify_one(); return val; } private: std::mutex mtx_; std::condition_variable notEmpty_; std::condition_variable notFull_; std::dequeint queue_; size_t maxSize_; }; int main() { BlockingQueue queue(2); std::thread consumer1([]() { for (int i 0; i 5; i) { queue.pop(); std::this_thread::sleep_for(std::chrono::milliseconds(20)); } }); std::thread consumer2([]() { for (int i 0; i 5; i) { queue.pop(); std::this_thread::sleep_for(std::chrono::milliseconds(30)); } }); for (int i 0; i 10; i) { queue.push(i); std::this_thread::sleep_for(std::chrono::milliseconds(50)); } consumer1.join(); consumer2.join(); return 0; }逐個拆解關(guān)鍵段落。push里第一步是加鎖然后調(diào)用notFull_.wait(lock, ...)。這個 wait 有兩個作用如果隊列已經(jīng)滿了當(dāng)前生產(chǎn)線程會阻塞在這里直到消費者取走數(shù)據(jù)后調(diào)用notFull_.notify_one()如果隊列沒滿wait 會立即返回繼續(xù)往下執(zhí)行。wait 內(nèi)部在阻塞期間會自動釋放鎖這使得消費者在隊列滿時依然能夠進入pop并取走數(shù)據(jù)。入隊之后我特意先lock.unlock()再notEmpty_.notify_one()。這么做的好處前面說過被喚醒的消費者可以立刻獲得鎖不必等待生產(chǎn)者在作用域末尾釋放。注意unique_lock不像lock_guard那樣析構(gòu)時才解鎖它允許你手動控制解鎖時機這就是我在這里用它的原因。pop的邏輯是對稱的。消費者在notEmpty_.wait上等待隊列為空時掛起彈出數(shù)據(jù)后手動解鎖然后notFull_.notify_one()喚醒一個等待中的生產(chǎn)者。3.3 運行結(jié)果觀察與參數(shù)調(diào)整建議正常運行時會看到生產(chǎn)者輸出幾條 “push” 之后消費者開始輸出 “pop”期間隊列大小在 0 到 2 之間波動。因為消費線程啟動后就會立刻嘗試pop而隊列初始是空的所以兩個消費者都會先阻塞在notEmpty_.wait。生產(chǎn)者每隔 50ms 推入一條消費者則按照各自的節(jié)奏取走到了第 5、6 條左右因為 maxSize 是 2生產(chǎn)者可能會被notFull_.wait卡住直到消費者取走數(shù)據(jù)騰出空位。我建議動手改幾個參數(shù)觀察行為變化。比如把 maxSize 改成 1整個隊列就退化為一個“槽位”生產(chǎn)者和消費者必須嚴(yán)格交替執(zhí)行你能看到非常清晰的阻塞-喚醒過程。再比如把消費者從這個改成 3 個或者更多看看notify_one是否會導(dǎo)致某些消費者長期得不到任務(wù)。這種微調(diào)比看任何理論講解都更能理解條件變量的行為。這里有一個小提醒代碼里的std::cout本身不是線程安全的但這個示例里每個輸出都發(fā)生在隊列鎖釋放之前所以從共享數(shù)據(jù)上講是安全的。實際項目中如果日志系統(tǒng)比較復(fù)雜建議給日志單獨加鎖或者用線程安全的日志庫不要順手往業(yè)務(wù)鎖里塞日志輸出。4. Java 對比視角從 Object.wait 到 Lock 再到 BlockingQueue4.1 Object.wait/notify 與 condition_variable 的等價物Java 里每個對象都可以作為鎖和等待集合這是和 C 一個很大的思維差異。synchronized(lock)代碼塊里你可以調(diào)用lock.wait()讓當(dāng)前線程釋放 monitor 并掛起其他線程持鎖時調(diào)用lock.notify()喚醒一個等待者。邏輯上這幾乎就是 condition_variable 的 Object 內(nèi)建版本// 消費者 synchronized (lock) { while (queue.isEmpty()) { lock.wait(); } int val queue.removeFirst(); } // 生產(chǎn)者 synchronized (lock) { queue.addLast(val); lock.notify(); }注意 Java 里wait()必須在synchronized塊內(nèi)調(diào)用這對應(yīng) C 的“wait 時當(dāng)前線程必須持有鎖”。Java 也必須使用 while 循環(huán)重新檢查條件原因和 C 完全一樣虛假喚醒和競爭喚醒在 Java 里一樣存在。最大的差異是 Java 的wait()聲明會拋出InterruptedException所以要么在方法簽名里加上throws要么用 try/catch 包起來。C 的線程模型沒有這種中斷機制這是兩種語言設(shè)計取向的不同不是簡單的誰好誰壞。4.2 Condition 接口結(jié)構(gòu)和 API 幾乎一致的對應(yīng)如果你用過 Java 的ReentrantLock會發(fā)現(xiàn)它提供的Condition接口和 C 的 condition_variable 在結(jié)構(gòu)上幾乎一一對應(yīng)ReentrantLock lock new ReentrantLock(); Condition notEmpty lock.newCondition(); Condition notFull lock.newCondition(); // 生產(chǎn)者 lock.lock(); try { while (count items.length) { notFull.await(); } // 寫入數(shù)組 notEmpty.signal(); } finally { lock.unlock(); }對應(yīng)的 C 結(jié)構(gòu)std::mutex mtx; std::condition_variable notEmpty; std::condition_variable notFull; // 生產(chǎn)者 std::unique_lockstd::mutex lock(mtx); notFull.wait(lock, []() { return count items.length; }); // 寫入數(shù)據(jù) notEmpty.notify_one();這里面的對應(yīng)關(guān)系非常清晰wait對應(yīng)awaitnotify_one對應(yīng)signalnotify_all對應(yīng)signalAll。Condition的優(yōu)勢在于它允許你在同一個鎖上創(chuàng)建多個獨立的等待集合這正好對應(yīng) C 里用多個 condition_variable 配合一個 mutex 的做法。我建議用一張表把這些對應(yīng)關(guān)系記下來面試被問到“C 和 Java 的并發(fā)原語怎么對應(yīng)”時可以直接拿出來用功能CJava ObjectJava Condition鎖std::mutex / unique_locksynchronizedReentrantLock等待集合對象condition_variable對象自身Condition等待wait(lock, pred)wait()await()喚醒單個notify_one()notify()signal()喚醒全部notify_all()notifyAll()signalAll()超時等待wait_for / wait_untilwait(timeout)await(timeout)中斷支持無InterruptedExceptionInterruptedException4.3 BlockingQueueJava 工程中的封裝替代實際寫 Java 生產(chǎn)代碼的時候我?guī)缀醪粫謱?Condition 去做生產(chǎn)者消費者直接用ArrayBlockingQueue或者LinkedBlockingQueue就結(jié)束了。這不是因為 Java 開發(fā)者比 C 開發(fā)者懶而是 JDK 已經(jīng)把“wait while signal”這套底層邏輯封裝好了還考慮到了公平性、超時、中斷等一堆細節(jié)。ArrayBlockingQueue內(nèi)部正是用兩個 ConditionnotEmpty和notFull實現(xiàn)的和上面 C 版本的思路完全一樣只是平時你不用自己寫而已。C 標(biāo)準(zhǔn)庫里沒有等價的阻塞隊列容器所以我們要自己封裝。這算是語言生態(tài)上的差異不是能力上的差異。理解這個背景之后你在面試時就可以這樣回答“Java 的 BlockingQueue 提高了并發(fā)編程的上層抽象C 則把這些控制權(quán)留給了開發(fā)者各有取舍?！边@個回答既展示了底層的理解又體現(xiàn)了對大廠封裝程度的認(rèn)知。4.4 兩邊在工程思維上的取舍對比除了 API 層面的差別兩邊在工程思維上也有值得注意的差異。Java 的ReentrantLock支持公平鎖、可中斷鎖等待、多個 Condition 綁定同一個鎖語言層面提供了更多托管運行時的便利。C 則講究零開銷抽象沒有 GC 兜底也沒有語言級別的中斷所以寫 C 并發(fā)時對資源管理和生命周期要更敏感。比如 C 里 condition_variable 不能拷貝要小心和對象生命周期綁定Java 里這些內(nèi)存管理問題被 JVM 接管了。不過在這些底層語義上兩者遵循的是同一套并發(fā)理論。虛假喚醒、丟失喚醒、線程競爭這些概念在兩邊都存在解決問題的思路也一致條件檢查必須用 while條件狀態(tài)的變化必須由鎖保護喚醒必須發(fā)生在狀態(tài)變化之后。能把 C 的機制理解透徹再去看 Java 的封裝幾乎是一馬平川反過來從 Java 的高層抽象出發(fā)也能幫你理解 C 底層為什么要提供這些原語。5. 工程實踐中常見的坑與排查技巧實錄5.1 丟失喚醒最隱蔽也最致命的坑丟失喚醒大概是條件變量領(lǐng)域最臭名昭著的問題。它難排查因為它不是每次都發(fā)生往往取決于線程調(diào)度的時序。我在上文講過如果“檢查條件”和“進入等待”不是原子的通知就可能落在兩者之間導(dǎo)致等待方永遠睡過頭。用 predicate 重載能解決這個問題因為 wait 內(nèi)部把“檢查條件 決定是否等待”綁定成了一個原子操作。這里有一個我的切身體會。之前維護一個老項目同事在代碼里用條件變量做緩存刷新刷新線程發(fā)現(xiàn)緩存過期后不是先置一個標(biāo)志位再通知而是直接 notify。等待線程醒來后再次檢查標(biāo)志位發(fā)現(xiàn)沒變又繼續(xù)睡。這個 bug 在測試環(huán)境整整兩天才復(fù)現(xiàn)一次后來通過打印日志才發(fā)現(xiàn)喚醒比狀態(tài)變更提前了。正確的順序永遠是先修改受鎖保護的條件再釋放鎖最后 notify。順序錯了代碼再漂亮也是定時炸彈。5.2 虛假喚醒與超時返回后的二次判斷標(biāo)準(zhǔn)庫文檔明確說 spurious wakeup 是合法的也就是說線程可能在沒有任何人調(diào)用 notify 的情況下自己醒來。操作系統(tǒng)層面很少見但你不能賭它不發(fā)生。C 的 predicate 重載和 while 循環(huán)自動幫你處理了這種情況所以只要你堅持用這兩種寫法虛假喚醒基本不用操心。麻煩的是超時。很多人寫超時邏輯時會這樣std::unique_lockstd::mutex lock(mtx); if (cv.wait_for(lock, std::chrono::seconds(1)) std::cv_status::timeout) { // 認(rèn)為條件不成立 } else { // 認(rèn)為條件成立 }這個寫法有隱患。wait_for 返回timeout只能說明超時了不能說明條件一定不成立返回no_timeout也不能保證條件一定成立因為可能是虛假喚醒。最穩(wěn)妥的寫法是帶 predicate 的重載bool success cv.wait_for(lock, std::chrono::seconds(1), []() { return flag; }); if (success) { // 條件成立 } else { // 超時或條件始終未成立 }這個版本直接給出“條件是否成立”的結(jié)論省去了手動二次判斷的麻煩。我用這個 API 之后超時相關(guān)的邏輯 bug 少了很多。5.3 持鎖 notify 帶來的喚醒抖動我在第 2 節(jié)提到過持鎖 notify 的問題這里展開講一下。假設(shè)你在lock_guard保護的作用域內(nèi)調(diào)用notify_one被喚醒的線程會立刻嘗試獲取同一把鎖。但此時鎖還在通知方手里要等lock_guard析構(gòu)才能釋放。于是被喚醒線程剛被喚起來馬上又因為搶鎖失敗而阻塞回去白白消耗一次調(diào)度切換。我實際測過一個簡化版的任務(wù)隊列持鎖 notify 和解鎖后 notify 的吞吐量有百分之幾的差別。在隊列本身非常短、競爭激烈的時候這個差距會更明顯。所以我的習(xí)慣是如果條件變量保護的臨界區(qū)很短就在臨界區(qū)外通知如果臨界區(qū)很長這個優(yōu)化就更值得做。C 的unique_lock可以手動 unlock很靈活Java 里由于 try-finally 釋放鎖一般就直接在鎖內(nèi) signal 了這算是工程習(xí)慣上的一個小差異。5.4 一次實戰(zhàn)排查線程卡死的定位思路最后分享一個我實際排查過的卡死問題。當(dāng)時一個監(jiān)控采集服務(wù)里用條件變量通知消費者處理過期數(shù)據(jù)線上出現(xiàn)消費者線程不工作的現(xiàn)象。第一反應(yīng)是懷疑丟失喚醒于是加日志準(zhǔn)備抓時序。用 gdb 掛上進程執(zhí)行thread apply all bt看所有線程棧發(fā)現(xiàn)消費者線程確實阻塞在cv.wait上但生產(chǎn)者線程并沒有死掉它正卡在一個網(wǎng)絡(luò)請求的超時等待里。也就是說條件變量本身沒有任何問題是生產(chǎn)者因為外部依賴變慢遲遲沒有產(chǎn)生新數(shù)據(jù)消費者才一直空等。這個案例讓我意識到排查條件變量問題時不要只盯著條件變量本身還要把整條數(shù)據(jù)鏈路看清楚。gdb 看線程棧是最直接有效的手段其次是在 wait 前后加帶時間戳的日志確認(rèn)到底是“沒收到通知”還是“收到了通知但條件不滿足”。5.5 我寫條件變量時固定檢查的三個問題經(jīng)過前面這些坑我現(xiàn)在每寫一段條件變量代碼都會在心里過三個問題第一所有對條件狀態(tài)的讀寫是否都在同一把鎖的保護下第二被喚醒之后是否重新檢查了條件而不是直接假設(shè)條件成立第三通知方是不是在狀態(tài)修改完成并且釋放鎖之后才調(diào)用 notify如果三個問題的答案都是肯定的這段代碼基本不會再出幺蛾子。另外我還會順手確認(rèn) close/stop 這類廣播事件用的是notify_all而不是notify_one否則十有八九會漏掉某個等待線程。從 C 的 condition_variable 到 Java 的 Object.wait 和 Condition再到成熟的 BlockingQueue 封裝你會發(fā)現(xiàn)并發(fā)編程的核心問題其實是相通的怎么讓線程在合適的時機睡下又怎么在合適的時機醒來。把 wait/notify 這套機制徹底理解透了后面看什么語言的高并發(fā)代碼都會順很多。