线程池 完全基于 C++ 11
文章目录
目标 用C++ 11 的多线程 实现一个 带返回值且接收可变参数的 线程池
组件：信号量类（C++11线程不包含）、返回值类、线程类、线程池类、任务类
架构设计
main函数执行过程
基本组件
信号量
class Semaphore
{
public:
Semaphore(int limit = 0)
:resLimit_(limit)
{}
~Semaphore() = default;
// 获取一个信号量资源
void wait()
{
std::unique_lock<std::mutex> lock(mtx_);
// 等待信号量有资源，没有资源的话，会阻塞当前线程
cond_.wait(lock, [&]()->bool {return resLimit_ > 0; });
resLimit_--;
}
// 增加一个信号量资源
void post()
{
std::unique_lock<std::mutex> lock(mtx_);
resLimit_++;
// linux下condition_variable的析构函数什么也没做
// 导致这里状态已经失效，无故阻塞
cond_.notify_all(); // 等待状态，释放mutex锁 通知条件变量wait的地方，可以起来干活了
}
private:
int resLimit_;
std::mutex mtx_;
std::condition_variable cond_;
};
返回值类型Result设计
// 实现接收提交到线程池的task任务执行完成后的返回值类型Result
class Result
{
public:
Result(std::shared_ptr<Task> task, bool isValid = true);
~Result() = default;
// 问题一：setVal方法，获取任务执行完的返回值的
void setVal(Any any);
// 问题二：get方法，用户调用这个方法获取task的返回值
Any get();
private:
Any any_; // 存储任务的返回值
Semaphore sem_; // 线程通信信号量
std::shared_ptr<Task> task_; // 指向对应获取返回值的任务对象
std::atomic_bool isValid_; // 返回值是否有效
};
接收任意类型的any类 —— 用于Result类存储任务返回值
利用基类std::unique_ptr<Base> base_; 存储 任意类型的派生类，将T的类型（派生类）延后确定（任务执行完成确定以及获取返回值时确定）
// Any类型：可以接收任意数据的类型
class Any
{
public:
Any() = default;
~Any() = default;
Any(const Any&) = delete;
Any& operator=(const Any&) = delete;
Any(Any&&) = default;
Any& operator=(Any&&) = default;
// 这个构造函数可以让Any类型接收任意其它的数据
template<typename T> // T:int Derive<int>
Any(T data) : base_(std::make_unique<Derive<T>>(data))
{}
// 这个方法能把Any对象里面存储的data数据提取出来
template<typename T>
T cast_()
{
// 我们怎么从base_找到它所指向的Derive对象，从它里面取出data成员变量
// 基类指针 =》 派生类指针 RTTI
Derive<T>* pd = dynamic_cast<Derive<T>*>(base_.get());
if (pd == nullptr)
{
throw "type is unmatch!";
}
return pd->data_;
}
private:
// 基类类型
class Base
{
public:
virtual ~Base() = default;
};
// 派生类类型
template<typename T>
class Derive : public Base
{
public:
Derive(T data) : data_(data)
{}
T data_; // 保存了任意的其它类型
};
private:
// 定义一个基类的指针
std::unique_ptr<Base> base_;
};
任务类
// 任务抽象基类
class Task
{
public:
Task();
~Task() = default;
void exec();
void setResult(Result* res);
// 用户可以自定义任意任务类型，从Task继承，重写run方法，实现自定义任务处理
virtual Any run() = 0;
private:
Result* result_; // Result对象的声明周期 》 Task的
};
线程类型
class Thread
{
public:
// 线程函数对象类型
using ThreadFunc = std::function<void(int)>;
// 线程构造
Thread(ThreadFunc func);
// 线程析构
~Thread();
// 启动线程
void start();
// 获取线程id
int getId()const;
private:
ThreadFunc func_;
static int generateId_;
int threadId_; // 保存线程id
};
线程池类型
class ThreadPool
{
public:
// 线程池构造
ThreadPool();
// 线程池析构
~ThreadPool();
// 设置线程池的工作模式
void setMode(PoolMode mode);
// 设置task任务队列上线阈值
void setTaskQueMaxThreshHold(int threshhold);
// 设置线程池cached模式下线程阈值
void setThreadSizeThreshHold(int threshhold);
// 给线程池提交任务
Result submitTask(std::shared_ptr<Task> sp);
// 开启线程池
void start(int initThreadSize = std::thread::hardware_concurrency());
ThreadPool(const ThreadPool&) = delete;
ThreadPool& operator=(const ThreadPool&) = delete;
private:
// 定义线程函数
void threadFunc(int threadid);
// 检查pool的运行状态
bool checkRunningState() const;
private:
std::unordered_map<int, std::unique_ptr<Thread>> threads_; // 线程列表
int initThreadSize_; // 初始的线程数量
int threadSizeThreshHold_; // 线程数量上限阈值
std::atomic_int curThreadSize_; // 记录当前线程池里面线程的总数量
std::atomic_int idleThreadSize_; // 记录空闲线程的数量
std::queue<std::shared_ptr<Task>> taskQue_; // 任务队列
std::atomic_int taskSize_; // 任务的数量
int taskQueMaxThreshHold_; // 任务队列数量上限阈值
std::mutex taskQueMtx_; // 保证任务队列的线程安全
std::condition_variable notFull_; // 表示任务队列不满
std::condition_variable notEmpty_; // 表示任务队列不空
std::condition_variable exitCond_; // 等到线程资源全部回收
PoolMode poolMode_; // 当前线程池的工作模式
std::atomic_bool isPoolRunning_; // 表示当前线程池的启动状态
};
重点代码分析 submitTask 提交任务
// 给线程池提交任务 用户调用该接口，传入任务对象，生产任务
Result ThreadPool::submitTask(std::shared_ptr<Task> sp)
{
// 获取锁
std::unique_lock<std::mutex> lock(taskQueMtx_);
// 线程的通信 等待任务队列有空余 wait wait_for wait_until
// 用户提交任务，最长不能阻塞超过1s，否则判断提交任务失败，返回
if (!notFull_.wait_for(lock, std::chrono::seconds(1),
[&]()->bool { return taskQue_.size() < (size_t)taskQueMaxThreshHold_; }))
{
// 表示notFull_等待1s种，条件依然没有满足
std::cerr << "task queue is full, submit task fail." << std::endl;
return Result(sp, false);
}
// 如果有空余，把任务放入任务队列中
taskQue_.emplace(sp);
taskSize_++;
// 因为新放了任务，任务队列肯定不空了，在notEmpty_上进行通知，赶快分配线程执行任务
notEmpty_.notify_all();
// cached模式 任务处理比较紧急 场景：小而快的任务 需要根据任务数量和空闲线程的数量，判断是否需要创建新的线程出来
if (poolMode_ == PoolMode::MODE_CACHED
&& taskSize_ > idleThreadSize_
&& curThreadSize_ < threadSizeThreshHold_)
{
std::cout << ">>> create new thread..." << std::endl;
// 创建新的线程对象
auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));
int threadId = ptr->getId();
threads_.emplace(threadId, std::move(ptr));
// 启动线程
threads_[threadId]->start();
// 修改线程个数相关的变量
curThreadSize_++;
idleThreadSize_++;
}
// 返回任务的Result对象
return Result(sp);
}
重点代码分析 threadFunc 获取任务、执行任务
// 定义线程函数 线程池的所有线程从任务队列里面消费任务
void ThreadPool::threadFunc(int threadid) // 线程函数返回，相应的线程也就结束了
{
auto lastTime = std::chrono::high_resolution_clock().now();
// 所有任务必须执行完成，线程池才可以回收所有线程资源
for (;;)
{
std::shared_ptr<Task> task;
{
// 先获取锁
std::unique_lock<std::mutex> lock(taskQueMtx_);
std::cout << "tid:" << std::this_thread::get_id()
<< "尝试获取任务..." << std::endl;
// 每一秒中返回一次 怎么区分：超时返回？还是有任务待执行返回
// 锁 + 双重判断
while (taskQue_.size() == 0)
{
// 线程池要结束，回收线程资源
if (!isPoolRunning_)
{
threads_.erase(threadid); // std::this_thread::getid()
std::cout << "threadid:" << std::this_thread::get_id() << " exit!"
<< std::endl;
exitCond_.notify_all();
return; // 线程函数结束，线程结束
}
// cached模式下，有可能已经创建了很多的线程，但是空闲时间超过60s，应该把多余的线程（当前线程）回收
if (poolMode_ == PoolMode::MODE_CACHED)
{
// 条件变量，超时返回了
if (std::cv_status::timeout ==
notEmpty_.wait_for(lock, std::chrono::seconds(1)))
{
auto now = std::chrono::high_resolution_clock().now();
auto dur = std::chrono::duration_cast<std::chrono::seconds>(now - lastTime);
if (dur.count() >= THREAD_MAX_IDLE_TIME
&& curThreadSize_ > initThreadSize_)
{
// 开始回收当前线程
// 记录线程数量的相关变量的值修改
// 把线程对象从线程列表容器中删除 没有办法 threadFunc《=》thread对象
// threadid => thread对象 => 删除
threads_.erase(threadid); // std::this_thread::getid()
curThreadSize_--;
idleThreadSize_--;
std::cout << "threadid:" << std::this_thread::get_id() << " exit!"
<< std::endl;
return;
}
}
}
else
{
// 等待notEmpty条件
notEmpty_.wait(lock);
}
}
idleThreadSize_--;
std::cout << "tid:" << std::this_thread::get_id()
<< "获取任务成功..." << std::endl;
// 从任务队列种取一个任务出来
task = taskQue_.front();
taskQue_.pop();
taskSize_--;
// 如果依然有剩余任务，继续通知其它得线程执行任务
if (taskQue_.size() > 0)
{
notEmpty_.notify_all();
}
// 取出一个任务，进行通知，通知可以继续提交生产任务
notFull_.notify_all();
} // 就应该把锁释放掉
// 当前线程负责执行这个任务
if (task != nullptr)
{
// task->run(); // 执行任务；把任务的返回值setVal方法给到Result
task->exec();
}
idleThreadSize_++;
lastTime = std::chrono::high_resolution_clock().now(); // 更新线程执行完任务的时间
}
}
重点代码分析 线程池析构 —— 保证线程池的任务执行完毕，才回收资源
ThreadPool::~ThreadPool()
{
isPoolRunning_ = false;
// 等待线程池里面所有的线程返回 有两种状态：阻塞 & 正在执行任务中
std::unique_lock<std::mutex> lock(taskQueMtx_);
notEmpty_.notify_all();
// threadFunc 里负责 exitCond_.notify_all();
exitCond_.wait(lock, [&]()->bool {return threads_.size() == 0; });
}