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