C++緩存線程池CachedThreadPool原理、實(shí)現(xiàn)與對比解析
更新時(shí)間：2026年05月14日 09:47:51 作者：CodeLens75842
在C++中實(shí)現(xiàn)一個(gè)緩存線程池（CachedThreadPool）是一種常用的并發(fā)編程模式,特別是在需要處理大量短生命周期任務(wù)的場景中,本文給大家介紹C++?緩存線程池（CachedThreadPool）原理、實(shí)現(xiàn)、對比分析,感興趣的朋友跟隨小編一起看看吧
前言
在高并發(fā)編程場景中，線程池是提升程序性能、降低資源消耗的核心組件。不同于固定大小的線程池，緩存線程池（CachedThreadPool） 能夠根據(jù)任務(wù)量動(dòng)態(tài)調(diào)整線程數(shù)量 —— 任務(wù)繁忙時(shí)創(chuàng)建新線程，空閑時(shí)回收多余線程，兼顧了高并發(fā)處理能力和資源利用率。
一、緩存線程池的核心設(shè)計(jì)理念
緩存線程池的核心目標(biāo)是解決任務(wù)量波動(dòng)場景下的線程資源高效管理，其設(shè)計(jì)理念圍繞四大核心目標(biāo)展開：
- 動(dòng)態(tài)擴(kuò)縮容：核心線程數(shù)保持穩(wěn)定，任務(wù)積壓時(shí)臨時(shí)創(chuàng)建新線程，空閑線程超時(shí)時(shí)自動(dòng)回收；
- 任務(wù)異步調(diào)度：通過線程安全的任務(wù)隊(duì)列解耦任務(wù)提交與執(zhí)行；
- 資源可控：限制最大線程數(shù)，避免無限制創(chuàng)建線程導(dǎo)致系統(tǒng)資源耗盡；
- 優(yōu)雅退出：支持線程池停止時(shí)等待所有任務(wù)執(zhí)行完畢，避免任務(wù)丟失。
本文實(shí)現(xiàn)的緩存線程池核心結(jié)構(gòu)如下：
- 線程安全的任務(wù)隊(duì)列（
SyncQueue）：存儲(chǔ)待執(zhí)行的任務(wù)，支持阻塞 / 超時(shí)獲取任務(wù)； - 線程管理模塊：維護(hù)線程集合，統(tǒng)計(jì)核心線程、空閑線程、當(dāng)前線程數(shù)；
- 動(dòng)態(tài)擴(kuò)縮容邏輯：根據(jù)空閑線程數(shù)和任務(wù)隊(duì)列狀態(tài)創(chuàng)建 / 回收線程；
- 任務(wù)提交接口：支持普通任務(wù)和帶返回值的異步任務(wù)提交。
二、核心組件拆解
2.1 線程安全的任務(wù)隊(duì)列（SyncQueue）
任務(wù)隊(duì)列是線程池的 “消息中樞”，需要保證多線程下的安全讀寫，同時(shí)支持 “隊(duì)列滿時(shí)阻塞提交”“隊(duì)列空時(shí)阻塞獲取”“超時(shí)退出” 等核心能力。
核心實(shí)現(xiàn)細(xì)節(jié)
#include <list>
#include <vector>
#include <deque>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <iostream>
using namespace std;
#include "Logger.hpp"
#define QUESTOP 2
#define QUEEMPTY 1
#define ok 0
#ifndef SYNC_QUEUE_1_HPP
#define SYNC_QUEUE_1_HPP
namespace tulun
{
static const size_t MaxTaskCount = 500;
template <class T> // Task = 隊(duì)列里存的類型
class SyncQueue
{
private:
std::deque<T> m_queue; // 底層容器：雙端隊(duì)列（存任務(wù)），頭尾操作O(1)
mutable std::mutex m_mutex; // 互斥鎖，保護(hù)隊(duì)列所有共享操作
std::condition_variable m_notEmpty; // 條件變量：隊(duì)列非空，喚醒消費(fèi)者線程
std::condition_variable m_notFull; // 條件變量：隊(duì)列未滿，喚醒生產(chǎn)者線程
std::condition_variable m_waitStop; // 條件變量：等待隊(duì)列為空再停止
int m_maxSize; // 隊(duì)列最大容量上限
bool m_needStop; // true停止標(biāo)記（初始為false）
int m_waitTime = 100; // 超時(shí)等待時(shí)間，單位ms
// 判斷隊(duì)列是否滿
bool IsFull() const
{
bool full = m_queue.size() >= m_maxSize;
// LOG_INFO << "full: " << full;
// LOG_FATAL<<"full: "<<full;
return full;
}
// 判斷隊(duì)列是否為空
bool IsEmpty() const
{
bool empty = m_queue.empty();
// LOG_INFO << "empty: " << empty;
return empty;
}
template <class F> // F 傳進(jìn)來的參數(shù)類型
int Add(F &&task)
{
std::unique_lock<std::mutex> locker(m_mutex); // 加鎖
// if(IsFull()) return ;
// m_notFull.wait(locker,[this]()->bool{
// return m_needStop || !IsFull();
// });
// 隊(duì)列滿 && 不停止 → 等待
// while (!m_needStop && IsFull())
//{
// auto tag = m_notFull.wait_for(locker, std::chrono::milliseconds(m_waitTime)); // 阻塞，等待不滿
// if (tag == std::cv_status::timeout && IsFull())
// {
// return 1;
// }
//}
// 帶超時(shí)等待：隊(duì)列未滿 或 線程池停止 就解除阻塞
auto tag = m_notFull.wait_for(locker,
std::chrono::milliseconds(m_waitTime),
[this]() -> bool
{
return m_needStop || !IsFull();
}); // lambda 返回 true → 停止等待，繼續(xù)往下執(zhí)行代碼
if (m_needStop)
{
return 2; // 線程池已停止，入隊(duì)失敗
}
if (!tag) // 超時(shí)隊(duì)列依舊滿，觸發(fā)拒絕策略
{
return 1; // IsFull() true;
}
// 完美轉(zhuǎn)發(fā)，零拷貝將任務(wù)放入隊(duì)列尾部
m_queue.push_back(std::forward<F>(task)); // 放入隊(duì)列
m_notEmpty.notify_all(); // 喚醒消費(fèi)者：有活干了！
return 0;
}
public:
SyncQueue(int maxsize = MaxTaskCount)
: m_maxSize(maxsize),
m_needStop(false)
{
}
~SyncQueue()
{
if (!m_needStop)
{
Stop();
}
}
SyncQueue(const SyncQueue &) = delete;
SyncQueue &operator=(const SyncQueue &) = delete;
// 生產(chǎn)者 Put（放任務(wù)）
int Put(const T &task)
{
return Add(task);
}
int Put(T &&task)
{
return Add(std::forward<T>(task));
}
// 消費(fèi)者 Take（取任務(wù)）
int Take(T &task)
{
std::unique_lock<std::mutex> locker(m_mutex); // 加鎖
// 隊(duì)列空 && 不停止 → 等待
// while (!m_needStop && IsEmpty())
// {
// m_notEmpty.wait(locker); // 阻塞，等待不為空
// }
bool tag = m_notEmpty.wait_for(locker,
std::chrono::milliseconds(m_waitTime),
[this]() -> bool
{
return m_needStop || !IsEmpty();
});
if (m_needStop)
{
return 2;
}
if (!tag)
{
return 1;
}
task = m_queue.front();
m_queue.pop_front(); // 取出任務(wù)
m_notFull.notify_all(); // 喚醒生產(chǎn)者：有空位啦！
return 0;
}
// 批量取任務(wù)（高性能)
int Task(std::deque<T> &tqu)
{
std::unique_lock<std::mutex> locker(m_mutex);
bool tag = m_notEmpty.wait_for(locker,
std::chrono::milliseconds(m_waitTime),
[this]() -> bool
{
return m_needStop || !IsEmpty();
});
if (m_needStop)
{
return 2;
}
if (!tag)
{
return 1;
}
tqu = std::move(m_queue); // 【一次性拿走所有任務(wù)】
m_notFull.notify_all();
return 0;
}
// 立即停止：直接置位標(biāo)記，喚醒所有線程
void Stop()
{
{
std::unique_lock<std::mutex> locker(m_mutex);
m_needStop = true;
}
// 鎖外喚醒！性能優(yōu)化點(diǎn)
m_notEmpty.notify_all();
m_notFull.notify_all();
}
// 等待隊(duì)空停止：處理完所有任務(wù)再停止（業(yè)務(wù)最常用）
void WaitQueueEmptyStop()
{
std::unique_lock<std::mutex> locker(m_mutex);
// 等待隊(duì)列所有任務(wù)消費(fèi)完畢
while (!IsEmpty())
{
m_waitStop.wait_for(locker, std::chrono::milliseconds(m_waitTime));
}
m_needStop = true;
m_notFull.notify_all();
m_notEmpty.notify_all();
}
bool Empty() const
{
std::unique_lock<std::mutex> locker(m_mutex);
return m_queue.empty();
}
bool Full() const
{
std::unique_lock<std::mutex> locker(m_mutex);
return m_queue.size() >= m_maxSize;
}
size_t Size() const
{
std::unique_lock<std::mutex> locker(m_mutex);
return m_queue.size();
}
size_t Count() const
{
return m_queue.size();
}
};
} // namespace tulun
#endif關(guān)鍵設(shè)計(jì)點(diǎn)
- 雙條件變量：
m_notEmpty供消費(fèi)者（線程池工作線程）等待，m_notFull供生產(chǎn)者（任務(wù)提交線程）等待，減少無效喚醒； - 超時(shí)等待：
wait_for替代無限等待，避免線程永久阻塞； - 移動(dòng)語義：
Put(T &&task)支持右值引用，減少任務(wù)拷貝開銷； - 異常安全：通過
unique_lock管理鎖，保證異常時(shí)鎖自動(dòng)釋放。
2.2 緩存線程池核心（CachedThreadPool）
線程池的核心邏輯集中在線程管理和任務(wù)調(diào)度，主要包含以下模塊：
2.2.1 核心成員變量
class CachedThreadPool {
public:
using Task = std::function<void(void)>;
private:
// static const int KeepAliveTime = 60; // s;
static int KeepAliveTime;
std::unordered_map<std::thread::id, std::shared_ptr<std::thread>> m_threadgroup;// 線程管理表
int m_coreThreadSize = 4; // 核心線程（永遠(yuǎn)不死）
int m_maxThreadSize = std::thread::hardware_concurrency();// 最大線程數(shù)
mutable std::mutex m_mutex;
std::condition_variable m_threadExit;
std::atomic<int> m_idleThreadSize = 0;// 空閑線程數(shù)量
std::atomic<int> m_curThreadSize = 0;// 當(dāng)前總線程數(shù)
tulun::SyncQueue<Task> m_queue; // 任務(wù)隊(duì)列
std::atomic<bool> m_running = false; // true;// 運(yùn)行標(biāo)記
std::once_flag m_flag;
};2.2.2CachedThreadPool.hpp
#include "SyncQueue_1.hpp"
#include <functional>
#include <map>
#include <unordered_map>
#include <thread>
#include <memory>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <future>
using namespace std;
#ifndef CACHED_THREAD_POOL_HPP
#define CACHED_THREAD_POOL_HPP
namespace tulun
{
class CachedThreadPool
{
public:
using Task = std::function<void(void)>;
private:
// static const int KeepAliveTime = 60; // s;
static int KeepAliveTime;
std::unordered_map<std::thread::id, std::shared_ptr<std::thread>> m_threadgroup;// 線程管理表
int m_coreThreadSize = 4; // 核心線程（永遠(yuǎn)不死）
int m_maxThreadSize = std::thread::hardware_concurrency();// 最大線程數(shù)
mutable std::mutex m_mutex;
std::condition_variable m_threadExit;
std::atomic<int> m_idleThreadSize = 0;// 空閑線程數(shù)量
std::atomic<int> m_curThreadSize = 0;// 當(dāng)前總線程數(shù)
tulun::SyncQueue<Task> m_queue; // 任務(wù)隊(duì)列
std::atomic<bool> m_running = false; // true;// 運(yùn)行標(biāo)記
std::once_flag m_flag;
void Start(int numthreads);
void RunInThread();
void StopThreadGroup();
void newThread();
public:
CachedThreadPool(int initnumthreads = 2, int taskqueuesize = MaxTaskCount);
~CachedThreadPool();
void Stop();
void AddTask(Task &&task);
void AddTask(const Task &task);
void PrintInfo() const
{
std::lock_guard<std::mutex> locker(m_mutex);
cout << "m_core: " << m_coreThreadSize << endl;
cout << "m_idle: " << m_idleThreadSize << endl;
cout << "m_cur: " << m_curThreadSize << endl;
cout << "m_max: " << m_maxThreadSize << endl;
cout<<"-------------------------------"<<endl;
}
template <class Func, class... Args>
auto submit(Func &&func, Args &&...args)
{
// typedef decltype(func(args...)) RetType;
// using RetType = decltype(func(args...));
// std::packaged_task<RetType()> task(
// std::bind(
// std::forward<Func>(func),
// std::forward<Args>(args)...)
//);
// 正確推導(dǎo)返回值（支持成員函數(shù)）
using RetType = decltype(std::invoke(std::forward<Func>(func), std::forward<Args>(args)...));
// 封裝任務(wù)
auto task = std::make_shared<std::packaged_task<RetType()>>(
[func = std::forward<Func>(func), ... args = std::forward<Args>(args)]() mutable
{
return std::invoke(func, args...);
});
std::future<RetType> result = task->get_future();
if (m_queue.Put([task]() -> void
{ (*task)(); }) != 0)
{
LOG_ERROR << "Add task run task";
(*task)();
}
newThread();
return result;
}
int getThreadNUm() const;
};
}
#endif2.2.3線程池初始化（Start）
初始化時(shí)創(chuàng)建核心線程，作為常駐線程處理任務(wù)：
void CachedThreadPool::Start(int numthreads)
{
m_running = true;
m_curThreadSize = numthreads;
// 創(chuàng)建核心線程
for (int i = 0; i < numthreads; ++i)
{
auto tha = std::make_shared<std::thread>(&CachedThreadPool::RunInThread, this);
std::thread::id tid = tha->get_id();
m_threadgroup.emplace(tid, tha);
// 初始時(shí)所有線程都處于空閑狀態(tài)（m_idleThreadSize 增加）。
++m_idleThreadSize;
}
}2.2.4 工作線程邏輯（RunInThread）
工作線程的核心邏輯是 “循環(huán)獲取任務(wù)執(zhí)行，空閑超時(shí)則回收”：
int CachedThreadPool::KeepAliveTime = 60;
void CachedThreadPool::RunInThread()
{
auto tid = std::this_thread::get_id();
auto startTime = std::chrono::high_resolution_clock().now();
while (m_running)
{
Task task;
if (m_queue.Empty())
{
auto now = std::chrono::high_resolution_clock().now();
auto intervalTime =
std::chrono::duration_cast<std::chrono::seconds>(now - startTime).count();
std::lock_guard<std::mutex> locker(m_mutex);
if (intervalTime >= KeepAliveTime && m_curThreadSize > m_coreThreadSize)
{
// m_threadgroup.find(tid)->second->join();// error;
// 超時(shí)且當(dāng)前線程數(shù)超過核心數(shù) -> 該線程退出
m_threadgroup.find(tid)->second->detach();
// detach 后線程對象仍然在 m_threadgroup 中
m_threadgroup.erase(tid);
m_curThreadSize--;
m_idleThreadSize--;
// m_threadExit.notify_all();//????需要嗎
return;
}
}
// 當(dāng)隊(duì)列空且未停止時(shí)，Take會(huì)阻塞或超時(shí)返回非
// 返回值：0成功，1超時(shí)無任務(wù)，2停止
if (m_queue.Take(task) != 0)
{
continue;
}
if (m_running && task)
{
m_idleThreadSize--; // 即將執(zhí)行任務(wù)，空閑數(shù)減1
task(); // 執(zhí)行任務(wù)
m_idleThreadSize++; // 執(zhí)行完畢，空閑數(shù)加1
// 重置空閑計(jì)時(shí)起點(diǎn)（剛執(zhí)行完任務(wù)，認(rèn)為現(xiàn)在開始進(jìn)入空閑期）
startTime = std::chrono::high_resolution_clock().now();
}
}
}2.2.5 動(dòng)態(tài)創(chuàng)建線程（newThread）
當(dāng)空閑線程不足且未達(dá)最大線程數(shù)時(shí)，創(chuàng)建新線程：
// 擴(kuò)容
void CachedThreadPool::newThread()
{
std::lock_guard<std::mutex> locker(m_mutex);
if (m_idleThreadSize <= 0 && m_curThreadSize < m_maxThreadSize)
{
auto tha = std::make_shared<std::thread>(&CachedThreadPool::RunInThread, this);
std::thread::id tid = tha->get_id();
m_threadgroup.emplace(tid, tha);
m_idleThreadSize++;
m_curThreadSize++;
}
}2.2.6 任務(wù)提交接口
支持兩種任務(wù)提交方式：普通任務(wù)（無返回值）和異步任務(wù)（帶返回值）。
普通任務(wù)提交：
void CachedThreadPool::AddTask(Task &&task)
{
if (m_queue.Put(std::forward<Task>(task)) != 0)
{
LOG_INFO << "task()";
task();
}
newThread(); // 嘗試創(chuàng)建新線程
}
void CachedThreadPool::AddTask(const Task &task)
{
if (m_queue.Put(task) != 0)
{
LOG_INFO << "task()";
task();
}
newThread();
}異步任務(wù)提交（帶返回值）：利用 packaged_task 和 future 實(shí)現(xiàn)異步結(jié)果獲取：
template <class Func, class... Args>
auto submit(Func &&func, Args &&...args)
{
// 正確推導(dǎo)返回值（支持成員函數(shù)）
using RetType = decltype(std::invoke(std::forward<Func>(func), std::forward<Args>(args)...));
// 封裝任務(wù)
auto task = std::make_shared<std::packaged_task<RetType()>>(
[func = std::forward<Func>(func), ... args = std::forward<Args>(args)]() mutable
{
return std::invoke(func, args...);
});
std::future<RetType> result = task->get_future();
if (m_queue.Put([task]() -> void
{ (*task)(); }) != 0)
{
LOG_ERROR << "Add task run task";
(*task)();
}
newThread();
return result;
}2.2.7 優(yōu)雅停止（StopThreadGroup）
停止線程池時(shí)，需等待所有任務(wù)執(zhí)行完畢，再回收所有線程：
void CachedThreadPool::StopThreadGroup()
{
// 等待隊(duì)列消費(fèi)完畢，然后設(shè)置停止標(biāo)志
m_queue.WaitQueueEmptyStop();
m_coreThreadSize = 0;
KeepAliveTime = 0;// 強(qiáng)制回收所有空閑線程
std::unique_lock<std::mutex> locker(m_mutex);
while (!m_threadgroup.empty())
{
m_threadExit.wait_for(locker, std::chrono::milliseconds(100));
}
m_running = false;
}
void CachedThreadPool::Stop()
{
std::call_once(m_flag, std::bind(&CachedThreadPool::StopThreadGroup, this));
}2.2.8 構(gòu)造+析構(gòu)
CachedThreadPool::CachedThreadPool(int initnumthreads, int taskqueuesize)
: m_coreThreadSize(initnumthreads),
// m_idleThreadSize(0),
// m_maxThreadSize(0),
// m_curThreadSize(0),
m_queue(taskqueuesize)
// m_running(false)
{
Start(initnumthreads);
}
CachedThreadPool::~CachedThreadPool()
{
if (m_running)
{
Stop();
}
}三、緩存線程池 vs 固定線程池：核心特性對比
| 特性 | FixedThreadPool | CachedThreadPool |
|---|---|---|
| 線程復(fù)用 | 支持復(fù)用，但無法動(dòng)態(tài)創(chuàng)建新線程 | 優(yōu)先復(fù)用空閑線程，無空閑線程時(shí)動(dòng)態(tài)創(chuàng)建新線程 |
| 池大小 | 固定數(shù)量（創(chuàng)建時(shí)指定） | 可增長，最大上限為 Integer.MAX_VALUE（或自定義上限） |
| 隊(duì)列大小 | 無限制 | 無限制 |
| 超時(shí)機(jī)制 | 無空閑超時(shí)，線程常駐 | 默認(rèn)空閑線程存活 60 秒，超時(shí)自動(dòng)回收 |
| 適用場景 | 穩(wěn)定高負(fù)載、CPU 密集型任務(wù)（避免頻繁線程切換） | 大量短期異步任務(wù)、低負(fù)載波動(dòng)場景 |
| 線程生命周期 | 線程池銷毀前不自動(dòng)銷毀 | 空閑超時(shí)自動(dòng)終止，避免資源浪費(fèi) |
從對比中可以看出：
FixedThreadPool適合穩(wěn)定的 CPU 密集型任務(wù)，避免線程頻繁創(chuàng)建 / 銷毀帶來的開銷；CachedThreadPool適合短期、并發(fā)波動(dòng)大的 IO 密集型任務(wù)，通過動(dòng)態(tài)擴(kuò)縮容平衡性能與資源消耗。
四、使用場景與最佳實(shí)踐
4.1 緩存線程池的適用場景
結(jié)合特性，緩存線程池最適合以下場景：
- 大量短期任務(wù)：適合處理大量短期任務(wù)，當(dāng)任務(wù)到來時(shí)盡可能創(chuàng)建新線程執(zhí)行，空閑線程可復(fù)用，避免頻繁創(chuàng)建 / 銷毀線程的額外開銷；
- 任務(wù)響應(yīng)快速：適合需要快速響應(yīng)的任務(wù)，可根據(jù)任務(wù)量快速創(chuàng)建 / 啟動(dòng)新線程，減少任務(wù)等待時(shí)間；
- 無需限制線程數(shù)量：適合任務(wù)到來時(shí)不限制線程數(shù)量的場景，只要內(nèi)存充足，可動(dòng)態(tài)創(chuàng)建新線程；
- 短期任務(wù)高并發(fā)：適合處理高并發(fā)的短期任務(wù)，任務(wù)完成后線程池保留空閑線程，為下一批任務(wù)做準(zhǔn)備。
4.2 最佳實(shí)踐與避坑指南
緩存線程池并非 “萬能方案”，高負(fù)載場景下存在明顯風(fēng)險(xiǎn)，因此工程實(shí)踐中需注意以下幾點(diǎn)：
- 警惕無界隊(duì)列風(fēng)險(xiǎn)：無界任務(wù)隊(duì)列可能導(dǎo)致內(nèi)存溢出、任務(wù)延遲過高，建議設(shè)置有界隊(duì)列；
- 控制最大線程數(shù)：默認(rèn)無上限的線程創(chuàng)建可能導(dǎo)致系統(tǒng)資源耗盡，需自定義最大線程數(shù)；
- 避免長時(shí)間阻塞任務(wù)：緩存線程池不適合執(zhí)行長時(shí)間任務(wù)，否則會(huì)導(dǎo)致后續(xù)任務(wù)堆積、線程數(shù)持續(xù)膨脹；
- 推薦使用
ThreadPoolExecutor：Java/C++ 中可直接使用ThreadPoolExecutor（或其 C++ 實(shí)現(xiàn)），通過自定義核心線程數(shù)、最大線程數(shù)、隊(duì)列大小、拒絕策略實(shí)現(xiàn)細(xì)粒度控制； - 擴(kuò)展鉤子函數(shù)：可重載
beforeExecute/afterExecute鉤子函數(shù)，實(shí)現(xiàn)任務(wù)執(zhí)行前后的自定義操作（如日志、監(jiān)控）； - 線程工廠定制：通過自定義
ThreadFactory為線程命名、設(shè)置守護(hù)線程屬性，便于問題排查； - 動(dòng)態(tài)調(diào)整池大?。嚎蓪?shí)現(xiàn)動(dòng)態(tài)線程池，運(yùn)行時(shí)根據(jù)任務(wù)量、CPU 使用率調(diào)整核心線程數(shù)與最大線程數(shù)。
五、緩存線程池的使用示例
以下示例展示了如何使用緩存線程池處理大規(guī)模任務(wù)（隨機(jī)數(shù)生成）：
#include <random>
#include <iostream>
using namespace std;
#include "Timestamp.hpp"
#include "Logger.hpp"
#include "CachedThreadPool.hpp"
static const size_t row = 10000;
static const size_t col = 100000;
std::vector<std::vector<int>> iveca, ivecb, ivecc, ivecd;
void RandInit(std::vector<int> &ivec)
{
std::random_device rd; // 真隨機(jī)種子源
std::mt19937 gen(rd()); // 隨機(jī)數(shù)引擎
std::uniform_int_distribution<int> dis(0, 100000); // 分布范圍 [1, 100]
// int random_num = dis(gen);
// ivec.reserve(col);
for (int i = 0; i < col; ++i)
{
// ivec.push_back(rand() % 10000);
ivec.push_back(dis(gen));
}
}
void SortVec(std::vector<int> &ivec)
{
std::sort(ivec.begin(), ivec.end());
}
void SaveFile(std::vector<int> &ivec, const std::string &filename)
{
FILE *pf = fopen(filename.c_str(), "a");
if (nullptr == pf)
{
LOG_FATAL << "fopen fail \n";
exit(EXIT_FAILURE);
}
for (int i = 0; i < col; ++i)
{
fprintf(pf, "%d ", ivec[i]);
}
fprintf(pf, "\n---------------------\n");
fclose(pf);
pf = nullptr;
}
void test_funb()
{
srand(time(nullptr));
cout << "4 線程測試...." << endl;
tulun::Timestamp start, end;
start = tulun::Timestamp::Now();
{
ivecb.resize(row);
tulun::CachedThreadPool mypool(4, 500);
for (int i = 0; i < row; ++i)
{
ivecb[i].reserve(col);
// RandInit(ivecb[i]);
mypool.AddTask(std::bind(RandInit, std::ref(ivecb[i])));
if ((i + 1) % 5000 == 0)
{
mypool.PrintInfo();
}
}
}
end = tulun::Timestamp::Now();
cout << "RandInit: " << (tulun::diffMicro(end, start)) / 1000000 << "." << (tulun::diffMicro(end, start)) % 1000000 << endl;
start = tulun::Timestamp::Now();
{
tulun::CachedThreadPool mypool(4, 500);
for (int i = 0; i < row; ++i)
{
// SortVec(ivecb[i]);
mypool.AddTask(std::bind(SortVec, std::ref(ivecb[i])));
if ((i + 1) % 5000 == 0)
{
mypool.PrintInfo();
}
}
}
end = tulun::Timestamp::Now();
cout << "SortVec : " << (tulun::diffMicro(end, start)) / 1000000 << "." << (tulun::diffMicro(end, start)) % 1000000 << endl;
start = tulun::Timestamp::Now();
{
tulun::CachedThreadPool mypool(4, 500);
for (int i = 0; i < row; ++i)
{
// SaveFile(ivecb[i], "iveca.txt");
mypool.AddTask(std::bind(SaveFile, std::ref(ivecb[i]), std::string("ivecb.txt")));
if ((i + 1) % 5000 == 0)
{
mypool.PrintInfo();
}
}
}
end = tulun::Timestamp::Now();
cout << "SaveFile: " << (tulun::diffMicro(end, start)) / 1000000 << "." << (tulun::diffMicro(end, start)) % 1000000 << endl;
return;
}
void funa()
{
for (int i = 0; i < 1000; ++i)
;
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
void funb()
{
for (int i = 0; i < 100; ++i)
;
std::this_thread::sleep_for(std::chrono::milliseconds(rand() % 100));
}
void test_func()
{
{
tulun::CachedThreadPool mypool(4, 500);
for (int i = 0; i < 1000; ++i)
{
mypool.AddTask(funa);
if ((i + 1) % 100 == 0)
{
mypool.PrintInfo();
}
}
cout << "###############################" << endl;
for (int i = 0; i < 1000; ++i)
{
mypool.AddTask(funb);
if ((i + 1) % 10 == 0)
{
mypool.PrintInfo();
}
}
std::this_thread::sleep_for(std::chrono::seconds(1));
mypool.PrintInfo();
std::this_thread::sleep_for(std::chrono::seconds(1));
mypool.PrintInfo();
std::this_thread::sleep_for(std::chrono::seconds(1));
mypool.PrintInfo();
std::this_thread::sleep_for(std::chrono::seconds(1));
mypool.PrintInfo();
std::this_thread::sleep_for(std::chrono::seconds(1));
mypool.PrintInfo();
std::this_thread::sleep_for(std::chrono::seconds(1));
mypool.PrintInfo();
std::this_thread::sleep_for(std::chrono::seconds(1));
mypool.PrintInfo();
std::this_thread::sleep_for(std::chrono::seconds(1));
mypool.PrintInfo();
std::this_thread::sleep_for(std::chrono::seconds(1));
mypool.PrintInfo();
}
}
int main()
{
tulun::Logger::setLogLevel(tulun::LOG_LEVEL::INFO);
//test_funb();
test_func();
return 0;
}六、核心設(shè)計(jì)亮點(diǎn)與優(yōu)化方向
6.1 設(shè)計(jì)亮點(diǎn)
- 原子變量保證線程安全：
m_idleThreadSize、m_curThreadSize等使用std::atomic，避免鎖競爭； - 超時(shí)機(jī)制避免永久阻塞：任務(wù)隊(duì)列的
wait_for和線程回收的超時(shí)判斷，提升系統(tǒng)魯棒性； - 移動(dòng)語義減少拷貝：任務(wù)提交支持右值引用，降低大任務(wù)的拷貝開銷；
- 降級(jí)策略：任務(wù)隊(duì)列滿時(shí)，直接在提交線程執(zhí)行任務(wù)，避免任務(wù)丟失；
- 異步任務(wù)支持：通過
packaged_task和future實(shí)現(xiàn)帶返回值的任務(wù)提交。
6.2 可優(yōu)化方向
- 線程回收機(jī)制：當(dāng)前使用
detach回收線程，可改為join并結(jié)合線程池銷毀邏輯，更安全； - 任務(wù)優(yōu)先級(jí)：擴(kuò)展任務(wù)隊(duì)列支持優(yōu)先級(jí)，優(yōu)先執(zhí)行高優(yōu)先級(jí)任務(wù)；
- 線程局部存儲(chǔ)：為工作線程添加 TLS（線程局部存儲(chǔ)），減少線程間資源競爭；
- 監(jiān)控與統(tǒng)計(jì)：增加任務(wù)執(zhí)行耗時(shí)、線程創(chuàng)建 / 回收次數(shù)等統(tǒng)計(jì)指標(biāo)，便于性能調(diào)優(yōu)；
- 異常捕獲：工作線程執(zhí)行任務(wù)時(shí)捕獲異常，避免單個(gè)任務(wù)崩潰導(dǎo)致線程退出。
七、總結(jié)
緩存線程池是高并發(fā)場景下的高效線程管理方案，其核心價(jià)值在于用動(dòng)態(tài)擴(kuò)縮容平衡性能與資源消耗：
- 用線程安全的任務(wù)隊(duì)列解耦任務(wù)提交與執(zhí)行；
- 核心線程常駐，臨時(shí)線程按需創(chuàng)建、超時(shí)回收；
- 原子變量和條件變量保證多線程下的狀態(tài)同步；
- 降級(jí)策略和超時(shí)機(jī)制提升系統(tǒng)魯棒性。
但緩存線程池并非 “銀彈”，在高負(fù)載、長時(shí)間任務(wù)場景下需謹(jǐn)慎使用，建議結(jié)合ThreadPoolExecutor 的設(shè)計(jì)思想，通過自定義參數(shù)實(shí)現(xiàn)更可控的線程池。
到此這篇關(guān)于C++ 緩存線程池CachedThreadPool原理、實(shí)現(xiàn)與對比解析的文章就介紹到這了,更多相關(guān)C++ 緩存線程池CachedThreadPool內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
C++實(shí)現(xiàn)LeetCode(203.移除鏈表元素)
這篇文章主要介紹了C++實(shí)現(xiàn)LeetCode(203.移除鏈表元素),本篇文章通過簡要的案例,講解了該項(xiàng)技術(shù)的了解與使用,以下就是詳細(xì)內(nèi)容,需要的朋友可以參考下2021-08-08
這篇文章主要介紹了c++11 新特性——智能指針使用的相關(guān)資料，幫助大家更好的理解和使用c++，感興趣的朋友可以了解下2021-02-02
C語言字符串函數(shù)與內(nèi)存函數(shù)精講
這篇文章主要介紹一些c語言中常用字符串函數(shù)和內(nèi)存函數(shù)的使用，并且為了幫助讀者理解和使用，也都模擬實(shí)現(xiàn)了他們的代碼,需要的朋友可以參考一下2022-04-04
深入探討C語言中局部變量與全局變量在內(nèi)存中的存放位置
本篇文章是對在C語言中局部變量與全局變量在內(nèi)存中的存放位置進(jìn)行了詳細(xì)的分析介紹，需要的朋友參考下2013-05-05
這篇文章主要為大家詳細(xì)介紹了C++實(shí)現(xiàn)bmp格式圖像讀寫，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2021-10-10
C++關(guān)鍵字之likely和unlikely詳解
這篇文章主要介紹了C++關(guān)鍵字之likely和unlikely,C++20之前的，likely和unlikely只不過是一對自定義的宏，而C++20中正式將likely和unlikely確定為屬性關(guān)鍵字，本文給大家詳細(xì)講解，需要的朋友可以參考下2022-10-10
C++實(shí)現(xiàn)strcmp字符串比較的深入探討
本篇文章是對使用C++實(shí)現(xiàn)strcmp字符串比較進(jìn)行了詳細(xì)的分析介紹，需要的朋友參考下2013-05-05
QtObject 類型是一個(gè)非可視元素,僅包含objectName屬性,如果需要一個(gè)非常輕量級(jí)的類型來包含一組自定義屬性,那么創(chuàng)建一個(gè)QtObject,本文給大家介紹qml中QtObject類型的使用,感興趣的朋友一起看看吧2024-01-01
詳解如何用alpine鏡像做一個(gè)最小的鏡像并運(yùn)行c++程序
這篇文章主要介紹了詳解如何用alpine鏡像做一個(gè)最小的鏡像并運(yùn)行c++程序，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-10-10
這篇文章主要給大家介紹了關(guān)于C++中string類（C++字符串）入門的相關(guān)資料，文中通過示例代碼介紹的非常詳細(xì)，對大家學(xué)習(xí)或者使用C++具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面來一起學(xué)習(xí)學(xué)習(xí)吧2019-11-11