1 组成
- 线程（生产者）
- 发布任务
- 队列
- 任务上下文
- 任务执行函数
- 线程池（消费者）
- 取出任务
- 执行任务
- 线程调度方式
2 作用
- 复用线程资源
- 节省线程创建的销毁的开销
- 可以异步处理生产者发布的任务
- 提高处理多个任务的效率
3 封装
3.1 变量对象
- 任务队列
- 工作线程数组（向量）
- 互斥锁
- 条件变量
- 状态标志
- 线程数量
- 原子变量（任务数量）
3.2 任务结构体
struct TaskFunc { TaskFunc(expireTime) : _expireTime(expireTime) {} std::function<void()> _func; uint64_t _expireTime = 0; } typedef std::shared_pte<TaskFunc> TaskFuncPtr;
3.3 构造函数
- 初始化线程数量
- 初始化状态标志
ThreadPool::ThreadPool() :thread_num_(1), terminate_(false) {}
3.4 析构函数
- 执行Stop函数
ThreadPool::~ThreadPool() { Stop(); }
3.5 线程池初始化
- 支持手动设置线程数量
bool ThreadPool::Init(int64_t num) { std::unique_lock<std::mutex> lock(mutex_); if(!threads_.empty()) return false; thread_num_ = num; return true; }
3.6 线程池停止
- 修改状态标志
- 唤醒所有线程，让他们在后台执行完后退出，释放
- 清空线程数组
void ThreadPool::Stop() { { std::unique_lock<std::mutex> lock(mutex_); terminate_ = true; condition_.notify_all(); } for(int i = 0; i < threads_.size(); i++) { if(threads_[i]->joinable()) threads_[i]->join(); delete threads_[i]; threads_[i] = nullptr; } std::unique_lock<std::mutex> lock(mutex_); threads_.clear(); }
3.7 获取线程池线程数量
int ThreadPool::GetThreadNum() { return threads_.size(); }
3.8 获取任务数量
int ThreadPool::GetTaskNum() { return tasks_.size(); }
3.9 线程池开启
- 创建thread_num个线程
- 用Run函数作为线程的执行函数
bool ThreadPool::Start() { std::unique_lock<std::mutex> lock(mutex_); if(!threads_.empty()) return false; for(size_t i = 0; i < thread_num_; i++) { threads_.push_back(new std::thread(&ThreadPool::Run, this)); } return true; }
3.10 传入任务
- 将（限定时间），任务函数，任务呢函数添加进任务队列
auto Exec(F &&f, Args... args) -> std::future<decltype(f(args...))> { Exec(0, f, args...); }
3.11 等待函数
- 如果任务队列为空，返回
- 如果传入时间小于0，等到所有任务执行完
- 如果传入时间大于0，规定时间执行任务，返回队列是否为空
bool ThreadPool::WaitForAllDone(int timewait) { std::unique_lock<std::mutex> lock(mutex_); if(tasks_.empty()) return true; if(timewait < 0) { condition_.wait(lock, [this]{return tasks_.empty();}); return true; } else { return condition_.wait_for(lock, std::chrono::milliseconds(timewait), [this]{return tasks_.empty();}); } }
3.12 获取任务
- 如果任务队列为空，释放互斥锁等待；直到有任务或者认为 停止线程池
- 如果任务队列不为空，获取并且释放一个任务
bool ThreadPool::Get(TaskFuncPtr &task) { std::unique_lock<std::mutex> lock(mutex_); if(tasks_.empty()) { condition_.wait(lock, [this] { return terminate_ || !tasks_.empty(); }); } if(terminate_) return false; if(!tasks_.empty()) { task = std::move(tasks_.front()); tasks_.pop(); return true; } return false; }
3.13 执行任务
- 如果成功获取任务，执行任务+1,执行任务
- 执行任务结束后，执行任务-1
- 通知等待函数
void ThreadPool::Run() { while (!terminate_) { TaskFuncPtr task; bool ok = Get(task); if(ok) { atomic_++; try { if(task->_expireTime > 0 && task->_expireTime > GetNowMs()) { //任务超时处理 } else task->_func; } catch (...) {} atomic_--; std::unique_lock<std::mutex> lock(mutex_); if(atomic_ == 0 && tasks_.empty()) condition_.notify_all(); } } }
