Linux：生产者与消费者模式-白红宇

Linux：生产者与消费者模式

阅读量：5960 次

发布时间：2019-06-19

本文共 7434 字，大约阅读时间需要 24 分钟。

生产者：生产数据

消费者：消费数据

提供场所：缓冲区，eg:超市

生产者消费者特点：三种关系，两类人，一个场所

三种关系指的是：生产者与生产者之间是互斥关系

消费者与消费者之间是互斥关系

生产者与消费者之间是同步与互斥关系

两类人：生产者，消费者

一个场所：存储数据（此处用带头单链表实现）

单生产者单消费者模式：此例取数据方式为FIFO先进先出。

利用互斥锁实现单生产者单消费者模式。

#include
     
      #include
      
       #include
       
        typedef int _dataType;typedef int *_dataType_p;typedef struct _node{	_dataType data;	struct _node* next;}node,*nodep,**nodepp;nodep head = NULL;pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;nodep buyNode(_dataType val){	nodep tmp = (nodep)malloc(sizeof(node));	if(tmp!= NULL)	{		tmp->data = val;		tmp->next = NULL;		return tmp;	}		return NULL;}void init(nodepp head){	*head = buyNode(0);}//存在头节点。void push_list(nodep head,_dataType val){	nodep tmp = buyNode(val);	tmp->next = head->next;	head->next = tmp;}int pop_back_list(nodep head,_dataType_p pval){	//当之存在一个哨兵位的时候.	if(NULL == head->next)		return -1;			nodep index = head;	while(NULL != index->next->next)	{	     index = index->next;	}	nodep del = index->next;	*pval = del->data;	index->next = del->next;	free(del);	return 0;}void* product(void* arg){	_dataType i = 0;	while(1)	{		sleep(1);		pthread_mutex_lock(&mutex);		push_list(head,i++);		pthread_mutex_unlock(&mutex);	}	pthread_exit((void *)1);}void* consumer(void *arg){	_dataType val = 0;	while(1){		sleep(2);		pthread_mutex_lock(&mutex);		if(pop_back_list(head,&val) == -1)		{			pthread_mutex_unlock(&mutex);			continue;		}		printf("data:%d\n",val);		pthread_mutex_unlock(&mutex);	}	pthread_exit((void *)1);}int main(){	pthread_t tid1,tid2;	init(&head);	pthread_create(&tid1,NULL,product,NULL);	pthread_create(&tid2,NULL,consumer,NULL);	pthread_join(tid1,NULL);	pthread_join(tid2,NULL);	free(head);	pthread_mutex_destroy(&mutex);	return 0;}

运行结果：

、

然后我们说一说条件变量：

线程间的同步还有这样一种情况:线程A需要等某个条件成立才能继续往下执行,现在这个条

件不成立,线程A就阻塞等待,而线程B在执行过程中使这个条件成立了,就唤醒线程A继续执

行。在pthread库中通过条件变量(Condition Variable)来阻塞等待一个条件,或者唤醒等待这个

条件的线程。Condition Variable用pthread_cond_t类型的变量表示,可以这样初始化和销毁:

返回值:成功返回0,失败返回错误号。

和Mutex的初始化和销毁类似,pthread_cond_init函数初始化.一个Condition Variable,attr参数为

NULL则表.示缺省属性,pthread_cond_destroy函数销毁.一个Condition Variable。如果Condition

Variable是静态分配的,也可以.用宏定义PTHEAD_COND_INITIALIZER初始化,相当于 .用

pthread_cond_init函数初始化并且attr参数为NULL。Condition Variable的操作可以

用下列函数:

返回值:成功返回0,失败返回错误号。

可见,一个Condition Variable总是和一个Mutex搭配使用的。一个线程可以调用

pthread_cond_wait在一个Condition Variable上阻塞等待,这个函数做以下三步操作:

1. 释放Mutex

2. 阻塞等待

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

pthread_cond_timedwait函数还有一个额外的参数可以设定等待超时,如果到达了abstime所指定的时刻仍然没有别的线程来唤醒当前线程,就返回ETIMEDOUT。一个线程可以调用

pthread_cond_signal唤醒在某个Condition Variable上等待的另一个线程,也可以调用

pthread_cond_broadcast唤醒在这个Condition Variable上等待的所有线程。

我们用条件变量来写一个单生产者单消费者模式：

#include
     
      #include
      
       #include
       
        typedef int _dataType;typedef int *_dataType_p;typedef struct _node{	_dataType data;	struct _node* next;}node,*nodep,**nodepp;nodep head = NULL;pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;pthread_cond_t cond = PTHREAD_COND_INITIALIZER;nodep buyNode(_dataType val){	nodep tmp = (nodep)malloc(sizeof(node));	if(tmp!= NULL)	{		tmp->data = val;		tmp->next = NULL;		return tmp;	}		return NULL;}void init(nodepp head){	*head = buyNode(0);}void push_list(nodep head,_dataType val){	nodep tmp = buyNode(val);	tmp->next = head->next;	head->next = tmp;}int pop_list(nodep head,_dataType_p pval){	if(NULL == head->next)		return -1;	nodep del = head->next;	*pval = del->data;	head->next = del->next;	free(del);	return 0;}void* product(void* arg){	_dataType i = 0;	while(1)	{		sleep(1);		pthread_mutex_lock(&mutex);		push_list(head,i++);		pthread_cond_signal(&cond);		pthread_mutex_unlock(&mutex);	}	pthread_exit((void *)1);}void* consumer(void *arg){	_dataType val = 0;	while(1){		sleep(1);		pthread_mutex_lock(&mutex);		if(pop_list(head,&val) == -1)			pthread_cond_wait(&cond,&mutex);		printf("data:%d\n",val);		pthread_mutex_unlock(&mutex);	}	pthread_exit((void *)1);}int main(){	pthread_t tid1,tid2;	init(&head);	pthread_create(&tid1,NULL,product,NULL);	pthread_create(&tid2,NULL,consumer,NULL);	pthread_join(tid1,NULL);	pthread_join(tid2,NULL);	free(head);	pthread_mutex_destroy(&mutex);	return 0;}

