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

更新時間：2021年12月30日 09:34:17 作者：mrbone11

C++11中，線程我們可以理解為對應一個thread對象，任務可以理解為要執(zhí)行的函數(shù)，通常是耗時的函數(shù)。線程過多或者頻繁創(chuàng)建和銷毀線程會帶來調度開銷，進而影響緩存局部性和整體性能

前言

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

什么是線程池

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

思路

個人對線程池的理解是：利用已經創(chuàng)建的固定數(shù)量的線程去執(zhí)行指定的任務，從而避免線程重復創(chuàng)建和銷毀帶來的額外開銷。
C++11中，線程我們可以理解為對應一個thread對象，任務可以理解為要執(zhí)行的函數(shù)，通常是耗時的函數(shù)。
我們的任務多少和順序并非固定的，因此需要有一個方法能添加指定的任務，任務存放的地方應該是一個任務隊列，因為我們的線程數(shù)量有限，當任務很多時同時執(zhí)行的任務數(shù)量也有限，因此任務需要排隊，遵循先來后到的原則。
當要執(zhí)行一個任務時，意味著先將這個任務從隊列取出，再執(zhí)行相應任務，而“取出”動作的執(zhí)行者是線程池中的線程，這意味我們的隊列需要考慮多個線程在同一隊列上執(zhí)行“取出”操作的問題，實際上，取出任務操作和添加任務操作也不能同時進行，否則會產生競爭條件；另一方面，程序本身如果就是多線程的，多個線程同時添加任務的操作也應該是互斥的。
當沒有任務可以執(zhí)行時，所有線程應該什么也不做，當出現(xiàn)了一個任務時，應該將這個任務分配到任一線程中執(zhí)行。實現(xiàn)上我們固然可以使用輪詢的方式判斷當前隊列是否有任務，有則取出（即使加了互斥鎖似乎也無法避免競爭條件？），但這樣會消耗無謂的CPU資源，寫輪詢周期難以選取。其實，我們可以使用condition_variable代替輪詢。
上述任務的創(chuàng)建和取出其實就是經典的生產者消費者模型。
我們將上面的內容都封裝在一個類中，取名ThreadPool，用戶可以在構造ThreadPool對象時指定線程池大小，之后可以隨時添加要執(zhí)行的任務。

實現(xiàn)

class ThreadPool
{
public:
	ThreadPool(int n);
	~ThreadPool();

	void pushTask(packaged_task<void()> &&task);

private:
	vector<thread*> threadPool;
	deque<packaged_task<void()>> taskQueue;

	void taskConsumer();
	mutex taskMutex;
	condition_variable taskQueueCond;
};

ThreadPool::ThreadPool(int n)
{
	for (int i = 0; i < n; i++)
	{
		thread *t = new thread(&ThreadPool::taskConsumer,this);
		threadPool.push_back(t);
		t->detach();
	}
}

ThreadPool::~ThreadPool()
{
	while (!threadPool.empty())
	{
		thread *t=threadPool.back();
		threadPool.pop_back();
		delete t;
	}
}

void ThreadPool::pushTask(packaged_task<void()> &&task)
{
	{
		lock_guard<mutex> guard(taskMutex);
		taskQueue.push_back(std::move(task));
	}
	taskQueueCond.notify_one();
}

void ThreadPool::taskConsumer()
{
	while (true)
	{
		unique_lock<mutex> lk(taskMutex);
		taskQueueCond.wait(lk, [&] {return !taskQueue.empty(); });
		packaged_task<void()> task=std::move(taskQueue.front());
		taskQueue.pop_front();
		lk.unlock();
		task();
	}
}

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

void Task1()
{
	Sleep(1000);
	cout << "Task1"<<endl;
}

void Task5()
{
	Sleep(5000);
	cout << "Task5" << endl;
}

class Worker
{
public:
	void run();
};

void Worker::run()
{
	cout << "Worker::run start" << endl;
	Sleep(5000);
	cout << "Worker::run end" << endl;
}

