当前位置：首页 > news >正文

基于生产者消费者模型的线程池【Linux操作系统】

news 2025/7/12 15:02:10

文章目录

线程池的模拟实现
- 线程池的简单介绍
- 成员变量
- 成员函数
- - 构造函数：
  - Equeue：生产任务
  - HandlerTask：消费任务
  - Start：启动线程池的线程
  - Wait：等待并回收线程池的线程
  - Stop：让线程池中的线程退出
- 线程池的单例化
- 代码实现：
- - class BlocQueue
- ThreadPool

线程池的模拟实现

线程池的简单介绍

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

线程池的应用场景：

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

如果来一个任务就创建一个线程，虽然不是不行，但是创建线程也需要花费时间
所以我们可以预先创建一个线程池，线程池里面维护了一批线程，他们在没有任务的时候会阻塞等待有任务的时候只需要唤醒它们就行了

在这里插入图片描述

一般线程池中的线程总个数不会太多，并且我们可以控制线程的个数

①线程个数太多，其实效率提升也并不明显，因为可以并行的线程是有上限的（即cpu的核数有上限）
甚至可能因为切换调度成本上升，而效率下降
②就算突然出现大量任务，线程池中的线程个数也不会有太大波动，只是会在任务队列里面堆积很多任务
这样可以一定程度上保证系统的稳定性
③线程池适用于：任务量多，并且单个任务执行时长较短的场景

成员变量

①int num：线程池中线程的个数
②queue<T>：线程池的任务队列，所以线程池也得搞成一个模板类，T表示任务的类型
③vector：存放指向线程池中所有线程的智能指针
④mutex：锁
⑤cond：条件变量
⑥wait_num：有线程池中有几个线程（消费者）在条件变量的等待队列中等待
这个成员变量，主要是为了方便生产者线程，生产数据之后，进行唤醒
⑦isrunning：标识线程池是否在运行
线程池调用构造函数时，isrunning默认设置为false
只有当第一次调用（Start只能被调用一次，所以isrunning如果为true，Start会直接return）线程池的Start方法之后，isrunning才会为true

成员函数

构造函数：

运行步骤：

①初始化wait_num为0
②初始化isrunning为false
③创建num个线程

ThreadPool(u_int32_t thread_num = default_num, u_int32_t cap = default_cap): _thread_num(thread_num),_isrunning(false),_rq(cap){}

Equeue：生产任务

运行步骤：

①加锁
②判断isrunning是否为true，如果不是就直接return，因为线程池都没启动，当然不能生产任务
③调用任务队列的push
④判断wait_num是否>0，如果大于0，就说明线程池中还有再条件变量的等待队列中阻塞等待，就唤醒

HandlerTask：消费任务

（注意：这个函数是private私有函数）

①whlie（true）保证线程不退出，因为一直要等待/处理任务，直到线程池关闭
②加锁
③while判断任务队列是否为空&&判断isrunning是否为true
- 1.只有任务队列为空并且isrunning为ture
  才
  wait_num++，再去条件变量下等，将来被成功唤醒的时候，再wait_num–
  因为如果isruning为false，就说明线程池要退出了，如果这个时候任务队列为空，线程就应该直接退出了，而不是在去条件变量下阻塞
- 2.其他情况都统一不阻塞等待
④if（队列为空&&isrunning==false）
如果这个判断成立，那么执行这个判断的线程就应该要自己退出了，即使用break跳出
HandlerTask的while（true）循环
⑤调用任务队列的front和pop
⑥调用任务类的operator（）处理任务

Start：启动线程池的线程

①加锁，因为isrunning是可能被修改的共享资源
②判断isrunning是否为true，如果是true就直接return，防止重复启动
③将isrunning设置为true，不能在for循环之后才将isrunning设置为true，因为如果这样的话，在创建线程3的时候，线程1已经进入HandlerTask里面了，此时isrunning为false，如果任务队列还为空的话，线程1就直接退了
④解锁
⑤使用for循环启动所有线程（即让线程池里面的线程们开始执行HandlerTask函数，开始从任务队列里拿任务，做任务）

