文章目录
前言
生产者消费者模型和读者写者模型这些模型是用于在线程间协调和管理资源访问的模式, 我们在之前已经理解了线程的概念以及同步与互斥, 现在我们来学习几个常见的线程实用场景
生产者消费者模型
这是一个经典的线程同步问题，通常用来解决多线程间的缓冲区共享问题。生产者线程负责生成数据并将其放入缓冲区，而消费者线程从缓冲区中取出数据进行处理。为了防止缓冲区的竞争访问，通常使用信号量、互斥锁或条件变量等同步机制来管理线程之间的协调
1.基于阻塞队列
阻塞队列是一种线程安全的数据结构，广泛用于多线程编程中
特点
线程安全
阻塞队列通常是线程安全的，可以在多个线程间安全地进行操作。无论是多个生产者线程还是多个消费者线程，都可以并发地进行数据的放入或取出操作，而不会引发竞态条件。
阻塞操作
阻塞的放入操作：如果队列已满，当生产者线程试图将数据放入队列时，它会被阻塞，直到队列有空余空间。
阻塞的取出操作：如果队列为空，当消费者线程试图从队列中取出数据时，它会被阻塞，直到队列中有数据可取。
实现
完整代码(GitHub)
整体框架
template<class T>//资源类型
class blockqueue
{
public:
blockqueue(int cap=5):_cap(cap)//初始化,设定阻塞队列最大资源数
{}
void push(const T& data)//生产者向阻塞队列生产资源
{}
T pop()//消费资源资源
{}
bool empty()//判断阻塞队列是否为空
{}
bool full()//判断阻塞队列是否为满
{}
~blockqueue()
{}
private:
std::queue<T> _q;//现成的队列,直接拿来用
pthread_mutex_t _mutex;//互斥锁
pthread_cond_t _con_cond;//消费者等待
pthread_cond_t _pro_cond;//生产者等待
int _con_wait_num=0;//等待的消费者数量
int _pro_wait_num=0;//等待的生产者的数量
int _cap;//队列最大资源数
};
初始化
blockqueue(int cap=5):_cap(cap)
{
pthread_mutex_init(&_mutex,nullptr);
pthread_cond_init(&_con_cond,nullptr);
pthread_cond_init(&_pro_cond,nullptr);
}
生产
void push(const T& data)
{
pthread_mutex_lock(&_mutex);//生产和消费都在访问共享资源,加锁
while(full())//队列满, 生产者线程就要去条件变量下等
{
_con_wait_num++;
pthread_cond_wait(&_pro_cond,&_mutex);
_con_wait_num--;
}
_q.push(data);//生产
if(_pro_wait_num>0) pthread_cond_broadcast(&_con_cond);//已经生产资源了,如果但消费者还在等待,叫醒它们
pthread_mutex_unlock(&_mutex);
}
注意: while(full())这里一定是while循环, 而不能单独判断一次, pthread_cond_broadcast会一次叫醒所有的等待的线程, 如果只是判断一次, 被叫醒的线程在队列满的时候竞争到锁就直接去生产资源了
消费
T pop()
{
pthread_mutex_lock(&_mutex);
while(empty())//队列为空, 消费者线程就要去等待
{
_pro_wait_num++;
pthread_cond_wait(&_con_cond,&_mutex);
_pro_wait_num--;
}
T temp=_q.front();
_q.pop();
if(_con_wait_num>0) pthread_cond_broadcast(&_pro_cond);//消费了,代表有空间了,叫醒等待的生产者线程
pthread_mutex_unlock(&_mutex);
return temp;
}
其他操作
bool empty()
{
return _q.size()==0;
}
bool full()
{
return _q.size()==_cap;
}
~blockqueue()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_con_cond);
pthread_cond_destroy(&_pro_cond);
}
使用
可以在多线程环境下使用此阻塞队列，例如多个生产者线程不断push数据到队列中，多个消费者线程不断从队列中pop数据。由于队列操作被锁定，多个线程可以安全地共享这个队列, 当时在构建阻塞队列的时候用了模板, 阻塞队列的资源可以是一个个任务
多个线程不断生产任务到队列中，多个线程不断从队列中拿到任务并执行, 下面是一个示例
简单封装一下pthread库的线程
template<class T>
class Thread
{
public:
Thread()
{}
Thread(const std::function<void(T)>& func,T args,const std::string& name):_thread_name(name),_func(func),_args(args)
{}
bool start()
{
int n=pthread_create(&_tid,nullptr,_thread,(void*)this);
if(n!=0) return false;
return true;
}
void join()
{
pthread_join(_tid,nullptr);
}
std::string get_thread_name()
{
return _thread_name;
}
pthread_t gettid()
{
return _tid;
}
private:
std::string _thread_name;
T _args;
pthread_t _tid;
std::function<void(T)> _func;
static void* _thread(void*args)
{
Thread<T>* p=(Thread<T>*)args;
p->_func(p->_args);
return nullptr;
}
};
这个封装就不做解释了
简单的一个调用阻塞队列的模型
class cp_thread {
public:
cp_thread(int consumer_num = 1, int productor_num = 1)
: _consumer_num(consumer_num), _productor_num(productor_num) {
_con_tid.resize(consumer_num);
_pro_tid.resize(productor_num);
}
// 启动消费者和生产者线程
bool start() {
// 启动消费者线程
for (int i = 0; i < _consumer_num; i++) {
// 创建线程对象并绑定消费者线程函数
_con_tid[i] = Thread<cp_thread*>([=](cp_thread* p) { c_thread(p, i); }, this, "con_thread_" + std::to_string(i));
_con_tid[i].start(); // 启动线程
}
// 启动生产者线程
for (int i = 0; i < _productor_num; i++) {
// 创建线程对象并绑定生产者线程函数
_pro_tid[i] = Thread<cp_thread*>([=](cp_thread* p) { p_thread(p, i); }, this, "pro_thread_" + std::to_string(i));
_pro_tid[i].start(); // 启动线程
}
return true;
}
~cp_thread() {
// 等待所有消费者线程结束
for (auto& i : _con_tid) {
i.join();
}
// 等待所有生产者线程结束
for (auto& i : _pro_tid) {
i.join();
}
}
private:
int _consumer_num; // 消费者线程数量
int _productor_num; // 生产者线程数量
blockqueue<task> _q; // 阻塞队列，存储任务
std::vector<Thread<cp_thread*>> _con_tid; // 消费者线程 ID 向量
std::vector<Thread<cp_thread*>> _pro_tid; // 生产者线程 ID 向量
static int temp; // 静态变量，用于任务编号
// 消费者线程函数，不断向队列中添加任务
static void c_thread(cp_thread* p, int i) {
while (true) {
// 模拟创建任务并推入队列，任务内容是输出任务编号
p->_q.push(task([=]() { std::cout << "task---" << std::to_string(temp) << " running" << std::endl; temp++; }));
// std::cout << p->_con_tid[i].get_thread_name() << ":" << "consumer......" << std::endl;
}
}
// 生产者线程函数，不断从队列中取出任务并执行
static void p_thread(cp_thread* p, int i) {
while (true) {
sleep(3); // 模拟任务处理的时间延迟
auto num = p->_q.pop(); // 从队列中取出任务
num.task_start(); // 执行任务
// std::cout << p->_pro_tid[i].get_thread_name() << ":" << "productor......" << std::endl;
}
}
};
int cp_thread::temp = 1; // 静态变量初始化，任务编号从1开始
2.基于环形队列和信号量
信号量是一种资源预定机制, 只有申请到了信号量, 就一定能拿到资源, 环形队列是一种特殊的队列数据结构，通常用于处理固定大小的缓冲区。与普通的线性队列不同，环形队列在达到队列的边界时不会停止插入操作，而是会从队列的开始位置继续插入, 直到满, 我们可以通过这个实现生产者消费者模型
生产可以看作申请空间资源, 消费可以看作申请数据资源, 使用POSIX信号量实现
实现
完整代码(GitHub)
环形队列
template<class T>
class ring_queue
{
public:
ring_queue(int cap=5): _cap(cap), _v(cap)
{
pthread_mutex_init(&_consumer_mutex, nullptr);
pthread_mutex_init(&_productor_mutex, nullptr);
sem_init(&_room, 0, cap);//空间资源, 初始化为队列大小
sem_init(&_data, 0, 0);//数据资源, 初始化0
}
void push(T& data)
{
sem_wait(&_room);//预定空间资源
//只要预定到了资源, 就一定有一个资源
pthread_mutex_lock(&_consumer_mutex);
_v[_consumer_index++] = data;
_consumer_index %= _cap;
// 模拟数据变化, 假设生产一个整形
data++;
pthread_mutex_unlock(&_consumer_mutex);
sem_post(&_data);//已经生产了一个数据, 数据资源增多
}
T pop()
{
sem_wait(&_data);//预定数据资源
//只要预定到了资源, 就一定有一个资源
pthread_mutex_lock(&_productor_mutex);
const T& temp = _v[_productor_index++];
_productor_index %= _cap;
pthread_mutex_unlock(&_productor_mutex);
sem_post(&_room);//已经消费了一个数据资源, 空间资源增多
return temp;
}
~ring_queue()
{
pthread_mutex_destroy(&_consumer_mutex);
pthread_mutex_destroy(&_productor_mutex);
sem_destroy(&_room);
sem_destroy(&_data);
}
private:
std::vector<T> _v; // 现成的数组, 直接拿来用
pthread_mutex_t _consumer_mutex; // 保护消费者操作的互斥锁
pthread_mutex_t _productor_mutex; // 保护生产者操作的互斥锁
sem_t _room; // 表示环形队列空间的信号量
sem_t _data; // 表示数据资源的信号量
int _consumer_index = 0; // 消费者索引
int _productor_index = 0; // 生产者索引
int _cap; // 队列的容量
};
考虑极端情况, 当环形队列没有数据的时候, 只有生产者能够申请信号量成功, 同理, 数据满的时候, 只有消费者能够申请信号量成功, 所以生产者和消费者的下标不可能对同一位置进行访问, 所以我们不用像阻塞队列那样只要访问共享资源就加锁, 只用给生产者和消费者单独配锁即可, 这样生产者和消费者能够真正并发运行
使用
使用和阻塞队列的使用类似, 引入任务队列即可, 把使用上文阻塞队列的代码拿过来改改就能用
读者写者模型
读者写者模型是一种经典的并发控制问题，主要用来解决在多线程环境下对共享数据的读写操作的同步问题。这个模型关注的是如何在多个读者（读取共享数据的线程）和写者（修改共享数据的线程）之间进行调度，以保证数据的一致性，同时最大限度地提高系统的并发性
关系
读者和写者 : 互斥&&同步, 写者和读者不能同时访问共享资源
写者和写者 : 互斥&&同步, 写者与写者不能同时访问共享资源
读者和读者 : 没有关系
实现思想
读者优先或者写者优先都可能出现线程饥饿问题, 所以采取线程请求的顺序来处理, pthread库也有读写锁
线程池
线程池是一种设计模式，用于在多线程环境下管理和重用线程。它的核心思想是提前创建一组线程（即线程池），然后根据需要将任务分配给这些线程执行，而不是每次需要新任务时都创建一个新的线程
线程池优点
- 提高性能 : 避免频繁创建和销毁线程：创建和销毁线程是开销较大的操作，特别是在高并发的应用程序中。线程池通过重用已经存在的线程，减少了这些开销。
- 控制线程数量 ：线程池可以限制同时运行的线程数量，避免系统资源（如内存、CPU）被过多的线程消耗，从而保持系统稳定性。
- 任务分配：线程池通常会提供一个任务队列，任务可以被动态地加入队列，然后由线程池中的线程处理。线程池负责从队列中取出任务并执行，开发者不需要手动管理每个线程的生命周期。
- 自动化管理 ：线程池可以自动处理线程的创建、调度、任务分配、销毁等细节，使得开发者能够专注于任务逻辑的实现，而不必关注底层的线程管理。
- 提高资源利用率 : 由于线程在执行完任务后不会被立即销毁，而是回到线程池中待命，可以提高线程的利用率，减少资源浪费。
- 负载均衡 ：线程池可以通过合适的调度算法，平衡线程之间的任务负载，确保每个线程都能均匀地分担任务。
实现
完整代码(GitHub)
首先来封装pthread库的线程, 这次参考了c++线程库线程的模板可变参数构造
线程的简单封装
class Thread
{
public:
Thread()
{}
//模板可变参数构造, 使用起来更方便
template<class Fn,class ...Args>
Thread(Fn && func,Args ... args):_func([=](){std::invoke(func,args...);})
{}
bool start()
{
int n=pthread_create(&_tid,nullptr,_thread,(void*)this);
if(n!=0) return false;
return true;
}
void join()
{
pthread_join(_tid,nullptr);
}
pthread_t gettid()
{
return _tid;
}
private:
pthread_t _tid;
std::function<void()> _func;
static void* _thread(void*args)
{
Thread* p=(Thread*)args;
p->_func();
return nullptr;
}
};
简单线程池框架
class Threadpool
{
public:
Threadpool(int thread_num,tasks& tasks) : _thread_num(thread_num), _v(thread_num),_q(tasks)
{}
//启动线程池
void start()
{}
//暂停线程池
void stop()
{}
//让所有线程退出
void quit()
{}
~Threadpool()
{}
private:
int _thread_num;//线程池数量
std::vector<Thread> _v;//使用数组管理线程
tasks& _q;//引用外面的任务队列
pthread_mutex_t _mutex;//访问任务队列互斥锁
pthread_cond_t _cond;//线程等待
int _wait_num=0;//等待线程的数量
bool _isrunning = false;//线程池是否运行
bool _isquit=false;//线程池是否要退出
//线程都执行的方法
void work(Thread*p)
{}
};
初始化
Threadpool(int thread_num,tasks& tasks) : _thread_num(thread_num), _v(thread_num),_q(tasks)
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_cond, nullptr);
for (int i = 0; i < _thread_num; i++)
{
_v[i] = Thread(&Threadpool::work,this,&_v[i]);//初始化每个线程
}
}
启动每个线程
void start()
{
_isrunning=true;
for(auto& i:_v)
{
i.start();
}
}
work
这是所有线程都执行的方法
void work(Thread*p)
{
while (1)
{
pthread_mutex_lock(&_mutex);
//线程池运行, 但任务队列没有任务, 线程就等待
while(_isrunning && _q.empty())
{
_wait_num++;
pthread_cond_wait(&_cond,&_mutex);
_wait_num--;
}
//线程池停止并且任务队列为空或者线程池退出, 线程直接退出
if ((!_isrunning&&_q.empty())||(_isquit))
{
INFO("thread is quit...\n");//这是写的日志工具里的
pthread_mutex_unlock(&_mutex);
break;
}
//到这里就能拿取任务队列的任务了
INFO("thread is running...\n");
task temp = _q.front();
_q.pop();
//有任务, 线程还在睡觉, 直接叫醒
if(!_q.empty()&&_wait_num>0) pthread_cond_signal(&_cond);
pthread_mutex_unlock(&_mutex);
temp();//执行任务
sleep(2);//模拟长时任务
}
}
控制线程池
//设置停止状态
void stop()
{
_isrunning=false;
if(_wait_num>0&&_q.empty())
pthread_cond_broadcast(&_cond);//如果任务为空,线程没必要等待
}
//设置退出状态
void quit()
{
_isquit=true;
pthread_cond_broadcast(&_cond);//线程没必要再等待任务了
}
销毁
~Threadpool()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_cond);
for(auto& i:_v)
{
i.join();//等待所有线程
}
}