C++線程池實(shí)現(xiàn)代碼

第C++線程池實(shí)現(xiàn)代碼這段時間看了《C++并發(fā)編程實(shí)戰(zhàn)》的基礎(chǔ)內(nèi)容，想著利用最近學(xué)的知識自己實(shí)現(xiàn)一個簡單的線程池。

什么是線程池

線程池（threadpool）是一種線程使用模式。線程過多或者頻繁創(chuàng)建和銷毀線程會帶來調(diào)度開銷，進(jìn)而影響緩存局部性和整體性能。而線程池維護(hù)著多個線程，等待著管理器分配可并發(fā)執(zhí)行的任務(wù)。這避免了在處理短時間任務(wù)時創(chuàng)建與銷毀線程的代價，以及保證了線程的可復(fù)用性。線程池不僅能夠保證內(nèi)核的充分利用，還能防止過分調(diào)度。

個人對線程池的理解是：利用已經(jīng)創(chuàng)建的固定數(shù)量的線程去執(zhí)行指定的任務(wù)，從而避免線程重復(fù)創(chuàng)建和銷毀帶來的額外開銷。

C++11中，線程我們可以理解為對應(yīng)一個thread對象，任務(wù)可以理解為要執(zhí)行的函數(shù)，通常是耗時的函數(shù)。

我們的任務(wù)多少和順序并非固定的，因此需要有一個方法能添加指定的任務(wù)，任務(wù)存放的地方應(yīng)該是一個任務(wù)隊(duì)列，因?yàn)槲覀兊木€程數(shù)量有限，當(dāng)任務(wù)很多時同時執(zhí)行的任務(wù)數(shù)量也有限，因此任務(wù)需要排隊(duì)，遵循先來后到的原則。

當(dāng)要執(zhí)行一個任務(wù)時，意味著先將這個任務(wù)從隊(duì)列取出，再執(zhí)行相應(yīng)任務(wù)，而“取出”動作的執(zhí)行者是線程池中的線程，這意味我們的隊(duì)列需要考慮多個線程在同一隊(duì)列上執(zhí)行“取出”操作的問題，實(shí)際上，取出任務(wù)操作和添加任務(wù)操作也不能同時進(jìn)行，否則會產(chǎn)生競爭條件；另一方面，程序本身如果就是多線程的，多個線程同時添加任務(wù)的操作也應(yīng)該是互斥的。

當(dāng)沒有任務(wù)可以執(zhí)行時，所有線程應(yīng)該什么也不做，當(dāng)出現(xiàn)了一個任務(wù)時，應(yīng)該將這個任務(wù)分配到任一線程中執(zhí)行。實(shí)現(xiàn)上我們固然可以使用輪詢的方式判斷當(dāng)前隊(duì)列是否有任務(wù)，有則取出（即使加了互斥鎖似乎也無法避免競爭條件？），但這樣會消耗無謂的CPU資源，寫輪詢周期難以選取。其實(shí)，我們可以使用condition_variable代替輪詢。

上述任務(wù)的創(chuàng)建和取出其實(shí)就是經(jīng)典的生產(chǎn)者消費(fèi)者模型。

我們將上面的內(nèi)容都封裝在一個類中，取名ThreadPool，用戶可以在構(gòu)造ThreadPool對象時指定線程池大小，之后可以隨時添加要執(zhí)行的任務(wù)。

classThreadPool

public:

ThreadPool(intn);

~ThreadPool();

voidpushTask(packaged_taskvoid()task);

private:

vectorthread*threadPool;

dequepackaged_taskvoid()taskQueue;

voidtaskConsumer();

mutextaskMutex;

condition_variabletaskQueueCond;

ThreadPool::ThreadPool(intn)

for(inti=0;ii++)

thread*t=newthread(ThreadPool::taskConsumer,this);

threadPool.push_back(t);

t-detach();

ThreadPool::~ThreadPool()

while(!threadPool.empty())

thread*t=threadPool.back();

threadPool.pop_back();

deletet;

voidThreadPool::pushTask(packaged_taskvoid()task)

lock_guardmutexguard(taskMutex);

taskQueue.push_back(std::move(task));

taskQueueCond.notify_one();

voidThreadPool::taskConsumer()

while(true)

unique_lockmutexlk(taskMutex);

taskQueueCond.wait(lk,[]{return!taskQueue.empty();});

packaged_taskvoid()task=std::move(taskQueue.front());

taskQueue.pop_front();

lk.unlock();

task();

這里我使用packaged_task作為任務(wù)，每當(dāng)添加一個任務(wù)，就調(diào)用condition_variable::notify_one方法，調(diào)用condition_variable::wait的線程就會被喚醒，并檢查等待條件。這里有個小細(xì)節(jié)是notify_one在解鎖后執(zhí)行，這樣避免線程喚醒后還要等待互斥鎖解鎖。

使用示例：

voidTask1()

Sleep(1000);

cout"Task1"endl;

voidTask5()

Sleep(5000);

cout"Task5"endl;

classWorker

public:

voidrun();

voidWorker::run()

cout"Worker::runstart"endl;

Sleep(5000);

cout"Worker::runend"endl;

intmain()

ThreadPoolpool(2);

pool.pushTask(packaged_taskvoid()(Task5));

pool.pushTask(packaged_taskvoid()(Task1));

pool.pushTask(packaged_taskvoid()(Task1));

Workerworker;

pool.pushTask(packaged_taskvoid()(bind(Worker::run,worker)));

pool.pushTask(packaged_taskvoid()([](){worker.run();}));

Sleep(20000);

這個線程池目前有幾個缺點(diǎn)：

只能傳入調(diào)用形式為void()形式的函數(shù)或可調(diào)用對象，不能返回任務(wù)執(zhí)行的值，只能通過其他方式同步任務(wù)執(zhí)行結(jié)果（如果有）

傳入?yún)?shù)較為復(fù)雜，必須封裝一層packaged_task，調(diào)用對象方法時需要使用bind或者lambda表達(dá)式的方法封裝

以上缺點(diǎn)在當(dāng)前版本的實(shí)現(xiàn)不予解決，日后另寫博文優(yōu)化。

2025/12/29更新之一：

事實(shí)上，我們只要將packaged_task改為funtion模板類，就可以簡化我們的調(diào)用參數(shù)：

classThreadPool

public:

ThreadPool(intn);

~ThreadPool();

voidpushTask(functionvoid()task);

private:

vectorthread*threadPool;

dequefunctionvoid()taskQueue;

voidtaskConsumer();

mutextaskMutex;

condition_variabletaskQueueCond;

ThreadPool::ThreadPool(intn)

for(inti=0;ii++)

thread*t=newthread(ThreadPool::taskConsumer,this);

threadPool.push_back(t);

t-detach();

ThreadPool::~ThreadPool()

while(!threadPool.empty())

thread*t=threadPool.back();

threadPool.pop_back();

deletet;

voidThreadPool::pushTask(functionvoid()task)

lock_guardmutexguard(taskMutex);

taskQueue.push_back(std::move(task));

taskQueueCond.notify_one();

voidThreadPool::taskConsumer()

while(true)

unique_lockmutexlk(taskMutex);

taskQueueCond.wait(lk,[]{return!taskQueue.empty();});

functionvoid()task=taskQueue.front();

taskQueue.pop_front();

lk.unlock();

task();

調(diào)用代碼改為如下：

ThreadPoolpool(2);

poo