227-C++的线程池

怼烎@ 2022-11-09 11:23 143阅读 0赞

**我们有两种线程池的实现：一种是半同步半异步，一种是领导者追随者线程池。**  
**领导者追随者线程池**：当某一个领导者作为领导者，它来响应任务。响应完，把追随者的一个线程变为领导者，之后的线程还是追随者。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xJTlpFWVU2NjY_size_16_color_FFFFFF_t_70_pic_center]  
**我们还有内存池**：预先开辟一些内存，你请求的时候从预先内存中分配出去，不需要malloc函数进行系统调用。当我用完，把内存还给池子，并没有真正的free掉。

## 半同步半异步线程池 ##

分为三层：**同步服务层（添加任务到同步队列）**，**排队层（同步队列）**，（就比如客户一客户二客户三从门口进入银行，不能一窝蜂到窗口，首先要叫号，相当于排队），**异步服务层（任务处理，从同步队列取出任务，交给执行线程）**

**相当于生产者消费者模型。在这里，同步服务层相当于生产者，异步服务层相当于消费者。**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xJTlpFWVU2NjY_size_16_color_FFFFFF_t_70_pic_center 1]  
![在这里插入图片描述][20210314213115407.png_pic_center]  
同步（线之间要有一定的次序感，只有把任务添加到队列，消费者才可以从队列中取任务），同步服务1，同步服务2，同步服务3向同步队列里添加，这里的同步服务1，2 ，3之间是互斥的关系。添加任务，1添加，2和3不能添加。不能同时塞进队列，会出现数据覆盖。  
异步服务唤起工作线程去处理任务。工作线程1和工作线程2是没有耦合的，他们处理任务，他们之间是没有关系的。

![在这里插入图片描述][20210314213555233.png_pic_center]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xJTlpFWVU2NjY_size_16_color_FFFFFF_t_70_pic_center 2]

**同步的过程是生产者和消费者这种模型。**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xJTlpFWVU2NjY_size_16_color_FFFFFF_t_70_pic_center 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xJTlpFWVU2NjY_size_16_color_FFFFFF_t_70_pic_center 4]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xJTlpFWVU2NjY_size_16_color_FFFFFF_t_70_pic_center 5]

## 半同步半异步线程池的代码 ##

