這篇解決一個(gè)問(wèn)題設(shè)備軟件里業(yè)務(wù)線程在跑同時(shí)要寫(xiě)日志、存數(shù)據(jù)庫(kù)、存圖像、上報(bào) SECS。這些 IO 操作慢直接在業(yè)務(wù)線程里做會(huì)卡住狀態(tài)機(jī)。怎么寫(xiě)才能讓業(yè)務(wù)線程「丟下就走」后臺(tái)慢慢處理。一、先從一個(gè)生活例子說(shuō)起你去銀行辦業(yè)務(wù)柜員辦完你的事后給你一張回執(zhí)單。柜員不會(huì)站在窗口等你把回執(zhí)單收好再叫下一個(gè)號(hào)而是把回執(zhí)單往旁邊一放立刻叫下一個(gè)人。旁邊那堆回執(zhí)單就是「隊(duì)列」。柜員是「生產(chǎn)者」不斷產(chǎn)生回執(zhí)單。你客戶是「消費(fèi)者」自己拿走。柜員不等消費(fèi)者消費(fèi)者也不影響柜員。如果柜員非要等你收好回執(zhí)單再叫下一個(gè)號(hào)那每個(gè)客戶多花 10 秒一天少辦一半業(yè)務(wù)。這就是生產(chǎn)者消費(fèi)者模式的核心生產(chǎn)者只管往隊(duì)列里丟消費(fèi)者只管從隊(duì)列里取兩邊各跑各的互不等待。二、回到代碼不用隊(duì)列會(huì)怎樣設(shè)備軟件里最常見(jiàn)的場(chǎng)景報(bào)警。手臂運(yùn)動(dòng)線程發(fā)現(xiàn)真空失敗要報(bào)警。報(bào)警要做這些事寫(xiě)日志快微秒級(jí)。響蜂鳴快。彈 UI 報(bào)警窗慢可能等用戶點(diǎn)確認(rèn)。寫(xiě)數(shù)據(jù)庫(kù)慢幾十毫秒。上報(bào) SECS慢網(wǎng)絡(luò)通信幾百毫秒。最直覺(jué)的寫(xiě)法業(yè)務(wù)線程里直接做void Arm::OnVacuumFail() { Log(真空失敗); // 快 Beeper::Beep(); // 快 pDlg-ShowAlarm(真空失敗); // 慢等用戶點(diǎn)確認(rèn) db-WriteAlarm(code, msg); // 慢幾十毫秒 secs-Report(msg); // 慢幾百毫秒 }問(wèn)題一堆問(wèn)題一業(yè)務(wù)線程卡住。ShowAlarm等用戶點(diǎn)確認(rèn)可能等幾十秒。這幾十秒里手臂線程不動(dòng)狀態(tài)機(jī)不推進(jìn)其它料堆積。問(wèn)題二數(shù)據(jù)庫(kù)卡住運(yùn)動(dòng)。WriteAlarm幾十毫秒里手臂線程阻塞可能正好該停機(jī)卻停不了。問(wèn)題三SECS 網(wǎng)絡(luò)超時(shí)拖死業(yè)務(wù)。 網(wǎng)絡(luò)不好時(shí)secs-Report可能卡幾秒手臂線程一直等。問(wèn)題四多條報(bào)警同時(shí)來(lái)互相阻塞。 第一條報(bào)警還在彈窗第二條又來(lái)了疊在一起。問(wèn)題五業(yè)務(wù)線程和 UI 線程混在一起。ShowAlarm在運(yùn)動(dòng)線程里調(diào) UI 控件跨線程操作 MFC 控件可能崩。根本問(wèn)題快節(jié)奏的業(yè)務(wù)線程和慢節(jié)奏的 IO 操作攪在一起。三、生產(chǎn)者消費(fèi)者模式怎么解決思路業(yè)務(wù)線程只管「把報(bào)警丟進(jìn)隊(duì)列」一個(gè)專(zhuān)門(mén)的消費(fèi)者線程從隊(duì)列里取出來(lái)慢慢處理。第一步定義隊(duì)列和事件struct AlarmEvent { string time; string code; string message; }; class AlarmQueue { queueAlarmEvent m_queue; mutex m_mtx; condition_variable m_cv; bool m_running true; public: // 生產(chǎn)者調(diào)用丟一個(gè)事件進(jìn)去立刻返回 void Push(AlarmEvent ev) { { lock_guardmutex lk(m_mtx); m_queue.push(ev); } m_cv.notify_one(); // 叫醒消費(fèi)者 } // 消費(fèi)者調(diào)用阻塞等事件 bool Pop(AlarmEvent out) { unique_lockmutex lk(m_mtx); m_cv.wait(lk, [] { return !m_queue.empty() || !m_running; }); if (!m_running m_queue.empty()) return false; out m_queue.front(); m_queue.pop(); return true; } void Stop() { { lock_guardmutex lk(m_mtx); m_running false; } m_cv.notify_all(); } };第二步消費(fèi)者線程AlarmQueue g_alarmQueue; void AlarmConsumerThread() { while (true) { AlarmEvent ev; if (!g_alarmQueue.Pop(ev)) return; // 隊(duì)列停了 // 慢操作在這里做不影響業(yè)務(wù)線程 db-WriteAlarm(ev.code, ev.time, ev.message); secs-Report(ev.message); // UI 操作發(fā)到 UI 線程做 PostMessage(pDlg-m_hWnd, WM_SHOW_ALARM, 0, (LPARAM)new string(ev.message)); } }第三步業(yè)務(wù)線程只 Pushvoid Arm::OnVacuumFail() { Log(真空失敗); // 快操作直接做 Beeper::Beep(); // 快操作直接做 g_alarmQueue.Push({ // 慢操作丟隊(duì)列立刻返回 CurrentTime(), 601_205, 手臂真空失敗 }); // 業(yè)務(wù)線程立刻繼續(xù)跑狀態(tài)機(jī)不等數(shù)據(jù)庫(kù)和 SECS }第四步初始化時(shí)啟動(dòng)消費(fèi)者// 程序啟動(dòng) std::thread(AlarmConsumerThread).detach(); // 程序退出 g_alarmQueue.Stop();關(guān)鍵變化業(yè)務(wù)線程Push后立刻返回不等數(shù)據(jù)庫(kù)、不等 SECS、不等 UI。消費(fèi)者線程慢慢處理處理完一個(gè)取下一個(gè)。業(yè)務(wù)線程和 IO 操作完全解耦互不阻塞。四、生產(chǎn)者消費(fèi)者模式的標(biāo)準(zhǔn)結(jié)構(gòu)角色例子Producer生產(chǎn)者業(yè)務(wù)線程Arm、Channel 等產(chǎn)生事件Consumer消費(fèi)者專(zhuān)門(mén)的處理線程消費(fèi)事件Queue緩沖隊(duì)列AlarmQueue線程安全帶條件變量Event事件數(shù)據(jù)AlarmEvent攜帶需要處理的數(shù)據(jù)Producer業(yè)務(wù)線程 Consumer處理線程 │ │ │ Push(event) │ Pop(event) ▼ ▲ Queue ──────────────────────────── [event1] [event2] [event3] ...核心生產(chǎn)者和消費(fèi)者通過(guò)隊(duì)列間接通信互不直接依賴互不等待。五、設(shè)備軟件里的真實(shí)場(chǎng)景場(chǎng)景一報(bào)警處理——最典型// 全局報(bào)警隊(duì)列 AlarmQueue g_alarmQueue; // 任意模塊報(bào)警 void Actor::Alarm(string msg) { Log(msg); Beeper::Beep(); g_alarmQueue.Push({time, code, msg}); // 丟隊(duì)列就走 }場(chǎng)景二日志寫(xiě)入日志如果直接寫(xiě)文件磁盤(pán) IO 慢時(shí)也會(huì)卡業(yè)務(wù)線程。用生產(chǎn)者消費(fèi)者struct LogEvent { string message; spdlog::level::level_enum level; }; class LogQueue { queueLogEvent m_queue; mutex m_mtx; condition_variable m_cv; public: void Push(LogEvent ev) { /* 入隊(duì) notify */ } bool Pop(LogEvent out) { /* 阻塞等 */ } }; LogQueue g_logQueue; // 業(yè)務(wù)線程 void Actor::Log(string msg) { g_logQueue.Push({msg, spdlog::level::info}); // 立刻返回不等磁盤(pán)寫(xiě)完 } // 日志消費(fèi)者線程 void LogConsumerThread() { while (true) { LogEvent ev; if (!g_logQueue.Pop(ev)) return; SpdLogger::Instance().Write(ev.message, ev.level); // 慢寫(xiě)磁盤(pán) } }場(chǎng)景三圖像存儲(chǔ)視覺(jué)檢測(cè)拍了一張圖要存到硬盤(pán)。存圖慢幾十毫秒不能卡檢測(cè)流程。struct ImageEvent { string name; cv::Mat image; }; ImageQueue g_imageQueue; // 視覺(jué)線程拍完 void OnCaptureDone(cv::Mat img) { g_imageQueue.Push({timestamp, img}); // 立刻繼續(xù)拍下一張 } // 圖像消費(fèi)者線程 void ImageConsumerThread() { while (true) { ImageEvent ev; if (!g_imageQueue.Pop(ev)) return; cv::imwrite(D:\\images\\ ev.name .bmp, ev.image); } }場(chǎng)景四SECS 上報(bào)SECS 通信慢且可能超時(shí)。業(yè)務(wù)事件上報(bào)用隊(duì)列SecsQueue g_secsQueue; // 任意模塊要上報(bào) void ReportEvent(string eventCode, string data) { g_secsQueue.Push({eventCode, data}); // 立刻返回不等網(wǎng)絡(luò) } // SECS 消費(fèi)者線程 void SecsConsumerThread() { while (true) { SecsEvent ev; if (!g_secsQueue.Pop(ev)) return; secs-SendEvent(ev.code, ev.data); // 慢網(wǎng)絡(luò)通信 } }場(chǎng)景五料號(hào)流轉(zhuǎn)更宏觀地看整臺(tái)設(shè)備的料流轉(zhuǎn)也是生產(chǎn)者消費(fèi)者上料站生產(chǎn)者──→ Buffer隊(duì)列──→ 測(cè)試臺(tái)消費(fèi)者上料站不斷往 Buffer 里放料測(cè)試臺(tái)從 Buffer 取料測(cè)。上料站不等測(cè)試臺(tái)測(cè)完才放下一個(gè)測(cè)試臺(tái)不等上料站放好才取。Buffer 就是中間隊(duì)列。六、關(guān)鍵實(shí)現(xiàn)細(xì)節(jié)細(xì)節(jié)一隊(duì)列必須線程安全void Push(T item) { { lock_guardmutex lk(m_mtx); // 加鎖 m_queue.push(item); } m_cv.notify_one(); // 通知 } bool Pop(T out) { unique_lockmutex lk(m_mtx); m_cv.wait(lk, [] { return !m_queue.empty() || !m_running; }); // 等待 if (!m_running m_queue.empty()) return false; out m_queue.front(); m_queue.pop(); return true; }要點(diǎn)Push加鎖入隊(duì)后notify_onePop用條件變量阻塞等待不空輪詢。細(xì)節(jié)二隊(duì)列要能停void Stop() { { lock_guardmutex lk(m_mtx); m_running false; } m_cv.notify_all(); // 叫醒所有等待的消費(fèi)者 }程序退出時(shí)要能優(yōu)雅停掉消費(fèi)者線程不能讓它永遠(yuǎn)阻塞在Pop上。細(xì)節(jié)三隊(duì)列滿了怎么辦如果生產(chǎn)速度遠(yuǎn)超消費(fèi)速度隊(duì)列會(huì)無(wú)限增長(zhǎng)。兩種策略// 策略一丟棄最舊的保留最新 void Push(T item) { lock_guardmutex lk(m_mtx); if (m_queue.size() m_maxSize) m_queue.pop(); // 丟最舊的 m_queue.push(item); } // 策略二丟棄最新的保留正在處理的 void Push(T item) { lock_guardmutex lk(m_mtx); if (m_queue.size() m_maxSize) return; // 滿了就不收 m_queue.push(item); }設(shè)備軟件里日志/報(bào)警通常用策略一丟最舊的保最新的圖像通常用策略二滿了就不收新的避免內(nèi)存爆。細(xì)節(jié)四UI 操作必須發(fā)到 UI 線程消費(fèi)者線程不能直接操作 MFC 控件要用PostMessage發(fā)到 UI 線程// 消費(fèi)者線程里 PostMessage(pDlg-m_hWnd, WM_SHOW_ALARM, 0, (LPARAM)new string(ev.message)); // UI 線程里處理 BEGIN_MESSAGE_MAP(CMyDlg, CDialogEx) ON_MESSAGE(WM_SHOW_ALARM, OnShowAlarm) END_MESSAGE_MAP() LRESULT CMyDlg::OnShowAlarm(WPARAM wParam, LPARAM lParam) { string* msg (string*)lParam; m_lblAlarm.SetWindowText(msg-c_str()); delete msg; return 0; }七、生產(chǎn)者消費(fèi)者 vs 觀察者什么區(qū)別這兩個(gè)都涉及「事件通知」容易混。生產(chǎn)者消費(fèi)者觀察者通信方式通過(guò)隊(duì)列間接直接調(diào)回調(diào)是否等待生產(chǎn)者不等消費(fèi)者事件源不等觀察者同步調(diào)但不等返回緩沖有隊(duì)列緩沖無(wú)緩沖直接調(diào)適合慢 IO、異步處理事件廣播、多訂閱例子日志寫(xiě)盤(pán)、報(bào)警存庫(kù)報(bào)警通知 UI 刷新關(guān)系觀察者可以和生產(chǎn)者消費(fèi)者組合——事件源用觀察者通知觀察者收到后 Push 到隊(duì)列消費(fèi)者從隊(duì)列取。這樣既解耦了事件源和響應(yīng)者又解耦了響應(yīng)者和慢操作。八、生產(chǎn)者消費(fèi)者的坑坑一隊(duì)列無(wú)限增長(zhǎng)導(dǎo)致內(nèi)存爆生產(chǎn)速度遠(yuǎn)超消費(fèi)速度時(shí)隊(duì)列越來(lái)越大最終內(nèi)存耗盡。對(duì)策設(shè)最大長(zhǎng)度滿了按策略丟棄?？佣M(fèi)者線程崩潰后隊(duì)列堆積消費(fèi)者線程掛了生產(chǎn)者還在 Push隊(duì)列堆積。對(duì)策消費(fèi)者線程加異常保護(hù)崩了自動(dòng)重啟或者監(jiān)控隊(duì)列長(zhǎng)度超限報(bào)警。坑三事件數(shù)據(jù)里有指針消費(fèi)者用時(shí)已失效struct Event { Station* station; // 指針 }; // 生產(chǎn)者 Push 時(shí) station 還在 // 消費(fèi)者 Pop 時(shí) station 已被 delete → 野指針對(duì)策事件里存 ID 或值拷貝不存指針或用shared_ptr?；??？铀亩鄠€(gè)消費(fèi)者時(shí)重復(fù)處理如果起了兩個(gè)消費(fèi)者線程從同一個(gè)隊(duì)列 Pop一個(gè)事件只會(huì)被一個(gè)消費(fèi)者處理。如果需要廣播給所有消費(fèi)者要用多個(gè)隊(duì)列每個(gè)消費(fèi)者一個(gè)。九、可復(fù)用結(jié)論生產(chǎn)者消費(fèi)者模式的本質(zhì)生產(chǎn)者往隊(duì)列丟消費(fèi)者從隊(duì)列取兩邊各跑各的互不等待。四個(gè)角色Producer、Consumer、Queue線程安全條件變量、Event。設(shè)備軟件最典型應(yīng)用報(bào)警處理、日志寫(xiě)入、圖像存儲(chǔ)、SECS 上報(bào)、料號(hào)流轉(zhuǎn)。核心價(jià)值快節(jié)奏業(yè)務(wù)線程不被慢 IO 卡住生產(chǎn)消費(fèi)解耦隊(duì)列緩沖削峰填谷。和觀察者的區(qū)別觀察者直接調(diào)回調(diào)無(wú)緩沖生產(chǎn)者消費(fèi)者通過(guò)隊(duì)列有緩沖?？雨?duì)列限長(zhǎng)防內(nèi)存爆、消費(fèi)者崩了要重啟、事件不存裸指針、多消費(fèi)者要多隊(duì)列。生產(chǎn)者消費(fèi)者模式在設(shè)備軟件里不是「課本上的 Producer-Consumer」而是「日志怎么不卡運(yùn)動(dòng)、報(bào)警怎么不卡狀態(tài)機(jī)、圖像怎么不卡檢測(cè)」的實(shí)打?qū)嵔夥āＳ脤?duì)了業(yè)務(wù)線程永遠(yuǎn)不被 IO 拖住用錯(cuò)了要么業(yè)務(wù)線程被磁盤(pán)卡死要么隊(duì)列無(wú)限增長(zhǎng)內(nèi)存爆掉。關(guān)鍵是分清「快操作直接做」和「慢操作丟隊(duì)列」在兩者之間放一個(gè)線程安全的緩沖隊(duì)列。