线程池设计

线程池:

        ⼀种线程使⽤模式。线程过多会带来调度开销，进⽽影响缓存局部性和整体性能。⽽线程池维护着多个线程，等待着监督管理者分配可并发执⾏的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利⽤，还能防⽌过分调度。可⽤线程数量应该取决于可⽤的并发处理器、处理器内核、内存、⽹络sockets等的数量。

线程池的本质是生产者消费者模型

线程池的应用场景：

• 需要大量的线程来完成任务，且完成任务的时间⽐较短。⽐如WEB服务器完成⽹⻚请求这样的任务，使⽤线程池技术是⾮常合适的。因为单个任务⼩，⽽任务数量巨⼤，你可以想象⼀个热⻔⽹站的点击次数。但对于⻓时间的任务，⽐如⼀个Telnet连接请求，线程池的优点就不明显了。因为 Telnet会话时间⽐线程的创建时间⼤多了
• 对性能要求苛刻的应用，⽐如要求服务器迅速响应客⼾请求。
• 接受突发性的⼤量请求，但不至于使服务器因此产生大量线程的应用。突发性⼤量客⼾请求，在没 有线程池情况下，将产⽣⼤量线程，虽然理论上⼤部分操作系统线程数⽬最⼤值不是问题，短时间 内产⽣⼤量线程可能使内存到达极限，出现错误.

线程池的种类

1. 创建固定数量线程池，循环从任务队列中获取任务对象，获取到任务对象后，执行任务对象中 的任务接⼝
2. 浮动线程池，其他同上

此处，我们选择固定线程个数的线程池。

线程池的主要成员:

1. 任务队列(存放接收任务)

2. vector管理线程

线程池退出步骤(问题):

1. 先保证不再继续向任务队列中进任务
2. 唤醒所有休眠的线程 ,让其处理完剩余任务

线程池初步   下一步就是向任务队列中入任务

#pragma once
#include <queue>
#include <iostream>
#include <string>
#include <vector> //用vector管理线程
#include <memory>
#include "log.hpp"
#include "Thread.hpp"
#include "Mutex.hpp"
#include "Cond.hpp"namespace ThreadPoolModule
{using namespace LockModule;using namespace ThreadModule;using namespace CondModule;using namespace LogModule;const static int defaultnum = 5;using thread_t = std::shared_ptr<Thread>;void Defaulttest(){while (true){LOG(LogLevel::INFO) << "test";sleep(1);}}template <typename T>class ThreadPool{public:ThreadPool(int num = defaultnum): _num(num){// 创建num个线程for (int i = 0; i < num; i++){_threads.push_back(std::make_shared<Thread>(Defaulttest));LOG(LogLevel::INFO) << "构建线程"<<_threads.back()->Name()<<"成功";}}~ThreadPool(){}//线程池启动void Start(){for(auto &thread_ptr: _threads){thread_ptr->Start();LOG(LogLevel::INFO) << "启动线程"<<thread_ptr->Name()<<"成功";}}//线程池回收线程void Wait(){for(auto &thread_ptr: _threads){thread_ptr->Join();}}//向线程池中传入任务void Enqueue(){}void Stop(){}private:int _num;                       // 线程个数std::queue<T> _taskq;           // 任务队列std::vector<thread_t> _threads; // 管理线程 ,其中是线程的指针};
}

测试

#include"ThreadPool.hpp"
#include<memory>using namespace ThreadPoolModule;int main()
{ENABLE_CONSOLE_LOG();std::unique_ptr<ThreadPool<int>> tp =std::make_unique<ThreadPool<int>>();tp ->Start();tp ->Wait();//tp->Enqueue();//tp->Stop();return 0;
}

最终实现

ThreadPool.hpp

#pragma once
#include <queue>
#include <iostream>
#include <string>
#include <vector> //用vector管理线程
#include <memory>
#include "log.hpp"
#include "Thread.hpp"
#include "Mutex.hpp"
#include "Cond.hpp"namespace ThreadPoolModule
{using namespace LockModule;using namespace ThreadModule;using namespace CondModule;using namespace LogModule;const static int defaultnum = 5;using thread_t = std::shared_ptr<Thread>;void Defaulttest(){while (true){LOG(LogLevel::INFO) << "test";sleep(1);}}template <typename T>class ThreadPool{private:bool IsEmpty(){return _taskq.empty();}// 执行任务void HandlerTask(std::string name){LOG(LogLevel::INFO)<<"线程"<< name <<"进入handlerTask";// 线程醒来就一直执行while (true){T t;{LockGuard lockguard(_mutex);while (IsEmpty() && _isrunning){_wait_num++;_cond.Wait(_mutex);_wait_num--;}// 任务队列为空 && 线程池退出了才退出    ,重要重要 if(IsEmpty() && !_isrunning)break;// 1.拿任务t = _taskq.front();_taskq.pop();}//在临界区外 ,处理任务 ,效率更高// 2.处理任务   规定传入的所有的任务, 必须提供()方法t(name);}LOG(LogLevel::INFO) << "线程: " << name << " 退出";}public:ThreadPool(int num = defaultnum): _num(num),_wait_num(0),_isrunning(false){// 创建num个线程for (int i = 0; i < num; i++){//bind  此时所有创建出来的线程,转而去执行HandlerTask_threads.push_back(std::make_shared<Thread>(std::bind(&ThreadPool::HandlerTask, this,std::placeholders::_1 )));LOG(LogLevel::INFO) << "构建线程" << _threads.back()->Name() << "成功";}}~ThreadPool(){}void Start(){if(_isrunning) return;_isrunning = true; // bug fix??for (auto &thread_ptr : _threads){thread_ptr->Start();LOG(LogLevel::INFO) << "启动线程" << thread_ptr->Name() << "成功";}}void Wait(){for (auto &thread_ptr : _threads){thread_ptr->Join();LOG(LogLevel::INFO) << "回收线程" << thread_ptr->Name() << "成功";}}//任务入队列void Enqueue(T &&in)//这个会被多线程调用 ,先加锁{//只要队列扛得住 ,就一直加LockGuard lockguard(_mutex);if(!_isrunning) return;_taskq.push(std::move(in));if(_wait_num > 0 ) _cond.Notify();}//退出线程池void Stop(){LockGuard lockguard(_mutex);if(_isrunning){// 3. 不能在入任务了_isrunning = false; // 不工作// 1. 让线程自己退出(要唤醒) && // 2. 历史的任务被处理完了if(_wait_num>0)_cond.NotifyAll();}}private:int _num;                       // 线程个数std::queue<T> _taskq;           // 任务队列  是临界资源std::vector<thread_t> _threads; // 管理线程 ,其中是线程的指针Mutex _mutex;Cond _cond;int _wait_num;bool _isrunning ;               //线程池的运行状态};
}

Main.cc

#include"ThreadPool.hpp"
#include<memory>
#include"Task.hpp"
using namespace ThreadPoolModule;int main()
{ENABLE_CONSOLE_LOG();//创建线程池std::unique_ptr<ThreadPool<task_t>> tp =std::make_unique<ThreadPool<task_t>>();//启动线程池tp ->Start();//传入任务int cnt =15;while(cnt--){tp->Enqueue(Push);sleep(1);}//线程池退出tp->Stop();//线程池回收线程tp ->Wait();return 0;
}