C++11 简单实现线程池的方法
时间：2021-03-05 12:41:41|栏目：C代码|点击： 次
什么是线程池
线程池是一种多线程处理形式，处理过程中将任务添加到队列，然后在创建线程后自动启动这些任务。线程池线程都是后台线程。每个线程都使用默认的堆栈大小，以默认的优先级运行，并处于多线程单元中。如果某个线程在托管代码中空闲（如正在等待某个事件）,则线程池将插入另一个辅助线程来使所有处理器保持繁忙。如果所有线程池线程都始终保持繁忙，但队列中包含挂起的工作，则线程池将在一段时间后创建另一个辅助线程但线程的数目永远不会超过最大值。超过最大值的线程可以排队，但他们要等到其他线程完成后才启动。
不使用线程池有哪些弊端
创建太多线程，将会浪费一定的资源，有些线程未被充分使用。
销毁太多线程，将导致之后浪费时间再次创建它们。
创建线程太慢，将会导致长时间的等待，性能变差。
销毁线程太慢，导致其它线程资源饥饿。
线程池的组成部分
1、线程池管理器（ThreadPoolManager）:用于创建并管理线程池
2、工作线程（WorkThread）: 线程池中线程
3、任务接口（Task）:每个任务必须实现的接口，以供工作线程调度任务的执行。
4、任务队列:用于存放没有处理的任务。提供一种缓冲机制
下面直接看代码实现
ThreadPoolManage.hpp
#pragma once
#include <thread>
#include <vector>
#include <queue>
#include <condition_variable>
#include <mutex>
/*
* 抽象一个任务 根据自己的需求扩展
*/
class AbsTask
{
public:
AbsTask() = default;
virtual ~AbsTask() = default;
public:
virtual void run() = 0;
};
template<class _Ty>
class ThreadPoolManage
{
public:
ThreadPoolManage(unsigned int nMaxThread)
:mMaxThreadNum(nMaxThread)
, mThreadStatus(false)
{
//启动的时候就要创建线程
auto maxNum = std::thread::hardware_concurrency();
//获取当前操作系统中CPU的核心数量 根据核心数量来设置 最大工作线程的并发数量
mMaxThreadNum = mMaxThreadNum > maxNum ? maxNum : mMaxThreadNum;
//创建工作线程池
for (auto i = 0; i < mMaxThreadNum; i++)
{
mWorkers.emplace_back([this] {
while (true)
{
std::unique_lock<std::mutex> lock(this->mQueue_mutex);
this->mCondition.wait(lock, [this]() {return this->mThreadStatus || !this->mTasks.empty(); });
if (this->mThreadStatus && this->mTasks.empty())
{
return;
}
//获取队列头部的任务
auto task = std::move(this->mTasks.front());
//任务出队
this->mTasks.pop();
//执行工作
task.run();
}
});
}
}
~ThreadPoolManage()
{
{
std::unique_lock<std::mutex> lock(this->mQueue_mutex);
this->mThreadStatus = true;
}
//通知所有线程起来工作 然后退出
this->mCondition.notify_all();
//等待所有线程工作完毕
for (std::thread& worker : this->mWorkers)
{
if (worker.joinable()) {
worker.join();
}
}
}
/*
* 添加任务到任务队列
*/
void addTask(_Ty& task)
{
std::unique_lock<std::mutex> lock(this->mQueue_mutex);
if (mThreadStatus) {
throw std::runtime_error("workers stop");
}
mTasks.emplace(std::move(task));
mCondition.notify_one();
}
private:
/*
* 工作线程池
*/
std::vector<std::thread> mWorkers;
/*
* 任务队列
*/
std::queue<_Ty> mTasks;
/*
工作线程的最大并发数量
*/
unsigned int mMaxThreadNum;
/*
条件变量 控制线程池中线程的工作状态
*/
std::condition_variable mCondition;
/*
* 工作线程锁
*/
std::mutex mQueue_mutex;
/*
* 控制线程的开关 false 继续工作 true 退出线程
*/
bool mThreadStatus;
};
调用代码
main.cpp
#include <iostream>
#include <chrono>
#include "ThreadPool.hpp"
class Task :public AbsTask
{
public:
void run() override
{
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "works ...... " << std::this_thread::get_id() << std::endl;
}
};
int main()
{
ThreadPoolManage<Task> ThreadPool(8);
for (size_t i = 0; i < 256; i++)
{
Task task;
ThreadPool.addTask(task);
}
std::cin.get();
system("pause");
}
