最近我在重構(gòu)一個內(nèi)部數(shù)據(jù)處理模塊核心管道寫得非常漂亮一段std::ranges視圖經(jīng)過filter、transform把非結(jié)構(gòu)化字段規(guī)整成干凈的結(jié)構(gòu)體最后落到ranges::sort做時序排列?？尚阅軠y試一出來就有點尷尬這些 ranges 算法全在單線程上跑機器上的多核基本在看熱鬧。想把并行執(zhí)行做進去讓 ranges 算法配合自己的線程池與工作隊列結(jié)果發(fā)現(xiàn)標準庫給出的支持比想象中少得多。這篇文章就是我在反復折騰后整理出來的一套落地思路如何拆范圍、建隊列、控線程數(shù)讓 range 管道真正吃滿多核順便把踩過的坑都交代清楚。1. std::ranges 并行這件事標準到底給了什么1.1 ranges 算法很香但為什么等不到并行重載std::ranges算法是 C20 帶來的最大語法糖之一。以往我們寫排序、變換、查找都要小心翼翼地傳迭代器對代碼一長就變成“迭代器地獄”。有了 ranges 之后std::ranges::sort(data)、std::ranges::for_each(v, fn)這種寫法不僅簡短而且能配合 view 做惰性求值和管道拼接表達力直接上一個臺階。但有一個非?，F(xiàn)實的坑標準庫的 ranges 算法并不帶并行執(zhí)行策略重載。至少到 C23 標準std::ranges::sort、std::ranges::transform這些算法都沒有接受std::execution::par的版本。你可能會寫出下面這種代碼然后發(fā)現(xiàn)編譯期直接報錯// 這只是理想寫法目前標準庫中并不存在這樣的重載 std::ranges::sort(std::execution::par, data); // 編譯不過 std::ranges::transform(std::execution::par, input, output, fn); // 編譯不過也就是說標題里那個“ranges 算法并行執(zhí)行”并不能靠標準庫一個參數(shù)白拿。想并行要么退回傳統(tǒng)迭代器算法加std::execution::par要么自己包一層調(diào)度把 range 按塊拆開派發(fā)給工作線程。委員會沒有把兩者接通原因比較復雜核心是 ranges 和后來的 executors 方向還沒有合并成穩(wěn)定的統(tǒng)一模型std::execution::par只是早期執(zhí)行策略形態(tài)真正的調(diào)度器體系直到后來的提案才逐漸成型。作為使用者我們不能干等標準最好的路線是自己把調(diào)度層補出來。1.2 標準并行執(zhí)行策略的真實工作方式C17 提供了四個執(zhí)行策略標簽seq串行、unseq向量化、par多線程、par_unseq線程加向量化。它的用法集中在傳統(tǒng)算法上std::sort(std::execution::par, data.begin(), data.end()); std::for_each(std::execution::par, data.begin(), data.end(), fn);它會“盡可能”并行但標準對并行粒度、線程數(shù)量、任務(wù)隊列完全不規(guī)定。也就是說并行行為完全取決于實現(xiàn)內(nèi)部的后端有的實現(xiàn)會用自己的一套任務(wù)調(diào)度器有的實現(xiàn)會綁硬件線程數(shù)做并行分塊有的是在第一次調(diào)用時初始化一個全局線程池。問題也藏在這里。這套后端對你是不可見的你沒法告訴它“我這個項目已經(jīng)有線程池了請復用”也沒法控制“我今天只想用 4 個線程不要跑滿 16 個邏輯核”更沒法在業(yè)務(wù)高峰期控制隊列深度。當一個模塊里同時出現(xiàn)多個并行算法調(diào)用時它們各自獨立調(diào)度線程數(shù)翻倍、CPU 超訂是常有的事。所以我一直認為std::execution::par適合“粗粒度地加速單次算法調(diào)用”的場景但如果你的項目里本身就有線程池或者希望精準控制任務(wù)隊列那必須自己接管調(diào)度層。這正好是本文標題里“線程池管理與工作隊列優(yōu)化”存在的價值。1.3 不讓標準庫替你調(diào)度那你需要解決什么問題當你不使用std::execution::par而是想自己控制并行時核心要解決四個問題并行度控制用多少線程線程數(shù)是否動態(tài)可變。任務(wù)切分range 那么大按什么粒度切成塊塊數(shù)多少合適。任務(wù)派發(fā)與回收塊切好了怎么把每個塊作為任務(wù)丟進隊列線程執(zhí)行完怎么通知主線程。結(jié)果合并每個線程處理完得到的是局部結(jié)果如何高效合成為全局結(jié)果。這四個問題聽起來多其實拆開就是一套線程池加一套分塊策略。下面我先把線程池骨架寫出來再補分塊和合并的完整玩法。2. 自建線程池隊列、線程數(shù)和提交接口的設(shè)計2.1 一個能直接跑起來的最小線程池先給一個最小但完整的線程池實現(xiàn)。它不算最優(yōu)但足夠作為后續(xù)優(yōu)化的起點#include condition_variable #include deque #include future #include mutex #include thread #include vector class ThreadPool { public: explicit ThreadPool(size_t threadCount) : stop_(false) { for (size_t i 0; i threadCount; i) { workers_.emplace_back([this] { for (;;) { std::packaged_taskvoid() job; { std::unique_lockstd::mutex lock(mutex_); cv_.wait(lock, [this] { return stop_ || !queue_.empty(); }); if (stop_ queue_.empty()) return; job std::move(queue_.front()); queue_.pop_front(); } job(); } }); } } template typename F auto submit(F func) - std::futurevoid { std::packaged_taskvoid() task(std::forwardF(func)); auto future task.get_future(); { std::lock_guardstd::mutex lock(mutex_); queue_.push_back(std::move(task)); } cv_.notify_one(); return future; } ~ThreadPool() { { std::lock_guardstd::mutex lock(mutex_); stop_ true; } cv_.notify_all(); for (auto worker : workers_) { worker.join(); } } private: std::atomicbool stop_; std::dequestd::packaged_taskvoid() queue_; std::mutex mutex_; std::condition_variable cv_; std::vectorstd::thread workers_; };這個類值得注意的細節(jié)有三個為什么用std::packaged_taskvoid()而不是std::functionvoid()packaged_task自帶future提交時就能拿到結(jié)果句柄省去再包一層shared_ptr的麻煩。worker 執(zhí)行job()時future狀態(tài)會被置為 ready調(diào)用方通過future.wait()或者future.get()回收結(jié)果。為什么析構(gòu)時先加鎖再notify_all因為 worker 循環(huán)里對停止狀態(tài)和隊列空判斷同時發(fā)生在鎖內(nèi)。如果不加鎖直接設(shè)置stop_可能出現(xiàn) worker 剛判完queue_.empty()主線程就設(shè)置停止并通知worker 醒來后看到隊列已空才退出邏輯上沒問題但 stop 標志寫入和讀取最好由同一把鎖保護避免晦澀的競態(tài)。為什么隊列用std::deque而不是std::vectordeque支持兩端高效操作。雖然我們目前只從尾部入隊、頭部出隊但后續(xù)如果要做 LIFO 策略新任務(wù)優(yōu)先執(zhí)行只需把push_back改成push_front不用換容器。這個實現(xiàn)里submit返回std::futurevoid對大多數(shù)“數(shù)據(jù)分塊處理、最終各塊互不相干”的任務(wù)完全夠用。如果后面需要每個塊返回一個具體值改成std::futureT版本即可下面會講到。2.2 線程數(shù)量不是越多越好線程池最容易被忽略的參數(shù)是線程數(shù)。很多人直接寫std::thread::hardware_concurrency()然后期望性能拉滿。實測下來這條規(guī)則對純 CPU 密集任務(wù)是“半個坑”。hardware_concurrency返回的是邏輯處理器數(shù)量包含超線程。在 8 核 16 線程的機器上它會返回 16但如果你跑的是純浮點計算16 個“邏輯處理器”共享物理核心的執(zhí)行單元線程開滿反而帶來上下文切換和調(diào)度開銷。我對純 CPU 密集任務(wù)的經(jīng)驗是先從物理核數(shù)或邏輯核數(shù)的一半起步跑一輪實測再向上調(diào)。遇到任務(wù)里有 IO 等待、鎖等待、系統(tǒng)調(diào)用時可以適當加線程數(shù)讓部分線程在等待間隙不阻塞整體進度。還有一個隱藏因素任務(wù)塊的數(shù)量和線程數(shù)的比值。這個比值直接決定調(diào)度開銷的占比。比如一塊數(shù)據(jù) 1.2 億個元素開 8 個線程每個線程處理 1500 萬個元素單塊執(zhí)行時間可能幾十毫秒調(diào)度開銷可忽略。但如果數(shù)據(jù)只有 10 萬元素同樣開 8 個線程每個塊只有一萬多元素單塊執(zhí)行時間可能只有幾十微秒那么隊列鎖、notify_one、future 狀態(tài)同步的開銷就可能占到 20% 以上。具體怎么定塊大小下面第四章我會給一組實測數(shù)據(jù)。2.3 提交接口從單任務(wù)到批量入隊上面的submit一次只能投遞一個任務(wù)每次都要加鎖、解鎖、notify_one。當任務(wù)塊數(shù)量達到幾千甚至幾萬時鎖競爭會成為瓶頸。一個很實用的優(yōu)化是增加批量提交接口template typename F void submitBatch(std::vectorF tasks) { { std::lock_guardstd::mutex lock(mutex_); for (auto task : tasks) { std::packaged_taskvoid() wrapped(std::move(task)); queue_.push_back(std::move(wrapped)); } } cv_.notify_all(); }批量提交把 N 次加鎖變成一次加鎖然后一次性喚醒所有 idle 線程。你可能覺得“一次提交 32 個任務(wù)還省什么鎖”但如果任務(wù)塊數(shù)上千每次 submit 都鎖同一個 mutex等待時間會迅速累積。我在實際工程里遇到的一個場景是對 1.2 億條日志按每 1024 條分塊分出一萬多個任務(wù)逐個 submit 的那一版光提交階段就消耗了接近 200ms改成批量化插入后提交階段縮到 30ms 以內(nèi)。另一個值得做的是去掉 future 的開銷。如果任務(wù)沒有返回值你只需要知道“全部執(zhí)行完了”保留一堆futurevoid其實有些浪費。可以用一個原子計數(shù)器加條件變量任務(wù)完成后計數(shù)器遞減最后一個完工的任務(wù)通知主線程。這樣既省 future 列表的內(nèi)存也省逐個 wait 的循環(huán)。std::atomicsize_t pending{0}; std::condition_variable doneCv; std::mutex doneMutex; void pushTask(std::packaged_taskvoid() task) { { std::lock_guardstd::mutex lock(mutex_); pending.fetch_add(1); queue_.push_back(std::move(task)); } cv_.notify_one(); } void onTaskFinished() { if (pending.fetch_sub(1) 1) { std::lock_guardstd::mutex lock(doneMutex); doneCv.notify_all(); } } void waitAll() { std::unique_lockstd::mutex lock(doneMutex); doneCv.wait(lock, [] { return pending.load() 0; }); }注意務(wù)必在任務(wù)內(nèi)部調(diào)用onTaskFinished()并確保異常安全。如果任務(wù)拋異常計數(shù)會丟失waitAll永遠等不到零。穩(wěn)妥做法是用 RAII 包一層析構(gòu)時遞減計數(shù)。3. 把 ranges 算法拆到工作隊列上三種模式與代碼落地線程池有了接下來就是把std::ranges算法真正派發(fā)上去。不同算法有不同的并行模式不能一把梭。我把常見場景分成三類映射型、歸約型、全局依賴型。3.1 映射型任務(wù)for_each / transform最基礎(chǔ)也最容易并行化的是for_each、transform這類逐元素操作塊與塊之間沒有任何數(shù)據(jù)依賴。核心思路把 range 切成連續(xù)塊每塊獨立調(diào)用ranges::for_each或ranges::transform。template typename Range, typename Fn void parallelTransformInplace(ThreadPool pool, Range range, Fn fn) { using std::ranges::begin; using std::ranges::end; size_t n std::ranges::distance(range); if (n 0) return; const size_t hw std::thread::hardware_concurrency(); const size_t blocks hw * 4; // 塊數(shù)約為線程數(shù) 4 倍 const size_t chunk (n blocks - 1) / blocks; // 每塊元素數(shù) auto it begin(range); std::vectorstd::futurevoid futures; futures.reserve(blocks); for (size_t offset 0; offset n; offset chunk) { auto blockBegin std::next(it, offset); auto blockEnd std::next(blockBegin, std::min(chunk, n - offset)); futures.push_back(pool.submit([blockBegin, blockEnd, fn]() { std::ranges::for_each(blockBegin, blockEnd, fn); })); } for (auto f : futures) { f.wait(); } }這個實現(xiàn)有個前提range 是隨機訪問范圍。std::next(it, offset)在隨機訪問迭代器上 O(1)但如果底層是filter_view、istream_view每次推進都要遍歷元素切塊的代價會很高。我的做法是先把非隨機訪問范圍 materialize 成 vector再做并行分塊。比如從文件讀出的日志行經(jīng)過filter過濾后可能只有 30% 通過直接按行號分塊容易導致負載不均先拷進 vector 再分塊反而更穩(wěn)定。對于transform要注意函數(shù)副作用。如果變換函數(shù)修改的是容器外共享狀態(tài)比如同一個計數(shù)器那并行時每個線程都在競爭結(jié)果既可能錯誤也會變慢。正確姿勢是內(nèi)部無狀態(tài)或只修改自己塊內(nèi)的元素。3.2 歸約型任務(wù)并行局部結(jié)果再合并求和、找最大值、統(tǒng)計滿足條件的數(shù)量這類任務(wù)的并行模式是“局部歸約 全局合并”。每個線程獨立計算自己的那塊最后匯總。很典型的例子是統(tǒng)計std::ranges::filter之后的元素數(shù)量template typename Range, typename Pred int64_t parallelCountIf(ThreadPool pool, const Range range, Pred pred) { auto n std::ranges::distance(range); if (n 0) return 0; const size_t hw std::thread::hardware_concurrency(); const size_t blocks hw * 4; const size_t chunk (n blocks - 1) / blocks; std::vectorint64_t partials(blocks, 0); std::vectorstd::futurevoid futures; auto it std::ranges::begin(range); for (size_t b 0; b blocks; b) { auto blockBegin std::next(it, b * chunk); auto blockEnd std::next(blockBegin, std::min(chunk, n - b * chunk)); futures.push_back(pool.submit([blockBegin, blockEnd, pred, partials, b]() { int64_t count 0; for (auto iter blockBegin; iter ! blockEnd; iter) { if (pred(*iter)) count; } partials[b] count; })); } for (auto f : futures) f.wait(); return std::accumulate(partials.begin(), partials.end(), int64_t{0}); }這里關(guān)鍵點是不要讓每個線程原子地累加一個全局計數(shù)器。原子fetch_add在溢出一級緩存后會形成 cacheline 乒乓性能遠不如各算各的、最后一次性合并。局部partials數(shù)組本身也可能踩偽共享的坑這點我放到第四章專門講。另一個技巧是如果任務(wù)塊內(nèi)還需要繼續(xù)用 ranges 表達比如auto view subRange | std::views::filter(pred);完全可以在 lambda 里構(gòu)建局部視圖讓代碼保持和單線程版本一樣的可讀性。并行化只是把外層切塊內(nèi)部該用什么算法就用什么算法。3.3 全局依賴型任務(wù)排序的分塊與歸并排序比前面兩種復雜因為全數(shù)組最終必須有序塊與塊之間不是無關(guān)的。最經(jīng)典的并行策略是分塊并行排序再做多路歸并。template typename T void parallelSort(std::vectorT data, ThreadPool pool) { const size_t n data.size(); if (n 10000) { std::ranges::sort(data); return; } const size_t hw std::thread::hardware_concurrency(); const size_t blockCount hw * 2; // 排序場景塊數(shù)不宜過多 const size_t chunk (n blockCount - 1) / blockCount; std::vectorstd::futurevoid futures; for (size_t i 0; i n; i chunk) { auto begin data.begin() i; auto end begin std::min(chunk, n - i); futures.push_back(pool.submit([begin, end]() { std::ranges::sort(begin, end); // 每個塊內(nèi)部先有序 })); } for (auto f : futures) f.wait(); // 多輪歸并 for (size_t width chunk; width n; width * 2) { for (size_t left 0; left n; left 2 * width) { auto mid data.begin() std::min(left width, n); auto last data.begin() std::min(left 2 * width, n); std::inplace_merge(data.begin() left, mid, last); } } }注意這里歸并階段還是串行的塊數(shù)越多歸并的輪數(shù)和成本越高所以排序時塊數(shù)不能像映射型任務(wù)那樣取太多。我實際測下來排序場景的塊數(shù)取線程數(shù)的 1 到 2 倍效果最好再多就會讓歸并成為新瓶頸。想進一步并行化歸并階段可以對兩個有序段做二分切分找到左右兩個子問題后并行歸并但實現(xiàn)復雜度高不少工作量需要和收益匹配。如果你的數(shù)據(jù)不是簡單 vector而是經(jīng)過views::filter或views::transform的視圖那并行排序前必須先 materialize。因為排序需要隨機訪問和原地交換視圖往往無法滿足。我的做法是std::vectorT tmp; if constexpr (std::ranges::random_access_rangeRange std::ranges::sized_rangeRange) { // 直接對目標 range 分塊 } else { tmp.assign(range.begin(), range.end()); // 對 tmp 排序若需要再寫回原容器 }4. 實測數(shù)據(jù)塊大小、線程數(shù)和隊列鎖的交叉影響4.1 測試場景與基線數(shù)據(jù)我在一臺 8 核 16 線程的 Linux 機器上做了組對照實驗。模擬項目X的數(shù)據(jù)集是 1.2 億個float總體數(shù)據(jù)量約 480MB。任務(wù)分三段先對每個元素做一次非線性變換再統(tǒng)計超過閾值的樣本數(shù)最后對結(jié)果排序輸出前 10000 個。單線程版本用純std::ranges管道耗時約 2180ms。傳統(tǒng)標準庫并行算法std::for_each(std::execution::par, ...)跑完約 640ms。自建線程池在參數(shù)沒調(diào)好時甚至跑到 900ms 以上調(diào)好之后降到 410ms。差距很大值得把參數(shù)關(guān)系展開講。4.2 塊數(shù)與線程數(shù)的交叉測試下面這組數(shù)據(jù)是同一批數(shù)據(jù)、同樣 16 個線程情況下不同塊數(shù)對耗時的影響配置transform 階段耗時 (ms)排序階段耗時 (ms)塊數(shù) 線程數(shù) x 116 塊548592塊數(shù) 線程數(shù) x 232 塊466545塊數(shù) 線程數(shù) x 464 塊430610塊數(shù) 線程數(shù) x 8128 塊408715塊數(shù) 線程數(shù) x 32512 塊475830transform 階段塊數(shù)太少時8 個元素少的線程很快做完剩下幾個線程在跑長尾負載不均衡塊數(shù)太多時每塊實際只做幾萬次浮點運算調(diào)度和 future 同步開銷抬頭耗時重新上升。所以 transform 場景里塊數(shù)取線程數(shù)的 4 到 8 倍是個比較好的甜點。排序階段則明顯不一樣塊數(shù)越少越穩(wěn)定因為歸并代價在塊數(shù)增多后快速上漲。512 塊時要進行 9 輪歸并最后幾輪幾乎每輪都要搬運全部 480MB 數(shù)據(jù)內(nèi)存帶寬都耗在歸并上。所以排序場景我建議塊數(shù)控制在線程數(shù)的 1 到 2 倍。4.3 隊列鎖爭用與緩存行帶來的意外開銷當我按“每個元素單獨提交一個任務(wù)”這種極端方式寫并行時提交 1.2 億個任務(wù)顯然不現(xiàn)實但即使只提交幾萬個任務(wù)線程池的互斥鎖也會成為熱點。用perf看線程采樣能看到大量時間花在std::mutex::lock的等待和 futex 喚醒上。解決辦法就是前面說的批量提交接口和合理塊大小。另一個容易被忽略的是緩存行偽共享。我在 3.2 節(jié)的統(tǒng)計例子里用了一個std::vectorint64_t partials(blocks, 0)保存各塊結(jié)果。16 個int64_t緊挨著放在內(nèi)存里很可能占用兩三個緩存行不同線程頻繁寫入相鄰地址導致緩存行在核間反復橫跳。將partials數(shù)組的元素對齊到 64 字節(jié)后統(tǒng)計階段耗時從 520ms 降到大約 440ms。這個優(yōu)化沒有任何算法改動只是內(nèi)存布局更友好了。對齊方式很簡單定義一個帶alignas(64)的結(jié)構(gòu)體struct alignas(64) PaddedCounter { int64_t value 0; }; std::vectorPaddedCounter partials(blocks);注意alignas(64)讓每個元素占滿一個緩存行就不會和鄰居互相污染。塊特別多時內(nèi)存會多花一些但對大多數(shù)項目成本可忽略。5. 避坑記錄死鎖、偽共享與過度并行5.1 任務(wù)內(nèi)嵌任務(wù)死鎖的真實案例我第一次把 ranges 管道接進線程池時寫過一個很自然的遞歸分治邏輯外層任務(wù)把 range 切成兩半然后向同一個線程池提交兩個子任務(wù)再future.wait()等子任務(wù)完成。結(jié)果程序在 16 個線程全忙時直接掛起。原因很經(jīng)典16 個 worker 都在執(zhí)行某個父任務(wù)每個父任務(wù)都在等待自己派生的子任務(wù)但子任務(wù)排在隊列后面沒有任何線程去取形成互相等待。解決方式有幾種。最簡單的是任務(wù)內(nèi)部不要再同步等待同一個線程池的任務(wù)改為把父任務(wù)自身拆成多個獨立任務(wù)最后統(tǒng)一在外部等待。如果一定要在任務(wù)內(nèi)部遞歸那就不能把所有 worker 都占滿預(yù)留一個線程專門處理派生任務(wù)或者用“當前線程參與執(zhí)行隊列任務(wù)”的方式不讓父任務(wù)死等。從工程角度我會優(yōu)先選擇前者結(jié)構(gòu)清晰也不容易踩邊界。5.2 先分塊再合并而不是靠原子變量到處加我在并行統(tǒng)計里犯過一個更隱蔽的錯每個線程直接用std::atomicint64_t total.fetch_add(localCount)累加全局計數(shù)。表面上看沒有數(shù)據(jù)競爭結(jié)果卻慢得離譜因為這本質(zhì)上就是所有線程對同一個原子變量反復做 RMW 操作緩存行被反復在不同核心間搬運。統(tǒng)計階段耗時比單線程還高。把局部結(jié)果先寫到獨立數(shù)組、最后合并看起來繞了一圈實際是最快的。這也是歸約型算法教科書式的做法局部歸約、全局合并。要可靠記住這一點最簡單的方法是把“原子操作”理解為一把昂貴的隱式鎖它能保證正確性但保證不了性能。5.3 并行度不是越高越好內(nèi)存帶寬和調(diào)度總有過剩時最后一個坑是“線程越多越快”的錯覺。當任務(wù)變成純內(nèi)存搬運或浮點變換時瓶頸往往落在內(nèi)存帶寬上。比如 480MB 數(shù)據(jù)要被遍歷多輪即便線程數(shù)從 8 加到 32CPU 計算單元可能早就等內(nèi)存數(shù)據(jù)了增加的線程只是增加競爭和切換。我在排序測試里看到過非常明顯的現(xiàn)象8 線程排序 620ms16 線程反而 750ms就是因為多線程并行排序后的歸并階段需要反復讀寫全部數(shù)據(jù)內(nèi)存帶寬已經(jīng)飽和額外線程毫無幫助?？紤]到這一點我的建議是給線程池留一個運行時可控的并發(fā)上限比如先跑一輪小規(guī)模預(yù)熱測試用數(shù)據(jù)決定線程數(shù)而不是盲信hardware_concurrency。你可以把塊數(shù)和線程數(shù)都作為可配置參數(shù)壓測雙維度組合后固定到配置文件里這樣不同機器上都能快速獲得接近最優(yōu)的性能。另外要提醒的是Ranges 視圖在單線程下惰性求值很安全但一旦切塊并行捕獲的引用和可變狀態(tài)就必須重新審視。比如某個 lambda 里捕獲了局部共享的臨時容器多個線程同時修改就是典型的數(shù)據(jù)競爭。并行化前先自問一句每個塊是否只訪問自己范圍內(nèi)的數(shù)據(jù)如果答案不明確不要急著上線程池先把數(shù)據(jù)處理成無共享的結(jié)構(gòu)。我在這次重構(gòu)里的一個非常直接的體會是std::ranges的價值在于把算法意圖寫清楚而開發(fā)者自己控制的線程池和工作隊列則把這些意圖真正轉(zhuǎn)化為硬件利用率。兩者結(jié)合比單純依賴std::execution::par更能貼近業(yè)務(wù)需求。最后分享一個操作層面的小技巧如果不想把所有 future 攢在內(nèi)存里可以像我在 2.3 節(jié)里寫的那樣用原子計數(shù)加條件變量代替 future 列表實測下來在任務(wù)數(shù)超過一萬時這種方式對內(nèi)存占用和喚醒開銷都更友好。希望這套“拆分、調(diào)度、合并”的思路能幫你在自己的項目里少走幾條彎路。 std::ranges并行化：自建線程池與工作隊列的實踐
最近我在重構(gòu)一個內(nèi)部數(shù)據(jù)處理模塊核心管道寫得非常漂亮一段std::ranges視圖經(jīng)過filter、transform把非結(jié)構(gòu)化字段規(guī)整成干凈的結(jié)構(gòu)體最后落到ranges::sort做時序排列?？尚阅軠y試一出來就有點尷尬這些 ranges 算法全在單線程上跑機器上的多核基本在看熱鬧。想把并行執(zhí)行做進去讓 ranges 算法配合自己的線程池與工作隊列結(jié)果發(fā)現(xiàn)標準庫給出的支持比想象中少得多。這篇文章就是我在反復折騰后整理出來的一套落地思路如何拆范圍、建隊列、控線程數(shù)讓 range 管道真正吃滿多核順便把踩過的坑都交代清楚。1. std::ranges 并行這件事標準到底給了什么1.1 ranges 算法很香但為什么等不到并行重載std::ranges算法是 C20 帶來的最大語法糖之一。以往我們寫排序、變換、查找都要小心翼翼地傳迭代器對代碼一長就變成“迭代器地獄”。有了 ranges 之后std::ranges::sort(data)、std::ranges::for_each(v, fn)這種寫法不僅簡短而且能配合 view 做惰性求值和管道拼接表達力直接上一個臺階。但有一個非?，F(xiàn)實的坑標準庫的 ranges 算法并不帶并行執(zhí)行策略重載。至少到 C23 標準std::ranges::sort、std::ranges::transform這些算法都沒有接受std::execution::par的版本。你可能會寫出下面這種代碼然后發(fā)現(xiàn)編譯期直接報錯// 這只是理想寫法目前標準庫中并不存在這樣的重載 std::ranges::sort(std::execution::par, data); // 編譯不過 std::ranges::transform(std::execution::par, input, output, fn); // 編譯不過也就是說標題里那個“ranges 算法并行執(zhí)行”并不能靠標準庫一個參數(shù)白拿。想并行要么退回傳統(tǒng)迭代器算法加std::execution::par要么自己包一層調(diào)度把 range 按塊拆開派發(fā)給工作線程。委員會沒有把兩者接通原因比較復雜核心是 ranges 和后來的 executors 方向還沒有合并成穩(wěn)定的統(tǒng)一模型std::execution::par只是早期執(zhí)行策略形態(tài)真正的調(diào)度器體系直到后來的提案才逐漸成型。作為使用者我們不能干等標準最好的路線是自己把調(diào)度層補出來。1.2 標準并行執(zhí)行策略的真實工作方式C17 提供了四個執(zhí)行策略標簽seq串行、unseq向量化、par多線程、par_unseq線程加向量化。它的用法集中在傳統(tǒng)算法上std::sort(std::execution::par, data.begin(), data.end()); std::for_each(std::execution::par, data.begin(), data.end(), fn);它會“盡可能”并行但標準對并行粒度、線程數(shù)量、任務(wù)隊列完全不規(guī)定。也就是說并行行為完全取決于實現(xiàn)內(nèi)部的后端有的實現(xiàn)會用自己的一套任務(wù)調(diào)度器有的實現(xiàn)會綁硬件線程數(shù)做并行分塊有的是在第一次調(diào)用時初始化一個全局線程池。問題也藏在這里。這套后端對你是不可見的你沒法告訴它“我這個項目已經(jīng)有線程池了請復用”也沒法控制“我今天只想用 4 個線程不要跑滿 16 個邏輯核”更沒法在業(yè)務(wù)高峰期控制隊列深度。當一個模塊里同時出現(xiàn)多個并行算法調(diào)用時它們各自獨立調(diào)度線程數(shù)翻倍、CPU 超訂是常有的事。所以我一直認為std::execution::par適合“粗粒度地加速單次算法調(diào)用”的場景但如果你的項目里本身就有線程池或者希望精準控制任務(wù)隊列那必須自己接管調(diào)度層。這正好是本文標題里“線程池管理與工作隊列優(yōu)化”存在的價值。1.3 不讓標準庫替你調(diào)度那你需要解決什么問題當你不使用std::execution::par而是想自己控制并行時核心要解決四個問題并行度控制用多少線程線程數(shù)是否動態(tài)可變。任務(wù)切分range 那么大按什么粒度切成塊塊數(shù)多少合適。任務(wù)派發(fā)與回收塊切好了怎么把每個塊作為任務(wù)丟進隊列線程執(zhí)行完怎么通知主線程。結(jié)果合并每個線程處理完得到的是局部結(jié)果如何高效合成為全局結(jié)果。這四個問題聽起來多其實拆開就是一套線程池加一套分塊策略。下面我先把線程池骨架寫出來再補分塊和合并的完整玩法。2. 自建線程池隊列、線程數(shù)和提交接口的設(shè)計2.1 一個能直接跑起來的最小線程池先給一個最小但完整的線程池實現(xiàn)。它不算最優(yōu)但足夠作為后續(xù)優(yōu)化的起點#include condition_variable #include deque #include future #include mutex #include thread #include vector class ThreadPool { public: explicit ThreadPool(size_t threadCount) : stop_(false) { for (size_t i 0; i threadCount; i) { workers_.emplace_back([this] { for (;;) { std::packaged_taskvoid() job; { std::unique_lockstd::mutex lock(mutex_); cv_.wait(lock, [this] { return stop_ || !queue_.empty(); }); if (stop_ queue_.empty()) return; job std::move(queue_.front()); queue_.pop_front(); } job(); } }); } } template typename F auto submit(F func) - std::futurevoid { std::packaged_taskvoid() task(std::forwardF(func)); auto future task.get_future(); { std::lock_guardstd::mutex lock(mutex_); queue_.push_back(std::move(task)); } cv_.notify_one(); return future; } ~ThreadPool() { { std::lock_guardstd::mutex lock(mutex_); stop_ true; } cv_.notify_all(); for (auto worker : workers_) { worker.join(); } } private: std::atomicbool stop_; std::dequestd::packaged_taskvoid() queue_; std::mutex mutex_; std::condition_variable cv_; std::vectorstd::thread workers_; };這個類值得注意的細節(jié)有三個為什么用std::packaged_taskvoid()而不是std::functionvoid()packaged_task自帶future提交時就能拿到結(jié)果句柄省去再包一層shared_ptr的麻煩。worker 執(zhí)行job()時future狀態(tài)會被置為 ready調(diào)用方通過future.wait()或者future.get()回收結(jié)果。為什么析構(gòu)時先加鎖再notify_all因為 worker 循環(huán)里對停止狀態(tài)和隊列空判斷同時發(fā)生在鎖內(nèi)。如果不加鎖直接設(shè)置stop_可能出現(xiàn) worker 剛判完queue_.empty()主線程就設(shè)置停止并通知worker 醒來后看到隊列已空才退出邏輯上沒問題但 stop 標志寫入和讀取最好由同一把鎖保護避免晦澀的競態(tài)。為什么隊列用std::deque而不是std::vectordeque支持兩端高效操作。雖然我們目前只從尾部入隊、頭部出隊但后續(xù)如果要做 LIFO 策略新任務(wù)優(yōu)先執(zhí)行只需把push_back改成push_front不用換容器。這個實現(xiàn)里submit返回std::futurevoid對大多數(shù)“數(shù)據(jù)分塊處理、最終各塊互不相干”的任務(wù)完全夠用。如果后面需要每個塊返回一個具體值改成std::futureT版本即可下面會講到。2.2 線程數(shù)量不是越多越好線程池最容易被忽略的參數(shù)是線程數(shù)。很多人直接寫std::thread::hardware_concurrency()然后期望性能拉滿。實測下來這條規(guī)則對純 CPU 密集任務(wù)是“半個坑”。hardware_concurrency返回的是邏輯處理器數(shù)量包含超線程。在 8 核 16 線程的機器上它會返回 16但如果你跑的是純浮點計算16 個“邏輯處理器”共享物理核心的執(zhí)行單元線程開滿反而帶來上下文切換和調(diào)度開銷。我對純 CPU 密集任務(wù)的經(jīng)驗是先從物理核數(shù)或邏輯核數(shù)的一半起步跑一輪實測再向上調(diào)。遇到任務(wù)里有 IO 等待、鎖等待、系統(tǒng)調(diào)用時可以適當加線程數(shù)讓部分線程在等待間隙不阻塞整體進度。還有一個隱藏因素任務(wù)塊的數(shù)量和線程數(shù)的比值。這個比值直接決定調(diào)度開銷的占比。比如一塊數(shù)據(jù) 1.2 億個元素開 8 個線程每個線程處理 1500 萬個元素單塊執(zhí)行時間可能幾十毫秒調(diào)度開銷可忽略。但如果數(shù)據(jù)只有 10 萬元素同樣開 8 個線程每個塊只有一萬多元素單塊執(zhí)行時間可能只有幾十微秒那么隊列鎖、notify_one、future 狀態(tài)同步的開銷就可能占到 20% 以上。具體怎么定塊大小下面第四章我會給一組實測數(shù)據(jù)。2.3 提交接口從單任務(wù)到批量入隊上面的submit一次只能投遞一個任務(wù)每次都要加鎖、解鎖、notify_one。當任務(wù)塊數(shù)量達到幾千甚至幾萬時鎖競爭會成為瓶頸。一個很實用的優(yōu)化是增加批量提交接口template typename F void submitBatch(std::vectorF tasks) { { std::lock_guardstd::mutex lock(mutex_); for (auto task : tasks) { std::packaged_taskvoid() wrapped(std::move(task)); queue_.push_back(std::move(wrapped)); } } cv_.notify_all(); }批量提交把 N 次加鎖變成一次加鎖然后一次性喚醒所有 idle 線程。你可能覺得“一次提交 32 個任務(wù)還省什么鎖”但如果任務(wù)塊數(shù)上千每次 submit 都鎖同一個 mutex等待時間會迅速累積。我在實際工程里遇到的一個場景是對 1.2 億條日志按每 1024 條分塊分出一萬多個任務(wù)逐個 submit 的那一版光提交階段就消耗了接近 200ms改成批量化插入后提交階段縮到 30ms 以內(nèi)。另一個值得做的是去掉 future 的開銷。如果任務(wù)沒有返回值你只需要知道“全部執(zhí)行完了”保留一堆futurevoid其實有些浪費。可以用一個原子計數(shù)器加條件變量任務(wù)完成后計數(shù)器遞減最后一個完工的任務(wù)通知主線程。這樣既省 future 列表的內(nèi)存也省逐個 wait 的循環(huán)。std::atomicsize_t pending{0}; std::condition_variable doneCv; std::mutex doneMutex; void pushTask(std::packaged_taskvoid() task) { { std::lock_guardstd::mutex lock(mutex_); pending.fetch_add(1); queue_.push_back(std::move(task)); } cv_.notify_one(); } void onTaskFinished() { if (pending.fetch_sub(1) 1) { std::lock_guardstd::mutex lock(doneMutex); doneCv.notify_all(); } } void waitAll() { std::unique_lockstd::mutex lock(doneMutex); doneCv.wait(lock, [] { return pending.load() 0; }); }注意務(wù)必在任務(wù)內(nèi)部調(diào)用onTaskFinished()并確保異常安全。如果任務(wù)拋異常計數(shù)會丟失waitAll永遠等不到零。穩(wěn)妥做法是用 RAII 包一層析構(gòu)時遞減計數(shù)。3. 把 ranges 算法拆到工作隊列上三種模式與代碼落地線程池有了接下來就是把std::ranges算法真正派發(fā)上去。不同算法有不同的并行模式不能一把梭。我把常見場景分成三類映射型、歸約型、全局依賴型。3.1 映射型任務(wù)for_each / transform最基礎(chǔ)也最容易并行化的是for_each、transform這類逐元素操作塊與塊之間沒有任何數(shù)據(jù)依賴。核心思路把 range 切成連續(xù)塊每塊獨立調(diào)用ranges::for_each或ranges::transform。template typename Range, typename Fn void parallelTransformInplace(ThreadPool pool, Range range, Fn fn) { using std::ranges::begin; using std::ranges::end; size_t n std::ranges::distance(range); if (n 0) return; const size_t hw std::thread::hardware_concurrency(); const size_t blocks hw * 4; // 塊數(shù)約為線程數(shù) 4 倍 const size_t chunk (n blocks - 1) / blocks; // 每塊元素數(shù) auto it begin(range); std::vectorstd::futurevoid futures; futures.reserve(blocks); for (size_t offset 0; offset n; offset chunk) { auto blockBegin std::next(it, offset); auto blockEnd std::next(blockBegin, std::min(chunk, n - offset)); futures.push_back(pool.submit([blockBegin, blockEnd, fn]() { std::ranges::for_each(blockBegin, blockEnd, fn); })); } for (auto f : futures) { f.wait(); } }這個實現(xiàn)有個前提range 是隨機訪問范圍。std::next(it, offset)在隨機訪問迭代器上 O(1)但如果底層是filter_view、istream_view每次推進都要遍歷元素切塊的代價會很高。我的做法是先把非隨機訪問范圍 materialize 成 vector再做并行分塊。比如從文件讀出的日志行經(jīng)過filter過濾后可能只有 30% 通過直接按行號分塊容易導致負載不均先拷進 vector 再分塊反而更穩(wěn)定。對于transform要注意函數(shù)副作用。如果變換函數(shù)修改的是容器外共享狀態(tài)比如同一個計數(shù)器那并行時每個線程都在競爭結(jié)果既可能錯誤也會變慢。正確姿勢是內(nèi)部無狀態(tài)或只修改自己塊內(nèi)的元素。3.2 歸約型任務(wù)并行局部結(jié)果再合并求和、找最大值、統(tǒng)計滿足條件的數(shù)量這類任務(wù)的并行模式是“局部歸約 全局合并”。每個線程獨立計算自己的那塊最后匯總。很典型的例子是統(tǒng)計std::ranges::filter之后的元素數(shù)量template typename Range, typename Pred int64_t parallelCountIf(ThreadPool pool, const Range range, Pred pred) { auto n std::ranges::distance(range); if (n 0) return 0; const size_t hw std::thread::hardware_concurrency(); const size_t blocks hw * 4; const size_t chunk (n blocks - 1) / blocks; std::vectorint64_t partials(blocks, 0); std::vectorstd::futurevoid futures; auto it std::ranges::begin(range); for (size_t b 0; b blocks; b) { auto blockBegin std::next(it, b * chunk); auto blockEnd std::next(blockBegin, std::min(chunk, n - b * chunk)); futures.push_back(pool.submit([blockBegin, blockEnd, pred, partials, b]() { int64_t count 0; for (auto iter blockBegin; iter ! blockEnd; iter) { if (pred(*iter)) count; } partials[b] count; })); } for (auto f : futures) f.wait(); return std::accumulate(partials.begin(), partials.end(), int64_t{0}); }這里關(guān)鍵點是不要讓每個線程原子地累加一個全局計數(shù)器。原子fetch_add在溢出一級緩存后會形成 cacheline 乒乓性能遠不如各算各的、最后一次性合并。局部partials數(shù)組本身也可能踩偽共享的坑這點我放到第四章專門講。另一個技巧是如果任務(wù)塊內(nèi)還需要繼續(xù)用 ranges 表達比如auto view subRange | std::views::filter(pred);完全可以在 lambda 里構(gòu)建局部視圖讓代碼保持和單線程版本一樣的可讀性。并行化只是把外層切塊內(nèi)部該用什么算法就用什么算法。3.3 全局依賴型任務(wù)排序的分塊與歸并排序比前面兩種復雜因為全數(shù)組最終必須有序塊與塊之間不是無關(guān)的。最經(jīng)典的并行策略是分塊并行排序再做多路歸并。template typename T void parallelSort(std::vectorT data, ThreadPool pool) { const size_t n data.size(); if (n 10000) { std::ranges::sort(data); return; } const size_t hw std::thread::hardware_concurrency(); const size_t blockCount hw * 2; // 排序場景塊數(shù)不宜過多 const size_t chunk (n blockCount - 1) / blockCount; std::vectorstd::futurevoid futures; for (size_t i 0; i n; i chunk) { auto begin data.begin() i; auto end begin std::min(chunk, n - i); futures.push_back(pool.submit([begin, end]() { std::ranges::sort(begin, end); // 每個塊內(nèi)部先有序 })); } for (auto f : futures) f.wait(); // 多輪歸并 for (size_t width chunk; width n; width * 2) { for (size_t left 0; left n; left 2 * width) { auto mid data.begin() std::min(left width, n); auto last data.begin() std::min(left 2 * width, n); std::inplace_merge(data.begin() left, mid, last); } } }注意這里歸并階段還是串行的塊數(shù)越多歸并的輪數(shù)和成本越高所以排序時塊數(shù)不能像映射型任務(wù)那樣取太多。我實際測下來排序場景的塊數(shù)取線程數(shù)的 1 到 2 倍效果最好再多就會讓歸并成為新瓶頸。想進一步并行化歸并階段可以對兩個有序段做二分切分找到左右兩個子問題后并行歸并但實現(xiàn)復雜度高不少工作量需要和收益匹配。如果你的數(shù)據(jù)不是簡單 vector而是經(jīng)過views::filter或views::transform的視圖那并行排序前必須先 materialize。因為排序需要隨機訪問和原地交換視圖往往無法滿足。我的做法是std::vectorT tmp; if constexpr (std::ranges::random_access_rangeRange std::ranges::sized_rangeRange) { // 直接對目標 range 分塊 } else { tmp.assign(range.begin(), range.end()); // 對 tmp 排序若需要再寫回原容器 }4. 實測數(shù)據(jù)塊大小、線程數(shù)和隊列鎖的交叉影響4.1 測試場景與基線數(shù)據(jù)我在一臺 8 核 16 線程的 Linux 機器上做了組對照實驗。模擬項目X的數(shù)據(jù)集是 1.2 億個float總體數(shù)據(jù)量約 480MB。任務(wù)分三段先對每個元素做一次非線性變換再統(tǒng)計超過閾值的樣本數(shù)最后對結(jié)果排序輸出前 10000 個。單線程版本用純std::ranges管道耗時約 2180ms。傳統(tǒng)標準庫并行算法std::for_each(std::execution::par, ...)跑完約 640ms。自建線程池在參數(shù)沒調(diào)好時甚至跑到 900ms 以上調(diào)好之后降到 410ms。差距很大值得把參數(shù)關(guān)系展開講。4.2 塊數(shù)與線程數(shù)的交叉測試下面這組數(shù)據(jù)是同一批數(shù)據(jù)、同樣 16 個線程情況下不同塊數(shù)對耗時的影響配置transform 階段耗時 (ms)排序階段耗時 (ms)塊數(shù) 線程數(shù) x 116 塊548592塊數(shù) 線程數(shù) x 232 塊466545塊數(shù) 線程數(shù) x 464 塊430610塊數(shù) 線程數(shù) x 8128 塊408715塊數(shù) 線程數(shù) x 32512 塊475830transform 階段塊數(shù)太少時8 個元素少的線程很快做完剩下幾個線程在跑長尾負載不均衡塊數(shù)太多時每塊實際只做幾萬次浮點運算調(diào)度和 future 同步開銷抬頭耗時重新上升。所以 transform 場景里塊數(shù)取線程數(shù)的 4 到 8 倍是個比較好的甜點。排序階段則明顯不一樣塊數(shù)越少越穩(wěn)定因為歸并代價在塊數(shù)增多后快速上漲。512 塊時要進行 9 輪歸并最后幾輪幾乎每輪都要搬運全部 480MB 數(shù)據(jù)內(nèi)存帶寬都耗在歸并上。所以排序場景我建議塊數(shù)控制在線程數(shù)的 1 到 2 倍。4.3 隊列鎖爭用與緩存行帶來的意外開銷當我按“每個元素單獨提交一個任務(wù)”這種極端方式寫并行時提交 1.2 億個任務(wù)顯然不現(xiàn)實但即使只提交幾萬個任務(wù)線程池的互斥鎖也會成為熱點。用perf看線程采樣能看到大量時間花在std::mutex::lock的等待和 futex 喚醒上。解決辦法就是前面說的批量提交接口和合理塊大小。另一個容易被忽略的是緩存行偽共享。我在 3.2 節(jié)的統(tǒng)計例子里用了一個std::vectorint64_t partials(blocks, 0)保存各塊結(jié)果。16 個int64_t緊挨著放在內(nèi)存里很可能占用兩三個緩存行不同線程頻繁寫入相鄰地址導致緩存行在核間反復橫跳。將partials數(shù)組的元素對齊到 64 字節(jié)后統(tǒng)計階段耗時從 520ms 降到大約 440ms。這個優(yōu)化沒有任何算法改動只是內(nèi)存布局更友好了。對齊方式很簡單定義一個帶alignas(64)的結(jié)構(gòu)體struct alignas(64) PaddedCounter { int64_t value 0; }; std::vectorPaddedCounter partials(blocks);注意alignas(64)讓每個元素占滿一個緩存行就不會和鄰居互相污染。塊特別多時內(nèi)存會多花一些但對大多數(shù)項目成本可忽略。5. 避坑記錄死鎖、偽共享與過度并行5.1 任務(wù)內(nèi)嵌任務(wù)死鎖的真實案例我第一次把 ranges 管道接進線程池時寫過一個很自然的遞歸分治邏輯外層任務(wù)把 range 切成兩半然后向同一個線程池提交兩個子任務(wù)再future.wait()等子任務(wù)完成。結(jié)果程序在 16 個線程全忙時直接掛起。原因很經(jīng)典16 個 worker 都在執(zhí)行某個父任務(wù)每個父任務(wù)都在等待自己派生的子任務(wù)但子任務(wù)排在隊列后面沒有任何線程去取形成互相等待。解決方式有幾種。最簡單的是任務(wù)內(nèi)部不要再同步等待同一個線程池的任務(wù)改為把父任務(wù)自身拆成多個獨立任務(wù)最后統(tǒng)一在外部等待。如果一定要在任務(wù)內(nèi)部遞歸那就不能把所有 worker 都占滿預(yù)留一個線程專門處理派生任務(wù)或者用“當前線程參與執(zhí)行隊列任務(wù)”的方式不讓父任務(wù)死等。從工程角度我會優(yōu)先選擇前者結(jié)構(gòu)清晰也不容易踩邊界。5.2 先分塊再合并而不是靠原子變量到處加我在并行統(tǒng)計里犯過一個更隱蔽的錯每個線程直接用std::atomicint64_t total.fetch_add(localCount)累加全局計數(shù)。表面上看沒有數(shù)據(jù)競爭結(jié)果卻慢得離譜因為這本質(zhì)上就是所有線程對同一個原子變量反復做 RMW 操作緩存行被反復在不同核心間搬運。統(tǒng)計階段耗時比單線程還高。把局部結(jié)果先寫到獨立數(shù)組、最后合并看起來繞了一圈實際是最快的。這也是歸約型算法教科書式的做法局部歸約、全局合并。要可靠記住這一點最簡單的方法是把“原子操作”理解為一把昂貴的隱式鎖它能保證正確性但保證不了性能。5.3 并行度不是越高越好內(nèi)存帶寬和調(diào)度總有過剩時最后一個坑是“線程越多越快”的錯覺。當任務(wù)變成純內(nèi)存搬運或浮點變換時瓶頸往往落在內(nèi)存帶寬上。比如 480MB 數(shù)據(jù)要被遍歷多輪即便線程數(shù)從 8 加到 32CPU 計算單元可能早就等內(nèi)存數(shù)據(jù)了增加的線程只是增加競爭和切換。我在排序測試里看到過非常明顯的現(xiàn)象8 線程排序 620ms16 線程反而 750ms就是因為多線程并行排序后的歸并階段需要反復讀寫全部數(shù)據(jù)內(nèi)存帶寬已經(jīng)飽和額外線程毫無幫助?？紤]到這一點我的建議是給線程池留一個運行時可控的并發(fā)上限比如先跑一輪小規(guī)模預(yù)熱測試用數(shù)據(jù)決定線程數(shù)而不是盲信hardware_concurrency。你可以把塊數(shù)和線程數(shù)都作為可配置參數(shù)壓測雙維度組合后固定到配置文件里這樣不同機器上都能快速獲得接近最優(yōu)的性能。另外要提醒的是Ranges 視圖在單線程下惰性求值很安全但一旦切塊并行捕獲的引用和可變狀態(tài)就必須重新審視。比如某個 lambda 里捕獲了局部共享的臨時容器多個線程同時修改就是典型的數(shù)據(jù)競爭。并行化前先自問一句每個塊是否只訪問自己范圍內(nèi)的數(shù)據(jù)如果答案不明確不要急著上線程池先把數(shù)據(jù)處理成無共享的結(jié)構(gòu)。我在這次重構(gòu)里的一個非常直接的體會是std::ranges的價值在于把算法意圖寫清楚而開發(fā)者自己控制的線程池和工作隊列則把這些意圖真正轉(zhuǎn)化為硬件利用率。兩者結(jié)合比單純依賴std::execution::par更能貼近業(yè)務(wù)需求。最后分享一個操作層面的小技巧如果不想把所有 future 攢在內(nèi)存里可以像我在 2.3 節(jié)里寫的那樣用原子計數(shù)加條件變量代替 future 列表實測下來在任務(wù)數(shù)超過一萬時這種方式對內(nèi)存占用和喚醒開銷都更友好。希望這套“拆分、調(diào)度、合并”的思路能幫你在自己的項目里少走幾條彎路。