bool Start(){_rq.CancelExternalSignal();//取消外部唤醒状态{lock_guard<mutex> m(_mutex); // 锁,保护isruningif (_isrunning == true)          // 不许重复启动{return false;}_isrunning = true;cout << "启动线程池" << endl;}for (int i = 0; i < _thread_num; i++) // 创建_thread_num个执行任务的线程{string s = "线程";s += to_string(i + 1);_threads.emplace_back([this](const string &name){ this->PerfTask(name); }, s); // 线程去PerfTask中等待任务cout << "创建线程" << i + 1 << "成功" << endl;}return true;}

Wait：等待并回收线程池的线程

主线程使用for循环等待线程池中的所有线程

void Wait()//回收线程{for(auto &t:_threads){t.join();}_threads.clear();//清除线程对象}

Stop：让线程池中的线程退出

是暂停线程池中的线程，不是取消，暂停是还可以通过调用Start恢复运行的不能直接取消线程池中的线程
因为调用Stop的时候，可能线程池中的线程还在处理任务/阻塞，直接取消线程，是非常不负责任的行为，因为有些任务可能很重要

所以线程池暂停的前提条件是

1 .不能再生产任务了，把isrunning设置为false就行了
2.任务队列为空，即历史的任务全部处理完了
3.线程自己退出，因为调用Stop的时候，有的线程可能还在条件变量的等待队列中阻塞着，必须把它唤醒，它才能自己安全退出

具体实现：

①加锁，因为isrunning是可能被修改的共享资源
②判断线程是否是运行状态，即isrunning是否为true，为true才可以Stop
③把iisrunning设置为false，这样如果还有线程调用Equeue：生产任务也生产不了任务，会直接return
④想让线程自己退出必须满足3个条件
- 1.isrunning为false
- 2.任务队列为空
- 3.线程没有被阻塞在条件变量那里，可以执行代码

所以当wait_num>0时，还得广播唤醒一次在条件变量的等待队列中阻塞的线程

为什么广播唤醒一次就够了？
因为去条件变量下等待的while判断是：任务队列为空并且isrunning为false
所以只要任务队列不为空，所有线程就都不会去条件变量下等
而当任务队列为空时，因为isrunning也为false，所以线程就自己退出了

bool Stop(){lock_guard<mutex> m(_mutex); // 锁,保护isruning_isrunning = false;_rq.ExternalSignal();//外部唤醒  阻塞队列中的  阻塞等待的线程cout << "关闭线程池" << endl<< endl<< endl;return true;}

线程池的单例化

我们之前说过，一个进程中的线程不应该太多，线程池的线程不应该太多
因为线程太多了并没有意义，对效率提升不明显，因为CPU的个数和核数就那么多

为什么要让线程池单例化？
因为一个线程池如果设定它有5个线程，如果创建n个线程池，线程的个数就会有5n个

所以用户如果无意识地创建了多个线程池对象，会导致线程的个数偏多

所以我们直接让线程池单例化，让用户无论如何都最多只能创建一个线程池对象

线程池单例化的线程安全问题

①饿汉模式是在线程池类中，直接定义一个static修饰的线程池对象，在进程启动的时候就已经定义了，所以根本不可能存在线程安全的问题
②懒汉模式是在线程池类中，定义一个static修饰指针，然后通过一个static修饰的成员函数new出那唯一的对象

因为static修饰的成员指针是共享资源，而且可能同时有多个线程去访问它，此时就可能因为线程安全问题，导致创建了多个线程池对象

所以我们要给那个获取对象的static成员函数加锁：
在这里插入图片描述

但是如果那唯一的一个线程池对象已经被创建了的话，每次调用这个函数如果都还是要加锁才能获取到那唯一的一个对象的指针
效率就会很低，因为加锁之后是互斥的，也就是说不能并行调用成员函数获取线程池对象的指针

那怎么办？
其实线程安全问题出现在，最开始的时候，static线程池对象指针为nullptr的时候，可能会有多个线程同时通过if判断，进而创建出多个线程池对象

当这个唯一的一个线程池对象被new出来之后，再调用这个静态成员函数，就只是获取指针，并不会修改，而且其他的地方里面也不可能修改成功私有的成员指针，所以不会有线程安全问题

所以我们只需要在私有static线程池指针为空时，才加锁就行了
即：
在这里插入图片描述

代码实现：

class BlocQueue

#include <iostream>
#include <queue>
#include <pthread.h>using namespace std;template <class T>
class BlocQueue
{static const u_int32_t default_capacity = 10;bool isExternalSignal(){pthread_mutex_lock(&_signal_mutex); //保护is_external_signalbool tmp=_is_external_signal==true;pthread_mutex_unlock(&_signal_mutex); return tmp;}
public:BlocQueue(int c = default_capacity): _capacity(c),_cwait(0),_pwait(0),_is_external_signal(false){pthread_mutex_init(&_mutex, nullptr);pthread_mutex_init(&_size_mutex, nullptr);pthread_mutex_init(&_signal_mutex, nullptr);pthread_cond_init(&_ccond, nullptr);pthread_cond_init(&_pcond, nullptr);}bool Push(const T &in) // 生产{pthread_mutex_lock(&_mutex);while (isFull()) // 如果阻塞队列满了{if(isExternalSignal())//如果外部唤醒了{pthread_mutex_unlock(&_mutex);return false;//线程退出该函数}_pwait++;pthread_cond_wait(&_pcond, &_mutex); // 阻塞等待_pwait--;}pthread_mutex_lock(&_size_mutex); // 保护阻塞队列的size_q.push(in); // 生产数据pthread_mutex_unlock(&_size_mutex);if (_cwait > 0)pthread_cond_broadcast(&_ccond); // 唤醒消费者线程pthread_mutex_unlock(&_mutex);return true;}bool Pop(T &out) // 消费{pthread_mutex_lock(&_mutex);while (isEmpty()) // 如果阻塞队列为空{if(isExternalSignal())//如果外部唤醒了{pthread_mutex_unlock(&_mutex);return false;//线程退出该函数}_cwait++;pthread_cond_wait(&_ccond, &_mutex); // 阻塞等待_cwait--;}pthread_mutex_lock(&_size_mutex); // 保护阻塞队列的size// 消费数据out = _q.front();_q.pop();pthread_mutex_unlock(&_size_mutex);if (_pwait > 0)pthread_cond_broadcast(&_pcond); // 唤醒生产者线程pthread_mutex_unlock(&_mutex);return true;}bool isFull(){pthread_mutex_lock(&_size_mutex); // 保护阻塞队列的sizebool tmp = _q.size() == _capacity;pthread_mutex_unlock(&_size_mutex); // 保护阻塞队列的sizereturn tmp;}bool isEmpty(){pthread_mutex_lock(&_size_mutex); // 保护阻塞队列的sizebool tmp=_q.size()==0;pthread_mutex_unlock(&_size_mutex); // 保护阻塞队列的sizereturn tmp;}void ExternalSignal()//进入外部唤醒状态{pthread_mutex_lock(&_signal_mutex); //保护is_external_signal_is_external_signal=true;pthread_cond_broadcast(&_pcond); // 唤醒生产者线程pthread_cond_broadcast(&_ccond); // 唤醒消费者线程pthread_mutex_unlock(&_signal_mutex); }void CancelExternalSignal()//取消外部唤醒状态{pthread_mutex_lock(&_signal_mutex); //保护is_external_signal_is_external_signal=false;pthread_mutex_unlock(&_signal_mutex); }~BlocQueue(){pthread_mutex_destroy(&_mutex);pthread_mutex_destroy(&_size_mutex);pthread_mutex_destroy(&_signal_mutex);pthread_cond_destroy(&_pcond);pthread_cond_destroy(&_ccond);}private:queue<T> _q;             // 阻塞队列u_int32_t _capacity;     // 队列容量bool _is_external_signal; // 是否需要外部唤醒pthread_mutex_t _mutex;      // 锁,保护阻塞队列pthread_mutex_t _size_mutex; // 锁，保护阻塞队列的sizepthread_mutex_t _signal_mutex; // 锁，保护is_external_signalpthread_cond_t _pcond; // 生产者条件变量pthread_cond_t _ccond; // 消费者条件变量u_int32_t _pwait;      // 生产者条件变量下阻塞等待的线程个数u_int32_t _cwait;      // 消费者条件变量下阻塞等待的线程个数
};

ThreadPool

#include <iostream>
#include <thread>
#include <functional>
#include <vector>
#include <mutex>
#include <string>
#include "prod.hpp"using namespace std;template <class T>
class ThreadPool
{static const u_int32_t default_num = 5;  // 默认线程个数static const u_int32_t default_cap = 10; // 默认生产者消费者模型的环形队列的最大容量bool isRunning(){lock_guard<mutex> m(_mutex); // 锁,保护isrunningreturn _isrunning == true;}void PerfTask(const string &name)//获取并执行任务{while (true){T task;if (_rq.isEmpty()) // 如果阻塞队列为空{if (!isRunning()) // 如果线程池  没有运行{cout << name << "退出" << endl;return; // 线程退出}}cout << name << "去获取任务....." << endl;bool rval=_rq.Pop(task); // 从阻塞队列里面拿任务if(rval){cout << name << "任务获取成功" << endl;task(); // 执行任务}else{cout << name << "任务获取失败" << endl;}}}ThreadPool(u_int32_t thread_num = default_num, u_int32_t cap = default_cap): _thread_num(thread_num),_isrunning(false),_rq(cap){}ThreadPool(const ThreadPool<T>&obj)=delete;ThreadPool<T>& operator=(const ThreadPool<T>&obj)=delete;public:static ThreadPool<T>* GetInstance(u_int32_t thread_num = default_num, u_int32_t cap = default_cap){if(_This==nullptr){lock_guard<mutex> m(_mutex); // 锁,保护_This_This=new ThreadPool<T>(thread_num,cap);}return _This;     }bool Start(){_rq.CancelExternalSignal();//取消外部唤醒状态{lock_guard<mutex> m(_mutex); // 锁,保护isruningif (_isrunning == true)          // 不许重复启动{return false;}_isrunning = true;cout << "启动线程池" << endl;}for (int i = 0; i < _thread_num; i++) // 创建_thread_num个执行任务的线程{string s = "线程";s += to_string(i + 1);_threads.emplace_back([this](const string &name){ this->PerfTask(name); }, s); // 线程去PerfTask中等待任务cout << "创建线程" << i + 1 << "成功" << endl;}return true;}void Wait()//回收线程{for(auto &t:_threads){t.join();}_threads.clear();//清除线程对象}bool Stop(){lock_guard<mutex> m(_mutex); // 锁,保护isruning_isrunning = false;_rq.ExternalSignal();//外部唤醒  阻塞队列中的  阻塞等待的线程cout << "关闭线程池" << endl<< endl<< endl;return true;}bool PushTask(const T &task){if (isRunning()) // 线程池运行才能生产{_rq.Push(task); // 生产任务cout << "主线程生产任务" << endl;}else{return false;}return true;}private:static ThreadPool<T>* _This;u_int32_t _thread_num; // 线程总个数BlocQueue<T> _rq;      // 基于阻塞队列的生产者消费者模型bool _isrunning;       // 是否运行static mutex _mutex;      // 锁,保护_isruningvector<thread> _threads; // 管理线程
};template<class T>
ThreadPool<T>* ThreadPool<T>::_This=nullptr;template<class T>
mutex ThreadPool<T>::_mutex;