int main()
{
	ThreadPool pool(2);
	pool.pushTask(packaged_task<void()>(Task5));
	pool.pushTask(packaged_task<void()>(Task1));
	pool.pushTask(packaged_task<void()>(Task1));
	Worker worker;
	pool.pushTask(packaged_task<void()>(bind(&Worker::run,&worker)));
	pool.pushTask(packaged_task<void()>([&](){worker.run();}));
	Sleep(20000);
}

這個線程池目前有幾個缺點：

只能傳入調用形式為void()形式的函數(shù)或可調用對象，不能返回任務執(zhí)行的值，只能通過其他方式同步任務執(zhí)行結果（如果有）
傳入參數(shù)較為復雜，必須封裝一層packaged_task，調用對象方法時需要使用bind或者lambda表達式的方法封裝

以上缺點在當前版本的實現(xiàn)不予解決，日后另寫博文優(yōu)化。
2021/12/29 更新之一：
事實上，我們只要將packaged_task改為funtion模板類，就可以簡化我們的調用參數(shù)：

class ThreadPool
{
public:
	ThreadPool(int n);
	~ThreadPool();

	void pushTask(function<void()> task);

private:
	vector<thread*> threadPool;
	deque<function<void()>> taskQueue;

	void taskConsumer();
	mutex taskMutex;
	condition_variable taskQueueCond;
};

ThreadPool::ThreadPool(int n)
{
	for (int i = 0; i < n; i++)
	{
		thread *t = new thread(&ThreadPool::taskConsumer,this);
		threadPool.push_back(t);
		t->detach();
	}
}

ThreadPool::~ThreadPool()
{
	while (!threadPool.empty())
	{
		thread *t=threadPool.back();
		threadPool.pop_back();
		delete t;
	}
}

void ThreadPool::pushTask(function<void()> task)
{
	{
		lock_guard<mutex> guard(taskMutex);
		taskQueue.push_back(std::move(task));
	}
	taskQueueCond.notify_one();
}

void ThreadPool::taskConsumer()
{
	while (true)
	{
		unique_lock<mutex> lk(taskMutex);
		taskQueueCond.wait(lk, [&] {return !taskQueue.empty(); });
		function<void()> task=taskQueue.front();
		taskQueue.pop_front();
		lk.unlock();
		task();
	}
}

調用代碼改為如下：

ThreadPool pool(2);

pool.pushTask(&Task5);

pool.pushTask(&Task1);

pool.pushTask(&Task1);

Worker worker;

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

pool.pushTask([&](){worker.run(); });//1

Sleep(15000);

我們可以執(zhí)行指定的函數(shù)，也可以將要執(zhí)行的代碼放入lambda表達式的函數(shù)體中，正如1處所示，這樣就能在其他線程中執(zhí)行指定的代碼了。
2021/12/29 更新之二：
我們發(fā)現(xiàn)，main最后都要調用sleep函數(shù)來避免主線程在線程任務完成之前就退出，因此我們希望添加一個接口，等待線程所有任務完成，改進如下，其他函數(shù)同前：

class ThreadPool
{
public:
	ThreadPool(int n);
	~ThreadPool();

	void pushTask(function<void()> task);
	void waitAllTask();

private:
	vector<thread*> threadPool;
	deque<function<void()>> taskQueue;
	atomic<int> busyCount;
	bool bStop;

	void taskConsumer();
	mutex taskQueueMutex;
	condition_variable taskQueueCond;
	condition_variable taskFinishedCond;
};

void ThreadPool::taskConsumer()
{
	while (!bStop)
	{
		unique_lock<mutex> lk(taskQueueMutex);
		taskQueueCond.wait(lk, [&] {return !taskQueue.empty(); });
		busyCount++;
		function<void()> task=taskQueue.front();
		taskQueue.pop_front();
		lk.unlock();
		task();
		busyCount--;
		taskFinishedCond.notify_one();
	}
}

void ThreadPool::waitAllTask()
{
	unique_lock<mutex> lk(taskQueueMutex);
	taskFinishedCond.wait(lk, [&] {return taskQueue.empty() && busyCount==0; });//所有任務均已完成
}

這樣我們只要調用waitAllTask就可以等待所有任務完成啦。

到此這篇關于C++線程池實現(xiàn)代碼的文章就介紹到這了,更多相關C++線程池內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: