生产者：生产数据
消费者：消费数据
提供场所：缓冲区，eg:超市
生产者消费者特点：三种关系，两类人，一个场所
三种关系指的是：生产者与生产者之间是互斥关系
消费者与消费者之间是互斥关系
生产者与消费者之间是同步与互斥关系
两类人：生产者，消费者
一个场所：存储数据（此处用带头单链表实现）
单生产者单消费者模式：此例取数据方式为FIFO先进先出。
利用互斥锁实现单生产者单消费者模式。
#include<stdio.h>
#include<malloc.h>
#include<pthread.h>
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<stdio.h>
#include<malloc.h>
#include<pthread.h>
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<stdio.h>
#include<pthread.h>
#include<semaphore.h>
#define _SEM_PRO_ 20
#define _SEM_COM_ 0
typedef 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<stdio.h>
#include<pthread.h>
#include<semaphore.h>
#define _SEM_PRO_ 20
#define _SEM_COM_ 0
typedef 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;
}运行结果：
总结：
其实本篇是借助生产者消费者模式来讲解线程的互斥锁，条件变量，信号量的操作。
对于生产者/消费者模型我们只需要记住三种关系，两类人，一个场所
三种关系指的是：生产者与生产者之间是互斥关系
消费者与消费者之间是互斥关系
生产者与消费者之间是同步与互斥关系
两类人：生产者，消费者
一个场所：存储数据。
然后对于生产者/消费者我以后会单独写一篇博客进行描述。
一定要熟悉运用本篇中的线程操作函数。
本文出自 “剩蛋君” 博客，请务必保留此出处http://memory73.blog.51cto.com/10530560/1769739
原文地址：http://memory73.blog.51cto.com/10530560/1769739
