多个生产者，单个消费者，只考虑生产者之间的同步互斥
#include "unpipc.h"
#define MAXNITEMS 1000000
#define MAXNTHREADS 100
int nitems; /* read-only by producer and consumer */
struct {
pthread_mutex_t mutex; //互斥信号量
int buff[MAXNITEMS];
int nput;//下一个存放元素的下标
int nval;//下一个存放的值
} shared = { PTHREAD_MUTEX_INITIALIZER };
void *produce(void *), *consume(void *);
int
main(int argc, char **argv)
{
int i, nthreads, count[MAXNTHREADS];
pthread_t tid_produce[MAXNTHREADS], tid_consume;
if (argc != 3)
err_quit("usage: prodcons2 <#items> <#threads>");
nitems = min(atoi(argv[1]), MAXNITEMS);//生产者存放的条目数
nthreads = min(atoi(argv[2]), MAXNTHREADS);//待创建的生产者线程数
Set_concurrency(nthreads);//通知线程系统我们希望并发多少线程
/* 4start all the producer threads */
for (i = 0; i < nthreads; i++) {
count[i] = 0;//计数器初始化为0
Pthread_create(&tid_produce[i], NULL, produce, &count[i]);//创建生产者线程
}
/* 4wait for all the producer threads */
for (i = 0; i < nthreads; i++) {
Pthread_join(tid_produce[i], NULL);//等待生产者线程结束
printf("count[%d] = %d\n", i, count[i]);
}
//这样做的目的是为了避免生产者和消费者之间的同步问题，这里只考虑生产者之间的同步
/* 4start, then wait for the consumer thread */
Pthread_create(&tid_consume, NULL, consume, NULL);//创建消费者线程
Pthread_join(tid_consume, NULL);
exit(0);
}
/* end main */
/* include producer */
void *
produce(void *arg)
{
for ( ; ; ) {
Pthread_mutex_lock(&shared.mutex);
if (shared.nput >= nitems) {//生产完毕
Pthread_mutex_unlock(&shared.mutex);
return(NULL); /* array is full, we're done */
}
shared.buff[shared.nput] = shared.nval;
shared.nput++;
shared.nval++;
Pthread_mutex_unlock(&shared.mutex);
*((int *) arg) += 1;//每个线程私有的计数器，不需要加锁
}
}
void *
consume(void *arg)
{
int i;
for (i = 0; i < nitems; i++) {
if (shared.buff[i] != i)
printf("buff[%d] = %d\n", i, shared.buff[i]);
}
return(NULL);
}
/* end producer */
生产者和消费者之间的同步，消费者采用的方式就是不断轮询。
#include "unpipc.h"
#define MAXNITEMS 1000000
#define MAXNTHREADS 100
int nitems; /* read-only by producer and consumer */
struct {
pthread_mutex_t mutex;
int buff[MAXNITEMS];
int nput;
int nval;
} shared = { PTHREAD_MUTEX_INITIALIZER };
void *produce(void *), *consume(void *);
/* include main */
int
main(int argc, char **argv)
{
int i, nthreads, count[MAXNTHREADS];
pthread_t tid_produce[MAXNTHREADS], tid_consume;
if (argc != 3)
err_quit("usage: prodcons3 <#items> <#threads>");
nitems = min(atoi(argv[1]), MAXNITEMS);
nthreads = min(atoi(argv[2]), MAXNTHREADS);
/* 4create all producers and one consumer */
Set_concurrency(nthreads + 1);
for (i = 0; i < nthreads; i++) {
count[i] = 0;
Pthread_create(&tid_produce[i], NULL, produce, &count[i]);
}
Pthread_create(&tid_consume, NULL, consume, NULL);
/* 4wait for all producers and the consumer */
for (i = 0; i < nthreads; i++) {
Pthread_join(tid_produce[i], NULL);
printf("count[%d] = %d\n", i, count[i]);
}
Pthread_join(tid_consume, NULL);
exit(0);
}
/* end main */
void *
produce(void *arg)
{
for ( ; ; ) {
Pthread_mutex_lock(&shared.mutex);
if (shared.nput >= nitems) {
Pthread_mutex_unlock(&shared.mutex);
return(NULL); /* array is full, we're done */
}
shared.buff[shared.nput] = shared.nval;
shared.nput++;
shared.nval++;
Pthread_mutex_unlock(&shared.mutex);
*((int *) arg) += 1;
}
}
/* include consume */
void
consume_wait(int i)
{
for ( ; ; ) {
Pthread_mutex_lock(&shared.mutex);
if (i < shared.nput) {
Pthread_mutex_unlock(&shared.mutex);
return; /* an item is ready */
}
Pthread_mutex_unlock(&shared.mutex);
}
}
void *
consume(void *arg)
{
int i;
for (i = 0; i < nitems; i++) {
consume_wait(i);
if (shared.buff[i] != i)
printf("buff[%d] = %d\n", i, shared.buff[i]);
}
return(NULL);
}
/* end consume */
每个条件变量都有一个互斥锁与之关联，调用pthread_cond_wait等待某个条件为真时，同时指定其条件变量的地址和所关联的互斥锁的地址
使用条件变量来通知，避免消费者轮询
/* include globals */
#include "unpipc.h"
#define MAXNITEMS 1000000
#define MAXNTHREADS 100
/* globals shared by threads */
int nitems; /* read-only by producer and consumer */
int buff[MAXNITEMS];
struct {//用于生产者之间的同步
pthread_mutex_t mutex;
int nput; /* next index to store */
int nval; /* next value to store */
} put = { PTHREAD_MUTEX_INITIALIZER };
struct {//用于生产者和消费者之间同步
pthread_mutex_t mutex;
pthread_cond_t cond;
int nready; /* number ready for consumer */
} nready = { PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER };
/* end globals */
void *produce(void *), *consume(void *);
/* include main */
int
main(int argc, char **argv)
{
int i, nthreads, count[MAXNTHREADS];
pthread_t tid_produce[MAXNTHREADS], tid_consume;
if (argc != 3)
err_quit("usage: prodcons6 <#items> <#threads>");
nitems = min(atoi(argv[1]), MAXNITEMS);
nthreads = min(atoi(argv[2]), MAXNTHREADS);
Set_concurrency(nthreads + 1);
/* 4create all producers and one consumer */
for (i = 0; i < nthreads; i++) {
count[i] = 0;
Pthread_create(&tid_produce[i], NULL, produce, &count[i]);
}
Pthread_create(&tid_consume, NULL, consume, NULL);
/* wait for all producers and the consumer */
for (i = 0; i < nthreads; i++) {
Pthread_join(tid_produce[i], NULL);
printf("count[%d] = %d\n", i, count[i]);
}
Pthread_join(tid_consume, NULL);
exit(0);
}
/* end main */
/* include prodcons */
void *
produce(void *arg)
{
for ( ; ; ) {
Pthread_mutex_lock(&put.mutex);
if (put.nput >= nitems) {
Pthread_mutex_unlock(&put.mutex);
return(NULL); /* array is full, we're done */
}
buff[put.nput] = put.nval;
put.nput++;
put.nval++;
Pthread_mutex_unlock(&put.mutex);
Pthread_mutex_lock(&nready.mutex);
if (nready.nready == 0)
Pthread_cond_signal(&nready.cond);//发出信号唤醒消费者
nready.nready++;//置为1
Pthread_mutex_unlock(&nready.mutex);
*((int *) arg) += 1;
}
}
void *
consume(void *arg)
{
int i;
for (i = 0; i < nitems; i++) {
Pthread_mutex_lock(&nready.mutex);
while (nready.nready == 0)
Pthread_cond_wait(&nready.cond, &nready.mutex);//wait条件变量
nready.nready--;//置为0
Pthread_mutex_unlock(&nready.mutex);
if (buff[i] != i)
printf("buff[%d] = %d\n", i, buff[i]);
}
return(NULL);
}
/* end prodcons */
本文转自Phinecos(洞庭散人)博客园博客，原文链接：http://www.cnblogs.com/phinecos/archive/2008/05/28/1209125.html，如需转载请自行联系原作者
