線程池是通過預(yù)先創(chuàng)建并維護一組線程來提高任務(wù)執(zhí)行效率的機制。1. 核心組件包括任務(wù)隊列、工作線程和線程池管理器，其中任務(wù)隊列用于存儲待執(zhí)行任務(wù)，工作線程負責執(zhí)行任務(wù)，管理器負責線程池的生命周期和任務(wù)調(diào)度。2. 線程池大小應(yīng)根據(jù)任務(wù)類型和系統(tǒng)資源合理設(shè)置：cpu密集型任務(wù)建議設(shè)為cpu核心數(shù)+1，i/o密集型任務(wù)可依據(jù)公式“線程數(shù) = cpu核心數(shù) × (1 + i/o等待時間 / cpu計算時間)”設(shè)定。3. 異常處理可通過在工作線程中捕獲異常并記錄日志實現(xiàn)，防止程序崩潰。4. 優(yōu)雅關(guān)閉線程池需通知線程停止接收新任務(wù)并在完成當前任務(wù)后退出，析構(gòu)函數(shù)中設(shè)置stop標志并喚醒所有線程以確保其退出循環(huán)，但如需確保所有任務(wù)執(zhí)行完畢，還需額外機制支持。
線程池，簡單來說，就是預(yù)先創(chuàng)建好一批線程，讓它們在那里“待命”，而不是每次需要執(zhí)行任務(wù)的時候才臨時創(chuàng)建。這樣做的好處顯而易見：避免了頻繁創(chuàng)建和銷毀線程的開銷，提高了程序的響應(yīng)速度和效率。
解決方案
要在c++中實現(xiàn)線程池，我們需要考慮以下幾個核心組件：
立即學(xué)習(xí)“C++免費學(xué)習(xí)筆記（深入）”；
- 任務(wù)隊列 (Task Queue): 用于存放待執(zhí)行的任務(wù)。通常使用線程安全的隊列，例如std::queue配合std::mutex和std::condition_variable。
- 工作線程 (Worker Threads): 負責從任務(wù)隊列中取出任務(wù)并執(zhí)行。這些線程在線程池創(chuàng)建時啟動，并在線程池銷毀時停止。
- 線程池管理器 (Thread Pool Manager): 負責線程池的創(chuàng)建、銷毀、任務(wù)提交等管理工作。
下面是一個簡單的C++線程池實現(xiàn)示例：
#include <iostream> #include <vector> #include <queue> #include <thread> #include <mutex> #include <condition_variable> #include <functional> class ThreadPool { public: ThreadPool(size_t numThreads) : stop(false) { threads.resize(numThreads); for (size_t i = 0; i < numThreads; ++i) { threads[i] = std::thread([this]() { while (true) { std::function<void()> task; { std::unique_lock<std::mutex> lock(queueMutex); condition.wait(lock, [this]() { return stop || !tasks.empty(); }); if (stop && tasks.empty()) { return; } task = tasks.front(); tasks.pop(); } task(); } }); } } template<typename F, typename... Args> void enqueue(F&& f, Args&&... args) { std::function<void()> task = std::bind(std::forward<F>(f), std::forward<Args>(args)...); { std::unique_lock<std::mutex> lock(queueMutex); tasks.push(task); } condition.notify_one(); } ~ThreadPool() { { std::unique_lock<std::mutex> lock(queueMutex); stop = true; } condition.notify_all(); for (std::thread& thread : threads) { thread.join(); } } private: std::vector<std::thread> threads; std::queue<std::function<void()>> tasks; std::mutex queueMutex; std::condition_variable condition; bool stop; }; int main() { ThreadPool pool(4); // 創(chuàng)建一個包含4個線程的線程池 for (int i = 0; i < 8; ++i) { pool.enqueue([i]() { std::cout << "Task " << i << " executed by thread " << std::this_thread::get_id() << std::endl; std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 模擬耗時操作 }); } std::this_thread::sleep_for(std::chrono::seconds(1)); // 等待任務(wù)完成 return 0; }
線程池大小如何確定？
線程池的大小并不是越大越好。過多的線程反而會增加上下文切換的開銷，降低效率。合理的線程池大小取決于任務(wù)的類型和系統(tǒng)的資源。
- CPU密集型任務(wù)： 線程數(shù)可以設(shè)置為CPU核心數(shù) + 1。 額外的線程可以在某個線程阻塞時，讓另一個線程繼續(xù)執(zhí)行，提高CPU的利用率。
- I/O密集型任務(wù)： 線程數(shù)可以設(shè)置得更大一些，因為線程在等待I/O時不會占用CPU。一個常用的公式是：線程數(shù) = CPU核心數(shù) * (1 + I/O等待時間 / CPU計算時間)。
實際應(yīng)用中，最好通過性能測試來確定最佳的線程池大小。
如何處理任務(wù)執(zhí)行過程中拋出的異常？
在線程池中執(zhí)行的任務(wù)可能會拋出異常，如果不處理，可能會導(dǎo)致程序崩潰。一種常見的做法是在工作線程中捕獲異常，并進行適當?shù)娜罩居涗浕蝈e誤處理。
修改上面的ThreadPool類，可以在工作線程的循環(huán)中添加異常處理：
threads[i] = std::thread([this]() { while (true) { std::function<void()> task; { std::unique_lock<std::mutex> lock(queueMutex); condition.wait(lock, [this]() { return stop || !tasks.empty(); }); if (stop && tasks.empty()) { return; } task = tasks.front(); tasks.pop(); } try { task(); } catch (const std::exception& e) { std::cerr << "Exception caught in thread: " << e.what() << std::endl; // 可以選擇將異常信息記錄到日志中，或者進行其他錯誤處理 } catch (...) { std::cerr << "Unknown exception caught in thread" << std::endl; } } });
如何優(yōu)雅地關(guān)閉線程池？
直接殺死線程可能會導(dǎo)致數(shù)據(jù)丟失或程序狀態(tài)不一致。正確的做法是通知工作線程停止接受新任務(wù)，并在完成當前任務(wù)后退出。
在ThreadPool的析構(gòu)函數(shù)中，我們設(shè)置了stop標志，并通知所有等待的線程。這樣，當線程池銷毀時，工作線程會檢查stop標志，如果為真，則退出循環(huán)。
但是，如果任務(wù)隊列中還有未執(zhí)行的任務(wù)，工作線程可能會在退出前繼續(xù)執(zhí)行這些任務(wù)。為了確保所有任務(wù)都已完成，可以在析構(gòu)函數(shù)中添加一個等待所有任務(wù)完成的機制。這可以通過使用std::future來實現(xiàn)，但這會增加代碼的復(fù)雜性。 簡單起見，上面的代碼只是確保線程能夠退出，但沒有強制等待所有任務(wù)完成。在實際應(yīng)用中，需要根據(jù)具體需求進行調(diào)整。
