文章目录
- Linux线程
- 6. 生产消费者模型
- 6.1 基于阻塞队列的生产消费者模型
- 6.1.1 阻塞队列模型实现
- 6.2 基于环形队列的生产消费者模型
- 6.2.1 POSIX信号量的概念
- 6.2.2 POSIX信号量的使用
- 6.2.3 环形队列模型实现
Linux线程
6. 生产消费者模型
生产消费者模型的概念
生产者消费者模型 是一种常见的并发编程模型。
在这个模型中，通常存在两类角色：生产者和消费者。
生产者负责生成数据或产品，并将其放入一个共享的缓冲区中。而消费者则从缓冲区中取出数据或产品进行消费处理。
为何要使用生产者消费者模型
生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。
简单的说：就是为了解决生产者和消费者之间的速度不匹配问题。
生产者消费者模型优点
解耦、支持并发、支持忙闲不均。
上图中有"321"原则：
3种关系：生产者和生产者(互斥)、消费者和消费者(互斥)、生产者和消费者(互斥、同步)。
2种角色：生产者和消费者。
1个交易场所：特定结构的内存空间。
6.1 基于阻塞队列的生产消费者模型
基于阻塞队列的生产消费者模型的概念
基于阻塞队列的生产消费者模型 是一种用于协调生产者和消费者之间工作流程的编程模式。
实现的原理
在这个模型中，生产者负责生成数据或产品，并将其放入一个阻塞队列中。阻塞队列是一种特殊的数据结构，当队列已满时，生产者尝试添加新元素的操作会被阻塞，直到队列有足够的空间。
消费者则从这个阻塞队列中获取数据或产品进行处理。当队列为空时，消费者尝试获取元素的操作会被阻塞，直到生产者向队列中添加了新的元素。
6.1.1 阻塞队列模型实现
这是我们的任务对象，一个简单的模拟加减乘除计算：
#pragma once
#include <iostream>
#include <string>
std::string opers="+-*/%";
enum{
DivZero=1,
ModZero,
Unknown
};
class Task
{
public:
Task()
{}
Task(int x,int y,char op)
:_data1(x),_data2(y),_oper(op),_result(0),_exitcode(0)
{}
void run()
{
switch (_oper)
{
case '+':
_result=_data1+_data2;
break;
case '-':
_result=_data1-_data2;
break;
case '*':
_result=_data1*_data2;
break;
case '/':
{
if(_data2==0) _exitcode=DivZero;
else _result=_data1/_data2;
}
break;
case '%':
{
if(_data2==0) _exitcode=ModZero;
else _result=_data1%_data2;
}
break;
default:
_exitcode=Unknown;
break;
}
}
//Task对象重载运算符(),()直接进行run函数
void operator()()
{
run();
}
std::string GetResult()
{
std::string r=std::to_string(_data1);
r+=_oper;
r+=std::to_string(_data2);
r+="=";
r+=std::to_string(_result);
r+="[code: ";
r+=std::to_string(_exitcode);
r+="]";
return r;
}
std::string GetTask()
{
std::string r=std::to_string(_data1);
r+=_oper;
r+=std::to_string(_data2);
r+="=?";
return r;
}
~Task()
{}
private:
int _data1;
int _data2;
char _oper;
int _result;
int _exitcode;
};
这是我们实现的阻塞队列模型：
#pragma once
#include <iostream>
#include <queue>
#include <unistd.h>
template<class T>
class blockQueue
{
static const int defaultnum=5;
public:
//构造函数
blockQueue(int maxp=defaultnum)
:_maxp(maxp)
{
pthread_mutex_init(&_mutex,nullptr);
pthread_cond_init(&_p_cond,nullptr);
pthread_cond_init(&_c_cond,nullptr);
}
T pop()
{
pthread_mutex_lock(&_mutex);
if(_q.size()==0)
{
pthread_cond_wait(&_c_cond,&_mutex);
}
T out=_q.front();
_q.pop();
pthread_cond_signal(&_p_cond);
pthread_mutex_unlock(&_mutex);
return out;
}
void push(const T &in)
{
pthread_mutex_lock(&_mutex);
if(_q.size()==_maxp)
{
pthread_cond_wait(&_p_cond,&_mutex);
}
_q.push(in);
pthread_cond_signal(&_c_cond);
pthread_mutex_unlock(&_mutex);
}
//析构函数
~blockQueue()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_p_cond);
pthread_cond_destroy(&_c_cond);
}
private:
std::queue<T> _q; //生产消费队列
int _maxp; //最大值
pthread_mutex_t _mutex; //锁
pthread_cond_t _p_cond; //生产同步队列
pthread_cond_t _c_cond; //消费同步队列
};
运行函数：
#include "blockQueue.hpp"
#include "Task.hpp"
#include <unistd.h>
#include <ctime>
//消费者模型
void *Consumer(void *args)
{
blockQueue<Task> *bp=static_cast<blockQueue<Task>*>(args);
while(true)
{
Task t=bp->pop();
t();
std::cout<<"thread id: "<<pthread_self()<<" 处理了一个任务："<<t.GetTask()\
<<" 处理后的任务结果："<<t.GetResult()<<std::endl;
sleep(3);
}
}
//生产者模型
void *Productor(void *args)
{
blockQueue<Task> *bp=static_cast<blockQueue<Task>*>(args);
int len=opers.size();
int data;
while(true)
{
int data1=rand()%10+1;
usleep(10);
int data2=rand()%10;
char op=opers[rand()%len];
Task t(data1,data2,op);
bp->push(t);
sleep(1);
std::cout<<"thread id: "<<pthread_self()<<" 生产了一个任务："<<t.GetTask()<<std::endl;
sleep(2);
}
}
int main()
{
srand(time(nullptr));
blockQueue<Task> *bp=new blockQueue<Task>();
pthread_t c[3],p[5];
for(int i=0;i<5;i++)
{
pthread_create(p+i,nullptr,Productor,bp);
}
for(int i=0;i<3;i++)
{
pthread_create(c+i,nullptr,Consumer,bp);
}
for(int i=0;i<5;i++)
{
pthread_join(p[i],nullptr);
}
for(int i=0;i<3;i++)
{
pthread_join(c[i],nullptr);
}
delete bp;
return 0;
}
基于阻塞队列的生产消费者的模型实现了(就是打出来乱乱的)：
6.2 基于环形队列的生产消费者模型
基于环形队列的生产消费者模型的概念
基于环形队列的生产消费者模型是一种常见的并发编程模型，其中使用环形队列作为生产者和消费者之间共享的数据结构。
实现的原理
环形队列是一种特殊的数据结构，它的特点是队列的尾部与头部相连，形成一个环状。在基于环形队列的生产消费者模型中，生产者负责向队列中添加数据，消费者负责从队列中取出数据。
为了保证生产者和消费者之间的正确协作，需要遵循以下三个原则：
生产者不能超过消费者：也就是说，生产者生产数据的速度不能超过消费者处理数据的速度，否则队列会溢出。
生产者不能领先消费者一圈：也就是说，生产者和消费者之间不能有太大的差距，否则会导致数据丢失或重复。
在同一位置时，生产者和消费者必须互斥：也就是说，当生产者和消费者在同一位置时，只能有一个进行操作，否则会导致数据不一致。
6.2.1 POSIX信号量的概念
POSIX信号量和SystemV信号量作用相同，都是用于同步操作，达到无冲突的访问共享资源目的。 但POSIX可以用于线程间同步。
实现的原理
信号量本质上是一个整数，其值表示资源的可用数量。 通过 sem_wait 操作（也称为 P 操作）来请求资源，如果信号量的值大于 0 ，则减 1 并继续执行；如果值为 0 ，则阻塞当前线程或进程，直到信号量的值大于 0 。通过 sem_post 操作（也称为 V 操作）来释放资源，使信号量的值增加 1 ，并唤醒一个等待该信号量的线程或进程。
6.2.2 POSIX信号量的使用
初始化信号量
#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);
参数：
pshared:0表示线程间共享，非零表示进程间共享
value：信号量初始值
销毁信号量
int sem_destroy(sem_t *sem);
等待信号量
int sem_wait(sem_t *sem); //P()
功能：等待信号量，会将信号量的值减1
发布信号量
int sem_post(sem_t *sem);//V()
功能：发布信号量，表示资源使用完毕，可以归还资源了。将信号量值加1。
6.2.3 环形队列模型实现
任务继续使用上面的简单模拟计算。
这是我们实现的环形队列模型：
#pragma once
//多生产多消费
#include <iostream>
#include <vector>
#include <ctime>
#include <semaphore.h>
#include <pthread.h>
const static int defaultcap=5;
template<class T>
class RingQueue
{
private:
void P(sem_t &sem) //申请信号量
{
sem_wait(&sem);
}
void V(sem_t &sem) //发布信号量
{
sem_post(&sem);
}
void Lock(pthread_mutex_t &mutex) //加锁
{
pthread_mutex_lock(&mutex);
}
void Unlock(pthread_mutex_t &mutex) //解锁
{
pthread_mutex_unlock(&mutex);
}
public:
RingQueue(int capacity=defaultcap)
:_ringqueue(capacity),_capacity(capacity),_c_step(0),_p_step(0)
{
sem_init(&_c_sem,0,0);
sem_init(&_p_sem,0,capacity);
pthread_mutex_init(&_c_mutex,nullptr);
pthread_mutex_init(&_p_mutex,nullptr);
}
void Push(const T &in) //生产
{
P(_p_sem);
Lock(_p_mutex); //先申请信号量，因为信号量是原子的
_ringqueue[_p_step]=in;
//向后移动，维持环形特性
_p_step++;
_p_step%=_capacity;
Unlock(_p_mutex);
V(_c_sem);
}
void Pop(T *out) //消费
{
P(_c_sem);
Lock(_c_mutex);
*out=_ringqueue[_c_step];
//向后移动，维持环形特性
_c_step++;
_c_step%=_capacity;
Unlock(_c_mutex);
V(_p_sem);
}
~RingQueue()
{
sem_destroy(&_c_sem);
sem_destroy(&_p_sem);
pthread_mutex_destroy(&_c_mutex);
pthread_mutex_destroy(&_p_mutex);
}
private:
std::vector<T> _ringqueue; //模拟环形队列
int _capacity; //容量大小
int _c_step; //消费者位置
int _p_step; //生产者位置
sem_t _c_sem; //消费者信号量，关注的数据资源
sem_t _p_sem; //生产者信号量，关注的空间资源
pthread_mutex_t _c_mutex;
pthread_mutex_t _p_mutex;
};
运行的函数：
#include <iostream>
#include <pthread.h>
#include <unistd.h>
#include "RingQueue.hpp"
#include "Task.hpp"
using std::cout;
using std::endl;
struct ThreadData
{
RingQueue<Task> *rq;
std::string threadname;
};
void *Productor(void *args)
{
//sleep(3);
ThreadData *td = static_cast<ThreadData*>(args);
RingQueue<Task> *rq = td->rq;
std::string name = td->threadname;
int len = opers.size();
while(true)
{
//获取数据
int data1 = rand() % 10 + 1;
usleep(10);
int data2 = rand() % 10;
char op = opers[rand() % len];
Task t(data1, data2, op);
//生产数据
rq->Push(t);
cout << "Productor task done, task is : " << t.GetTask() << " who: " << name << endl;
sleep(3);
}
return nullptr;
}
void *Consumer(void *args)
{
ThreadData *td = static_cast<ThreadData*>(args);
RingQueue<Task> *rq = td->rq;
std::string name = td->threadname;
while(true)
{
//消费数据
Task t;
rq->Pop(&t);
//处理数据
t();
cout << "Consumer get task, task is : " << t.GetTask() << " who: " << name << " result: " << t.GetResult() << endl;
//sleep(1);
}
return nullptr;
}
int main()
{
srand(time(nullptr));
RingQueue<Task> *rq=new RingQueue<Task>();
pthread_t c[2],p[3];
for(int i=0;i<3;i++)
{
ThreadData *td = new ThreadData();
td->rq = rq;
td->threadname = "Productor-" + std::to_string(i);
pthread_create(p+i,nullptr,Productor,td);
}
for(int i=0;i<2;i++)
{
ThreadData *td = new ThreadData();
td->rq = rq;
td->threadname = "Consumer-" + std::to_string(i);
pthread_create(c+i,nullptr,Consumer,td);
}
for(int i=0;i<3;i++)
{
pthread_join(p[i],nullptr);
}
for(int i=0;i<2;i++)
{
pthread_join(c[i],nullptr);
}
return 0;
}
基于环形队列的生产消费者的模型实现了：
