前言線程池thread pool要解決的問題很具體頻繁地創(chuàng)建和銷毀線程是有代價的。每次std::thread構(gòu)造都要向操作系統(tǒng)申請內(nèi)核對象和棧空間銷毀時要回收任務(wù)本身卻可能只執(zhí)行幾微秒。當(dāng)任務(wù)數(shù)量遠大于 CPU 核心數(shù)時為每個任務(wù)開一個線程還會讓調(diào)度器忙于切換緩存局部性變差。線程池的思路是預(yù)先創(chuàng)建固定數(shù)量的工作線程讓它們從一個共享隊列里取任務(wù)執(zhí)行把「線程的生命周期」和「任務(wù)的生命周期」解耦。關(guān)于線程池最常見的誤解有三個。第一認為線程數(shù)越多吞吐越高——實際上超過硬件并發(fā)度之后額外線程只會增加上下文切換和內(nèi)存占用對 CPU 密集型任務(wù)沒有幫助。第二認為「無鎖隊列一定更快」——在競爭激烈時自旋和 CAS 失敗重試的代價可能超過一把短臨界區(qū)的互斥量而且無鎖隊列的正確性極難保證。第三認為線程池能自動解決數(shù)據(jù)競爭——池只負責(zé)調(diào)度任務(wù)之間共享數(shù)據(jù)的同步仍然要由調(diào)用者負責(zé)。本文按「為什么需要 → 由哪些部件組成 → 完整實現(xiàn) → 關(guān)閉與異常邊界」的順序講解給出一個基于 C17 標準庫、可直接放進項目的線程池并逐條列出實現(xiàn)中最容易踩的坑。文中所有代碼都只按標準語義推演本機沒有編譯器不聲稱任何執(zhí)行結(jié)果。一、線程池的組成部件一個最小可用的線程池由四部分組成。工作線程集合一組長期存活的std::thread每個線程運行同一個循環(huán)函數(shù)。任務(wù)隊列存放待執(zhí)行任務(wù)的容器通常是std::queuestd::functionvoid()。類型擦除讓不同簽名的任務(wù)能排進同一個隊列。同步原語一把std::mutex保護隊列一個std::condition_variable在隊列為空時掛起工作線程。停止標志告訴工作線程「不會再有新任務(wù)」讓它們把剩余任務(wù)做完后退出。工作線程的主循環(huán)可以概括成一句話加鎖、在條件變量上等待「有任務(wù)或要停止」、取一個任務(wù)、解鎖、執(zhí)行。解鎖必須在執(zhí)行任務(wù)之前完成否則所有工作線程會被串行化線程池退化成單線程。這里有一個關(guān)鍵設(shè)計點隊列里存什么。如果只存std::functionvoid()調(diào)用者拿不到返回值也拿不到異常。標準庫給出的答案是std::packaged_taskR()它把可調(diào)用對象包裝成一個任務(wù)內(nèi)部持有一個共享狀態(tài)關(guān)聯(lián)一個std::futureR。任務(wù)執(zhí)行時返回值或異常被寫進共享狀態(tài)調(diào)用者通過 future 取回。線程池只需要把packaged_task再包一層成void()塞進隊列即可。二、任務(wù)提交與類型推導(dǎo)提交接口通常是可變參數(shù)模板template class F, class... Argsauto submit(F f, Args... args) - std::futurestd::invoke_result_tF, Args...;返回類型用std::invoke_result_tF, Args...推導(dǎo)C17 起可用C11/14 用std::result_of_t后者在 C17 棄用、C20 移除。std::invoke_result比decltype(f(args...))更準確因為它同時支持普通可調(diào)用對象、函數(shù)指針和指向成員的指針。參數(shù)如何綁定到任務(wù)上有兩種寫法。第一種是std::bind(std::forwardF(f), std::forwardArgs(args)...)第二種是 lambda 捕獲。lambda 的完美轉(zhuǎn)發(fā)捕獲[f std::forwardF(f), ...]這種包展開需要 C20C17 下寫不出來所以本文用std::bind。但必須知道std::bind的兩個限制它按值 decay-copy 所有實參引用語義要用std::ref顯式表達它要求被綁定的可調(diào)用對象和實參都是可拷貝構(gòu)造的因此 move-only 的捕獲比如std::unique_ptr不能直接走這條路。如果項目里大量使用 move-only 捕獲更穩(wěn)妥的做法是讓submit直接接收一個已經(jīng)綁定好的std::functionvoid()或無參 lambda。還有一個容易忽略的點std::bind返回的對象在調(diào)用時并不會保留實參的值類別傳進來的右值在綁定那一刻就被移動或拷貝進了 bind 的存儲。這對大多數(shù)任務(wù)沒有影響但如果任務(wù)內(nèi)部依賴引用語義修改外部變量必須寫std::ref(x)。三、完整實現(xiàn)下面是一個完整版本包含構(gòu)造、提交、工作循環(huán)和優(yōu)雅關(guān)閉。// 適用于 C17#include condition_variable#include functional#include future#include mutex#include queue#include stdexcept#include thread#include type_traits#include utility#include vectorclass ThreadPool {public:explicit ThreadPool(std::size_t n std::thread::hardware_concurrency()): stop_(false) {if (n 0) n 1; // hardware_concurrency 可能返回 0workers_.reserve(n);for (std::size_t i 0; i n; i) {workers_.emplace_back([this] { workerLoop(); });}}~ThreadPool() { shutdown(); }ThreadPool(const ThreadPool) delete;ThreadPool operator(const ThreadPool) delete;template class F, class... Argsauto submit(F f, Args... args)- std::futurestd::invoke_result_tF, Args... {using R std::invoke_result_tF, Args...;auto bound std::bind(std::forwardF(f), std::forwardArgs(args)...);auto task std::make_sharedstd::packaged_taskR()(std::move(bound));std::futureR fut task-get_future();{std::lock_guardstd::mutex lk(mtx_);if (stop_) {throw std::runtime_error(submit on a stopped ThreadPool);}// 類型擦除packaged_taskR() 不能直接放進 void() 的隊列tasks_.emplace([task] { (*task)(); });}cv_.notify_one();return fut;}void shutdown() {{std::lock_guardstd::mutex lk(mtx_);if (stop_) return;stop_ true;}cv_.notify_all();for (auto t : workers_) {if (t.joinable()) t.join();}}private:void workerLoop() {for (;;) {std::functionvoid() job;{std::unique_lockstd::mutex lk(mtx_);cv_.wait(lk, [this] { return stop_ || !tasks_.empty(); });if (stop_ tasks_.empty()) return; // 停止且已排空job std::move(tasks_.front());tasks_.pop();} // 先解鎖job(); // 再執(zhí)行}}std::vectorstd::thread workers_;std::queuestd::functionvoid() tasks_;std::mutex mtx_;std::condition_variable cv_;bool stop_;};幾個細節(jié)值得展開。為什么stop_不用std::atomicbool。它只在持鎖時讀寫鎖已經(jīng)提供了同步普通bool就夠。若改成原子變量而仍然在鎖外讀就必須自己保證「檢查停止標志」和「檢查隊列」這兩步之間不出現(xiàn)競態(tài)反而更容易寫錯。為什么用std::make_sharedstd::packaged_taskR()。packaged_task不可拷貝而std::function要求目標可拷貝。把它放進shared_ptrlambda 捕獲的是這個指針拷貝的是指針本身任務(wù)對象仍然只有一份。make_shared讓控制塊和對象一次分配比shared_ptr(new ...)少一次分配也讓異常安全性更好。等待謂詞為什么是stop_ || !tasks_.empty()。條件變量存在虛假喚醒spurious wakeup不帶謂詞的wait返回后條件未必成立。帶謂詞的重載會在內(nèi)部循環(huán)只有謂詞為真才返回這是標準庫推薦寫法。為什么shutdown里要notify_all而不是notify_one。停止信號是廣播性質(zhì)的所有工作線程都要醒來檢查stop_只喚醒一個會讓其余線程永遠睡在條件變量上join隨之卡死。四、關(guān)閉語義與返回值關(guān)閉有兩種常見語義。第一種是「排空后退出」已經(jīng)入隊的任務(wù)全部執(zhí)行完再結(jié)束上面的實現(xiàn)就是這種判斷條件是stop_ tasks_.empty()。第二種是「立即丟棄」清空隊列直接退出適合取消場景。兩者沒有對錯但必須在文檔里寫清楚否則調(diào)用者無法預(yù)期析構(gòu)要花多久。如果某個任務(wù)會阻塞很久等待網(wǎng)絡(luò)、等待另一個池的任務(wù)排空式關(guān)閉就會讓析構(gòu)函數(shù)長時間阻塞。因此生產(chǎn)代碼常常再加一個std::atomicbool cancelled_讓長任務(wù)自己檢查取消標志。返回值路徑是這樣的packaged_task::get_future()在提交線程調(diào)用拿到std::futureR任務(wù)在工作線程執(zhí)行時把結(jié)果寫進共享狀態(tài)調(diào)用線程fut.get()阻塞直到結(jié)果就緒。如果任務(wù)拋出異常packaged_task會捕獲它并存入共享狀態(tài)get()重新拋出——這一點很重要它意味著工作者線程不會因為任務(wù)異常而std::terminate。但前提是任務(wù)確實被packaged_task包著。如果隊列里存的是裸std::function且直接調(diào)用job()異常會逃出workerLoop而線程函數(shù)拋出未捕獲異常會調(diào)用std::terminate。使用示例如下。// 適用于 C17#include iostream#include stringint main() {ThreadPool pool(4);auto sum pool.submit([](int a, int b) { return a b; }, 2, 3);auto text pool.submit([] { return std::string(hello); });auto boom pool.submit([]() - int { throw std::runtime_error(bad); });std::cout sum.get() \n; // 5std::cout text.get() \n; // hellotry {boom.get(); // 異常在這里重新拋出} catch (const std::exception e) {std::cout caught: e.what() \n;}pool.shutdown(); // 也可以交給析構(gòu)return 0;}注意sum.get()與text.get()的先后順序只影響等待順序不影響任務(wù)執(zhí)行順序——多線程下任務(wù)的完成次序本來就不確定。要保證順序只能用單工作線程的池或者在任務(wù)之間自行加同步。常見坑點? 在析構(gòu)函數(shù)里對工作線程調(diào)用detach()然后成員隊列、互斥量先被銷毀。? 析構(gòu)里先置停止標志、notify_all、再逐個join()確保線程在成員銷毀前退出。?cv_.wait(lk)不帶謂詞被虛假喚醒后直接tasks_.front()。?cv_.wait(lk, [this]{ return stop_ || !tasks_.empty(); })醒來后條件一定成立。? 持鎖執(zhí)行任務(wù){(diào) lock_guard lk(mtx_); job(); }所有工作者被串行化。? 在鎖內(nèi)只取出任務(wù)出作用域解鎖后再執(zhí)行。? 工作線程中拋出異常且沒人捕獲程序直接std::terminate。? 用packaged_task包裝任務(wù)或在workerLoop里try/catch兜底記錄日志。? 任務(wù) lambda 按引用捕獲局部變量submit返回后局部變量已銷毀。? 按值捕獲或確保被引用對象的生命周期覆蓋任務(wù)執(zhí)行期。? 在池內(nèi)任務(wù)里調(diào)用pool.shutdown()工作線程join自己。? 關(guān)閉由池外線程發(fā)起池內(nèi)任務(wù)只設(shè)置取消標志。? 用std::thread::hardware_concurrency()直接當(dāng)線程數(shù)不處理返回 0 的情況。? 判零后回退到 1或由調(diào)用者顯式傳入。? 丟棄submit返回的 future任務(wù)拋出的異常被靜默吞掉。? 需要感知失敗時保留 future 并get()或在任務(wù)內(nèi)部自行處理異常。總結(jié)部件作用關(guān)鍵約束工作線程數(shù)組長期存活循環(huán)取任務(wù)析構(gòu)前必須join任務(wù)隊列類型擦除后統(tǒng)一存放必須由互斥量保護條件變量隊列空時掛起線程必須帶謂詞防虛假喚醒packaged_task傳遞返回值與異常不可拷貝用shared_ptr持有停止標志通知線程退出與隊列檢查在同一把鎖下線程池的本質(zhì)是把「線程」這個昂貴資源變成可復(fù)用的池化資源核心難點不在語法而在同步邊界鎖的粒度、條件變量的謂詞、關(guān)閉時機的定義。把這三件事寫對剩下的就是按業(yè)務(wù)補充取消、優(yōu)先級和動態(tài)擴縮容。任何涉及共享狀態(tài)的修改都要在鎖內(nèi)完成任何可能阻塞的調(diào)用都不要持鎖執(zhí)行。