多个生产者与消费者
生产者消费者问题属于有界缓冲区问题。我们现在讲述多个生产者向一个缓冲区中存入数据,多个生产者从缓冲区中取数据。共享缓冲区作为一个环绕缓冲区,存数据到头时再从头开始。我们使用一个互斥量保护生产者向缓冲区中存入数据。由于有多个生产者,因此需要记住现在向缓冲区中存入的位置。使用一个互斥量保护缓冲区中消息的数目,这个生产的数据数目作为生产者和消费者沟通的桥梁。使用一个条件变量用于唤醒消费者。由于有多个消费者,同样消费者也需要记住每次取的位置。
#include "unpipc.h" #include "my_err.h" #include <pthread.h> #define MAXITEMS 6 #define MAXTHREADS 3 int buffer[MAXITEMS];
//生产者使用的结构体 struct shared { pthread_mutex_t mutex; int input;//下标 int i;//存入的值 }shared={PTHREAD_MUTEX_INITIALIZER};
//消费者使用的结构体,生产者也会使用 struct ready { pthread_mutex_t mutex; pthread_cond_t cond; int nready;//生产者为消费者准备好的可用的商品数量 int nget;//消费者从仓库中抓取商品的次数 }ready={PTHREAD_MUTEX_INITIALIZER,PTHREAD_COND_INITIALIZER}; void *produce(void *arg) { while(1) { pthread_mutex_lock(&shared.mutex); if(shared.i>=MAXITEMS)//仓库满,不能再生产 { pthread_mutex_unlock(&shared.mutex); return NULL; } buffer[shared.input]=shared.i; if(++shared.input>=MAXITEMS)//1...MAXITEMS,循环缓冲,如果抓取的次数大于MAXITEMS,从头在开始计数 shared.input=0; //++shared.input; ++shared.i;//值 pthread_mutex_unlock(&shared.mutex); //使临界区尽量短使用多个互斥量,生产完毕可以消费了 pthread_mutex_lock(&ready.mutex); if(ready.nready==0)//如果有消费者等待消费,则唤醒 pthread_cond_signal(&ready.cond); ++ready.nready;//可以消费的商品的数目+1 pthread_mutex_unlock(&ready.mutex); } return NULL; } void *consume(void *arg) { while(1) { pthread_mutex_lock(&ready.mutex); if(ready.nready==0)//如果可以消费的数目==0,仓库中没有可消费的商品,等待 pthread_cond_wait(&ready.cond,&ready.mutex); if(++ready.nget>=MAXITEMS)//1...MAXITEMS { if(ready.nget==MAXITEMS)//抓取商品的次数==仓库的最大容量时少操作一次, printf("buffer[%d] = %d \n",ready.nget-1,buffer[(ready.nget-1)%MAXITEMS]); pthread_cond_signal(&ready.cond);// pthread_mutex_unlock(&ready.mutex); return NULL; } --ready.nready; pthread_mutex_unlock(&ready.mutex); printf("buffer[%d] = %d\n",ready.nget-1,buffer[(ready.nget-1)%MAXITEMS]); } return NULL; } int main() { pthread_setconcurrency(MAXTHREADS*2);// int i; pthread_t pid_produce[MAXTHREADS],pid_consume[MAXTHREADS]; for(i=0;i<MAXTHREADS;++i) pthread_create(&pid_produce[i],NULL,produce,NULL); for(i=0;i<MAXTHREADS;++i) pthread_create(&pid_consume[i],NULL,consume,NULL); for(i=0;i<MAXTHREADS;++i) pthread_join(pid_produce[i],NULL); for(i=0;i<MAXTHREADS;++i) pthread_join(pid_consume[i],NULL); exit(0); }