运行结果同互斥锁一样。

信号量：

Mutex变量是非0即1的,可看作一种资源的可用数量,初始化时Mutex是1,表示有一个可用资源, 加锁时获得该资源,将Mutex减到0,表示不再有可用资源,解锁时释放该资源,将Mutex重新加到1,表示又有了一个可用资源。

信号量(Semaphore)和Mutex类似,表示可用资源的数量,和Mutex不同的是这个数量可以大于

1。即，如果信号量描述的资源数目是1时，此时的信号量和互斥锁相同！

本节介绍的是POSIX semaphore库函数,详见sem_overview(7),这种信号量不仅可用于同一进程的线程间同步,也可用于不同进程间的同步。

semaphore变量的类型为sem_t,sem_init()初始化一个semaphore变量,value参数表示可用资源的数量,pshared参数为0表示信号量用于同一进程的线程间同步,在用完semaphore变量之后应该调用sem_destroy()释放与semaphore相关的资源。

调用sem_wait()可以获得资源（P操作）,使semaphore的值减1,如果调用sem_wait()时

semaphore的值已经是0,则挂起等待。如果不希望挂起等待,可以调用sem_trywait() 。调用

sem_post() 可以释放资源（V操作）,使semaphore 的值加1,同时唤醒挂起等待的线程。

上面生产者-消费者的例子是基于链表的,其空间可以动态分配,现在基于固定大小的环形队

列重写这个程序（POSIX信号量）:

#include
     
      #include
      
       #include
       
        #define _SEM_PRO_ 20#define _SEM_COM_ 0typedef int _dataType;_dataType blank[_SEM_PRO_];sem_t sem_product;sem_t sem_consumer;void *product(void *arg){	int index = 0;	int count = 0;	while(1)	{		sleep(rand()%5);		sem_wait(&sem_product);		blank[index++] = count ++;		sem_post(&sem_consumer);		index %= _SEM_PRO_;	}	pthread_exit((void *)1);}void* consumer(void * arg){	int index = 0;	while(1)	{		sem_wait(&sem_consumer);		printf("data%d\n",blank[index++]);		sem_post(&sem_product);		index %= _SEM_PRO_;	}	pthread_exit((void*)1);}int main(){	pthread_t tid1,tid2;	sem_init(&sem_product,0,20);	sem_init(&sem_consumer,0,0);	pthread_create(&tid1,NULL,product,NULL);	pthread_create(&tid2,NULL,consumer,NULL);	pthread_join(tid1,NULL);	pthread_join(tid2,NULL);	sem_destroy(&sem_product);	sem_destroy(&sem_consumer);	return 0;}

运行结果同上。

最后我们运用信号量来实现一个多消费者多生产者模式：

#include
     
      #include
      
       #include
       
        #define _SEM_PRO_ 20#define _SEM_COM_ 0typedef int _dataType;_dataType blank[_SEM_PRO_];sem_t sem_product;sem_t sem_consumer;pthread_mutex_t mutex_product = PTHREAD_MUTEX_INITIALIZER;pthread_mutex_t mutex_consumer = PTHREAD_MUTEX_INITIALIZER;void *product(void *arg){	int index = 0;	int count = 0;	while(1)	{		sleep(rand()%5);		sem_wait(&sem_product);		pthread_mutex_lock(&mutex_product);		printf("%d thread id doing\n",(int )arg);		blank[index++] = count ++;		index %= _SEM_PRO_;		pthread_mutex_unlock(&mutex_product);		sem_post(&sem_consumer);	}	pthread_exit((void *)1);}void* consumer(void * arg){	int index = 0;	while(1)	{		sem_wait(&sem_consumer);		pthread_mutex_lock(&mutex_consumer);		printf("%d thread is consumer ,data%d\n",(int)arg,blank[index++]);		index %= _SEM_PRO_;		pthread_mutex_unlock(&mutex_consumer);		sem_post(&sem_product);	}	pthread_exit((void*)1);}int main(){	pthread_t tid1,tid2,tid3,tid4;	sem_init(&sem_product,0,20);	sem_init(&sem_consumer,0,0);	pthread_create(&tid1,NULL,product,NULL);	pthread_create(&tid1,NULL,product,NULL);	pthread_create(&tid3,NULL,consumer,NULL);	pthread_create(&tid4,NULL,consumer,NULL);	pthread_join(tid1,NULL);	pthread_join(tid2,NULL);	pthread_join(tid3,NULL);	pthread_join(tid4,NULL);	sem_destroy(&sem_product);	sem_destroy(&sem_consumer);	pthread_mutex_destroy(&mutex_product);	pthread_mutex_destroy(&mutex_consumer);	return 0;}

运行结果：

总结：

其实本篇是借助生产者消费者模式来讲解线程的互斥锁，条件变量，信号量的操作。

对于生产者/消费者模型我们只需要记住三种关系，两类人，一个场所

三种关系指的是：生产者与生产者之间是互斥关系

消费者与消费者之间是互斥关系

生产者与消费者之间是同步与互斥关系

两类人：生产者，消费者

一个场所：存储数据。

然后对于生产者/消费者我以后会单独写一篇博客进行描述。

一定要熟悉运用本篇中的线程操作函数。

转载于:https://blog.51cto.com/memory73/1769739