pthread-生产者-消费者问题

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>
#include <time.h>
#include <string.h>
#define BUFFER_SIZE 10
#define END_FLAG (-1)

// 缓冲区定义
typedef struct Product_Buffer
{
int buffer[BUFFER_SIZE];
int pos_head;
int pos_tail;
pthread_mutex_t locker;		// 互斥锁，保证存取互斥
pthread_cond_t not_empty;	// 用于判断队列是否非空
pthread_cond_t not_full;	// 用于判断队列是否非满
} Product_Buffer;

// 产品入缓冲区
void produce(Product_Buffer *pbf, int item)
{
// 加锁
pthread_mutex_lock(&pbf->locker);

while((pbf->pos_tail+1)%BUFFER_SIZE == pbf->pos_head)
{
// 缓冲区已满，线程阻塞，等待不满
pthread_cond_wait(&pbf->not_full, &pbf->locker);
}

// 产品入缓冲区
pbf->buffer[pbf->pos_tail] = item;
pbf->pos_tail = (pbf->pos_tail+1)%BUFFER_SIZE;

// 唤醒因缓冲区空而阻塞的线程
pthread_cond_signal(&pbf->not_empty);

// 解锁
pthread_mutex_unlock(&pbf->locker);
}

// 产品出缓冲区
int consume(Product_Buffer *pbf)
{
// 加锁
pthread_mutex_lock(&pbf->locker);

while(pbf->pos_head == pbf->pos_tail)
{
// 缓冲区空，线程阻塞，等待不空
pthread_cond_wait(&pbf->not_empty, &pbf->locker);
}

// 从缓冲区取数据
int product = pbf->buffer[pbf->pos_head];
pbf->pos_head = (pbf->pos_head+1)%BUFFER_SIZE;

// 唤醒因缓冲区满而阻塞的线程
pthread_cond_signal(&pbf->not_full);

// 解锁
pthread_mutex_unlock(&pbf->locker);

return product;
}

// 生产者线程函数
void * producer_thread(void *pbf)
{
int i;

for(i = 0; i < 20; i++)
{
// 产品入缓冲区
produce(pbf, i);
printf("producer: %d\t", i);
}
produce(pbf, END_FLAG);

return NULL;
}

// 消费者线程函数
void* consumer_thread(void *pbf)
{
int item;

while(1)
{
item = consume(pbf);
if(END_FLAG == item)
break;

printf("consumer: %d\t", item);
}

return NULL;
}

// 初始化缓冲区及互斥量、条件变量
void init_product_buffer(Product_Buffer* pbf)
{
memset(pbf->buffer, 0x00, BUFFER_SIZE*sizeof(int));
pthread_mutex_init(&pbf->locker, NULL);
pthread_cond_init(&pbf->not_empty, NULL);
pthread_cond_init(&pbf->not_full, NULL);
// pbf->not_empty = PTHREAD_COND_INITIALIZER;
// pbf->not_full = PTHREAD_COND_INITIALIZER;
pbf->pos_head = pbf->pos_tail = 0;
}

int main()
{
Product_Buffer pbf;
pthread_t producer;			// 生产者线程
pthread_t consumer;			// 消费者线程

// 初始化缓冲区及互斥量、条件变量
init_product_buffer(&pbf);

// 创建生产者线程
if(pthread_create(&producer, NULL, producer_thread, &pbf) != 0)
{
printf("producer 创建失败\n");
}
else
{
printf("producer 创建成功\n");
}

// 创建消费者线程
if(pthread_create(&consumer, NULL, consumer_thread, &pbf) != 0)
{
<span style="white-space:pre">	</span>printf("Consumer 创建失败\n");
<span style="white-space:pre">	</span>}
<span style="white-space:pre">	</span>else
<span style="white-space:pre">	</span>{
<span style="white-space:pre">	</span>printf("Consumer 创建成功\n");
<span style="white-space:pre">	</span>}

// 等待生产者及消费者线程结束
pthread_join(producer, NULL);
pthread_join(consumer, NULL);

system("pause");
return 1;
}