【Linux】生产者消费者模型介绍

基本概念

说起生产者消费者模型，我们将该模型理解为商店的供货者和前来购买商品的消费者。

他们需要通过商店提供的缓冲区（货架）来进行货物的摆放和购买。

假设没有这个货架，那么生产者只能生产一个商品，直到等到消费者购买后，才可以进行生产，这样的效率是很低的。

我们可以通过一个图片来理解此模型

三种关系

1、生产者和生产者

互斥关系，一个货架上不能同时让两个或者多个生产者来放入数据

2、消费者和消费者

互斥关系，两个或多个消费者不能同时获取货架上的一个商品

3、生产者和消费者

互斥和同步的关系。消费者在消费时，生产者不能放入。同样，生产者在放商品时，消费者不能拿走，这说明了是互斥；而除此之外，生产者必须先生产商品，才可以让消费者消费，这样的按照顺序访问资源的关系称之为同步。

利用单链表进行模型模拟

原理

用两个线程来分别表示生产者和消费者，用单链表来表示中间的缓冲区

当生产者生产数据时，将新的数据放入单链表的头部。

当消费者消费数据时，将头部的数据弹出。

所用函数

pthread_mutex_t

初始化函数

加锁函数

lock为加锁，unlock为解锁

如果一个线程想要获取锁，又不想挂起等待，则调用trylock

条件变量

pthread_cond_t

条件变量的初始化

线程等待

操作方法 

1、释放Mutex
 

2、阻塞等待
 

3、当被唤醒时,重新获得Mutex并返回。 

唤醒等待的消费者线程

代码实现

#include<stdio.h>
#include<pthread.h>
#include<stdlib.h>
#include<unistd.h>
#include<assert.h>

pthread_mutex_t mylock = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t mycond = PTHREAD_COND_INITIALIZER;

//定义节点的结构体
typedef struct node
{
int _data;
struct node* _next;
}node,*pNode;

//定义链表的结构体
typedef struct LinkList
{
node* phead;
}LinkList,*pLinkList;

//初始化头节点
void InitList(pLinkList plist)
{
assert(plist);
plist->phead = NULL;
}

//初始化并返回新的节点
pNode buyNewNode(int data)
{
pNode newnode = (pNode)malloc(sizeof(node));
if(newnode == NULL)
{
perror("malloc");
exit(1);
}
newnode->_data = data;
newnode->_next = NULL;
return newnode;
}

//头节点前进行插入
void pushFront(pLinkList pList,int data)
{
//产生新的节点
pNode newNode = buyNewNode(data);

//判断有无元素
if(pList->phead == NULL)
{
pList->phead = newNode;
return;
}

//进行头插
pNode pFristNode = pList->phead;
newNode->_next = pFristNode;
pList->phead = newNode;
}

//弹出一个头节点,data用来保存弹出的元素
void popFront(pLinkList pList,int* data)
{
assert(pList);

pNode delNode = pList->phead;

if(delNode == NULL)
return;

*data = delNode->_data;
pList->phead = delNode->_next;
delNode->_next = NULL;
free(delNode);
delNode = NULL;
}

//销毁单链表
void DestroyList(pLinkList pList)
{
assert(pList);
pNode cur = pList->phead;
pNode delNode = NULL;
while(cur)
{
delNode = cur;
cur = cur->_next;
delNode->_next = NULL;
free(delNode);
}
pList->phead = NULL;
}

//打印单链表中的内容
void ShowList(pLinkList pList)
{
assert(pList);
pNode cur = pList->phead;
while(cur)
{
printf("%d ",cur->_data);
cur = cur->_next;
}
printf("\n");
}

void* producterThread(void* arg)
{
pLinkList list = (pLinkList)arg;
int data = 0;
while(1)
{
sleep(1);
pthread_mutex_lock(&mylock);//访问临界区前进行加锁
data = rand()%100;
pushFront(list,data);//生产者进行生产
pthread_cond_signal(&mycond);//生产完毕后唤醒在该条件
pthread_mutex_unlock(&mylock);//访问完毕后解锁
printf("生产者生产了: %d \n",data);
}
}

