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