#include<iostream>
    #include<thread>
    #include<mutex>
    #include<condition_variable>
    #include<semaphore>
    #include <queue>
    #include <atomic>
    #include<list>
    using namespace std;
    
    template<class T>//定义模板类
    class SyncQueue //同步队列
    {
        
    public:
    	SyncQueue(int maxSize) :m_maxSize(maxSize), m_needStop(false) {
        }//构造函数
    	void Put(const T& x)//添加任务，常引用
    	{
        
    		Add(x);
    	}
    	void Put(T&& x)//添加任务，右值引用
    	{
        
    		Add(std::forward<T>(x));//完美转发
    	}
    
    	void Take(std::list<T>& list)//获取任务，把整个队列所有的任务一次性提走
    	{
        
    		std::unique_lock<std::mutex> locker(m_mutex);
    		while (!m_needStop && Is_Empty())//同步队列工作中并且缓冲区队列空了
    		{
        
    			m_notEmpty.wait(locker);//不能获取任务，线程挂起！
    		}
    		if (m_needStop)//如果同步队列停止工作了
    		{
        
    			return;//退出函数
    		}
    		list = std::move(m_queue);//如果同步队列工作中，就取走整个队列的所有任务
    		m_notFull.notify_one();//唤醒因为队列满而不能添加任务而阻塞的线程
    	}
    	void Take(T& t)//获取任务，每一次只提取队列中的一个任务
    	{
        
    		std::unique_lock<std::mutex> locker(m_mutex);
    		while (!m_needStop && Is_Empty())//同步队列工作中并且缓冲区队列空了
    		{
        
    			m_notEmpty.wait(locker);//不能获取任务，线程挂起！
    		}
    		if (m_needStop)//如果同步队列停止工作了
    		{
        
    			return;//退出函数
    		}
    		t = m_queue.front(); //获取队列的第一个任务
    		m_queue.pop_front();//将第一个任务出队列
    		m_notFull.notify();//唤醒因为队列满而不能添加任务而阻塞的线程
    	}
    	void Stop()//让同步队列停止工作
    	{
        
    		{
        
    			std::unique_lock<std::mutex> locker(m_mutex);//需要加锁
    			m_needStop = true;//改为真
    		}
    		//停止工作，有可能还有阻塞的线程，要唤醒起来！
    		m_notFull.notify_all();//唤醒线程
    		m_notEmpty.notify_all();//唤醒线程
    	}
    	bool Empty() const//判空
    	{
        
    		std::unique_lock<std::mutex> locker(m_mutex);
    		return m_queue.empty();
    	}
    	bool Full() const//判满
    	{
        
    		std::unique_lock<std::mutex> locker(m_mutex);
    		return m_queue.size() >= m_maxSize;
    	}
    	bool Size() const//获取队列的任务数
    	{
        
    		std::unique_lock<std::mutex> locker(m_mutex);
    		return m_queue.size();
    	}
    	int Count() const//获取队列的任务数
    	{
        
    		return m_queue.size();
    	}
    
    private:
    	bool Is_Full() const//判断缓冲区是否为满，需要等待的是同步服务层
    	{
        
    		bool full = (m_queue.size() >= m_maxSize);
    		if (full)
    		{
        
    			cout << " 缓冲区满了,需要等待..... 同步服务的线程ID: "<< this_thread::get_id() << endl;
    		}
    		return full;
    	}
    	bool Is_Empty() const//判断缓冲区是否为空，需要等待的是异步服务层
    	{
        
    		bool empty = m_queue.empty();
    		if (empty)
    		{
        
    			cout << "缓冲区空了,需要等待.... 异步服务的线程ID： " << this_thread::get_id() << endl;
    		}
    		return empty;
    	}
    	bool NOtFull() const {
         return !Is_Full(); }//判断是不是没有满
    	bool NotEmpty() const {
        	return !Is_Empty();  }//判断是不是没有空
    
    	template<class U>
    	void Add(U&& x)//往缓冲区添加任务，右值引用，尽可能产生少的拷贝
    	{
        
    		std::unique_lock<std::mutex> locker(m_mutex);
    		while (!m_needStop && Is_Full())//同步队列在工作中并且缓冲区满了
    		{
        
    			m_notFull.wait(locker);//不能添加任务，挂起线程！
    		}
    		if (m_needStop)//如果同步队列停止工作了
    		{
        
    			return;//退出函数
    		}
    		m_queue.push_back(x);//如果同步队列要工作，就入队
    		m_notEmpty.notify_one();//唤醒一个在等待获取任务而阻塞的线程
    	}
    	std::list<T> m_queue; // 缓冲区，放任务，取任务的
    	std::mutex m_mutex;//互斥量
    
    	//定义两个条件变量
    	std::condition_variable  m_notEmpty;//取任务的时候，有可能空了，得阻塞到条件变量的阻塞队列
    	std::condition_variable  m_notFull;	//放任务的时候，有可能满了，得阻塞到条件变量的阻塞队列
    	//当我们取得任务，就要把不能添加任务的线程唤醒。当我们放任务，就要把不能取任务的线程唤醒
    	size_t m_maxSize;//队列可以放的最大任务量
    	bool m_needStop;//这个任务队列是否可以工作
    };
    //同步队列是核心控制点
    
    
    class MyTask//我的任务，可以派生出很多不同的任务
    {
        
    public:
    	virtual void run()
    	{
        
    		cout << "MyTask" << endl;
    	}
    };
    
    
    const int MaxTaskCount = 10;//最大的任务个数，同步队列最大任务数10个。
    
    
    class ThreadPool//线程池
    {
        
    public:
    	ThreadPool(int numThreads = std::thread::hardware_concurrency())//构造函数，获得CPU的内核的个数，获得的是12个。
    		:m_queue(MaxTaskCount)//最大任务个数10去初始化
    	{
        
    		Start(numThreads);//启动函数，启动运行环境
    	}
    	~ThreadPool()//析构函数
    	{
        
    		Stop();//停止工作
    	}
    	void Stop()//停止线程池工作
    	{
        
    		std::call_once(m_flag, [this] {
        StopThreadGroup(); });//调动停止工作方案，只能调动1次哦！
    	}
    	void AddTask(MyTask*&& task)//添加任务
    	{
        
    		m_queue.Put(task);
    	}
    	void AddTask(const MyTask*& task)//添加任务
    	{
        
    		//m_queue.Put(task);
    	}
    
    private:
    	void Start(int numThreads)//启动线程
    	{
        
    		m_running = true;//线程池可以工作，处于运行的状态
    		for (int i = 0; i < numThreads; ++i)
    		{
        
    			m_threadgroup.push_back(std::make_shared<std::thread>(&ThreadPool::RunInThread, this));
    			//创建线程对象，运行函数RunInThread，参数是当前对象，添加到线程组
    		}
    	}
    	void  RunInThread()//此成员函数被线程化
    	{
        
    		while (m_running)//线程池可以工作，处于运行的状态
    		{
        
    			//MyTask* sp = nullptr;
    			//m_queue.Take(sp);
    			//sp->run();
    			std::list<MyTask*> list;
    			m_queue.Take(list);//把整个队列的所有任务取出来到list上
    			for (auto& x : list)//对容器元素遍历
    			{
        
    				if (!m_running)//如果线程池停止工作了
    				{
        
    					return;//退出函数
    				}
    				x->run();//线程池工作中，调用运行任务的函数run
    			}
    		}
    	}
    
    	void StopThreadGroup()//停止线程组的工作
    	{
        
    		m_queue.Stop();//先把同步队列停止
    		m_running = false;//线程池运行状态变为false
    		for (auto thread : m_threadgroup)//对线程组的线程遍历
    		{
        
    			if (thread)//如果当前线程处于活动中，用共享型智能指针实现
    			{
        
    				thread->join();//等待线程结束
    			}
    		}
    		m_threadgroup.clear();//调动线程组的清空函数
    	}
    
    	std::list<std::shared_ptr<std::thread> > m_threadgroup;//线程组，放着一个个的线程
    	SyncQueue<MyTask*> m_queue; //创建同步队列
    	atomic_bool m_running;//原子操作，线程池是否停止工作，运行标记
    	std::once_flag m_flag;//单次调动，无论在多线程还是单线程，只能调动1次。
    };
    
    
    void funa(ThreadPool& pool)//同步线程funa
    {
        
    	for (int i = 0; i < 10; ++i)
    	{
        
    		auto thdId = this_thread::get_id();//获得当前线程的ID号
    		cout << "同步线程1 的ID: " << thdId << endl;
    		pool.AddTask(new MyTask());//添加任务
    	}
    }
    void funb(ThreadPool& pool)//同步线程funb
    {
        
    	for (int i = 0; i < 10; ++i)
    	{
        
    		auto thdId = this_thread::get_id();//获得当前线程的ID号
    		cout << "同步线程2 的ID: " << thdId << endl;
    		pool.AddTask(new MyTask());//添加任务
    	}
    }
    int main()
    {
        
    	ThreadPool pool;//创建线程池
    	std::thread thd1(funa, std::ref(pool));//拿线程池引用传进去
    	std::thread thd2(funb, std::ref(pool));//拿线程池引用传进去
    
    	this_thread::sleep_for(std::chrono::seconds(20));//睡眠20秒
    	pool.Stop();//线程池停止工作
    
    	thd1.join();
    	thd2.join();
    	return 0;
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xJTlpFWVU2NjY_size_16_color_FFFFFF_t_70_pic_center]: /images/20221023/9e0edac7e5f744ffabb5d9c4fe7b0d71.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xJTlpFWVU2NjY_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221023/dacdc4de646e4e2d8aad3323869a8fea.png
[20210314213115407.png_pic_center]: /images/20221023/67a51554f4384863b2ba0db0c2bca3b5.png
[20210314213555233.png_pic_center]: /images/20221023/4b4c898851d74c34b3d74d9f69759d25.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xJTlpFWVU2NjY_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221023/d2a97750487e4ba79345baa55b9359d3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xJTlpFWVU2NjY_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221023/f9924cecbab34965aae4fb4fcdfe6f16.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xJTlpFWVU2NjY_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221023/6386ea30f82f4ee58aea6e9cd45cb71c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xJTlpFWVU2NjY_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221023/bc909158d8ff4fb8a8dd59dd9496e3d4.png