void* consumerThread(void* arg)
{
pLinkList list = (pLinkList)arg;
int data = 0;
while(1)
{
sleep(1);

pthread_mutex_lock(&mylock);//访问临界区前进行加锁

while(list->phead == NULL)//如果缓存区中没有数据，则进行等待
{
pthread_cond_wait(&mycond,&mylock);
}

popFront(list,&data);//消费者进行消费
pthread_mutex_unlock(&mylock);//访问完毕，进行解锁
printf("消费者进行消费 : %d\n",data);
}
}

int main()
{
LinkList list;
InitList(&list);

//创建消费者和生产者两个线程
pthread_t tid1,tid2;
pthread_create(&tid1,NULL,producterThread,(void*)&list);
pthread_create(&tid2,NULL,consumerThread,(void*)&list);

//等待线程的结束回收进程
pthread_join(tid1,NULL);
pthread_join(tid2,NULL);

DestroyList(&list);
return 0;
}

运行结果

基于环形队列的多生产者多消费者模型

原理

1、生产者先进行生产

2、消费者消费的数据不能够超过生产者

3、生产者生产数据不能比消费者快一圈

图解

代码实现

#include<stdio.h>
#include<semaphore.h>//信号量包含的头文件
#include<pthread.h>
#include<unistd.h>
#include<stdlib.h>

#define _SIZE_ 64

int ringBuf[_SIZE_];//定义环形队列，64为环形队列的长度

//定义两个信号量
sem_t semBlack;//格子
sem_t semData;//数据
sem_t proLock;//生产者之间的互斥
sem_t conLock;//消费者之间的互斥

//消费者
void* consumer(void* arg)
{
//成功返回0,失败返回错误码
pthread_detach(pthread_self());

static int i = 0;
int id = (int)arg;
while(1)
{
//sleep(1);
usleep(1000);
sem_wait(&semData);//对于临界区数据进行P操作
sem_wait(&conLock);//对于消费者互斥锁进行P操作

printf("消费者 %d 消费了: %d  , tid : %lu\n",id,ringBuf[i++],pthread_self());

i %= _SIZE_;

sem_post(&conLock);//对于消费者互斥锁进行V操作
sem_post(&semBlack);//对于临界区数据进行V操作
}
}

void* producter(void* arg)
{
pthread_detach(pthread_self());
static int i = 0;
int id = (int)arg;

while(1)
{
sleep(1);
//usleep(1000);

sem_wait(&semBlack);//对临界区数据进行P操作
sem_wait(&proLock);//对生产者互斥锁进行P操作

int num = rand()%100;
ringBuf[i++] = num;

printf("生产者 %d 生产了 : %d , tid : %lu\n",id, num ,pthread_self());
i %= _SIZE_;

sem_post(&proLock);//对生产者互斥锁进行V操作
sem_post(&semData);//对临界区数据进行V操作
}
}

int main()
{
//分别定义两个生产者和两个消费者
pthread_t con0,con1,pro0,pro1;

//初始化信号量
sem_init(&semData,0,0);
sem_init(&semBlack,0,_SIZE_);
sem_init(&proLock,0,1);
sem_init(&conLock,0,1);

int i = 0;

pthread_create(&pro0,NULL,producter,(void*)i);
pthread_create(&con0,NULL,consumer,(void*)i);

i = 1;

pthread_create(&pro1,NULL,producter,(void*)i);
pthread_create(&con1,NULL,consumer,(void*)i);

//销毁信号量
sem_destroy(&semBlack);
sem_destroy(&semData);

pthread_join(pro0,NULL);
pthread_join(con0,NULL);

return 0;
}

运行结果

如果将代码中注释的时间做修改，使得生产者的生产速度高于消费者，那么运行结果为下图