生产者消费者问题（Producer-consumer problem）

概述

生产者消费者问题，也称有限缓冲问题（英语：Bounded-buffer problem），是一个多线程同步问题的经典案例。该问题描述了两个共享固定大小缓冲区的线程——即所谓的“生产者”和“消费者”——在实际运行时会发生的问题。其中生产者的主要作用是生成一定量的数据放到缓冲区中，然后重复此过程。与此同时，消费者也在缓冲区消耗这些数据。该问题的关键就是要保证生产者不会在缓冲区满时加入数据，消费者也不会在缓冲区中空时消耗数据。

问题分析

1.要保证不让生产者在缓存还是满的时候仍然要向内写数据;
2.不让消费者试图从空的缓存中取出数据。
生产者和消费者对缓冲区互斥访问是互斥关系，同时生产者和消费者又是一个相互协作的关系，只有生
产者生产之后，消费者才能消费，他们也是同步关系。

只有生产者和消费者两个进程，正好是这两个进程存在着互斥关系和同步关系。那么需要解决的是互斥
和同步PV操作的位置。我们使用了三个信号量：full 和 empty用来解决唤醒的问题 ，而信号量mutex作为互斥信号量，它用于控制互斥访问缓冲池，互斥信号量初值为 1；信号量 full 用于记录当前缓冲池中“满”缓冲区数，初值为0。信号量 empty 用于记录当前缓冲池中“空”缓冲区数，初值为n。新的数据添加到缓存中后，full 在增加，而 empty 则减少。如果生产者试图在 empty 为0时减少其值，生产者就会被阻塞。下一轮中有数据被消费掉时，
empty就会增加，生产者就会被“唤醒”。

mypro.c

/*************************************************************************
	> File Name: mypro.c
	> Author: Xianghao Jia
	> mail: xianghaojia@sina.com
	> Created Time: Mon 18 Nov 2019 07:03:41 PM PST
 ************************************************************************/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/types.h>
#include "queue.h"

#define CONSUMER_COUNT

sem_t vacancy;//空位
sem_t goods;//物品数量
sem_t sem_mutex;//互斥信号量

int main(int argc, char ** argv)
{
	pthread_t consumer_id[3];
	pthread_t productor_id;
	// 创建队列
	QUEUE * queue = queue_init();
	if(queue == NULL)
	{
		return -1;
	}
	// 初始化信号量
	sem_init(&vacancy, 0, 10);
	sem_init(&good, 0, 0);
	sem_init(&sem_mutex ,0 , 1);

	int i = 0;
	// 创建3个消费者
	for(i = 0; i < CONSUMER_COUNT; i++)
	{
		if(0 == pthread_create(&consumer_id[i], NULL, consumer_routine, queue_head))
		{
			perror("pthread_create");
			exit(1);
		}
	}
	// 创建生产者
	if(0 == pthread_create(&consumer_id[i], NULL, productor_routine, queue_head))
	{
		perror("pthread_create");
		exit(1);
	}
	// 用来停止程序
	getchar();

	// 取消线程
	for(i = 0; i < CONSUMER_COUNT; i++)
	{
		pthread_cancel(consumer_id[i]);
	}
	// 接合线程
	pthread_join(productor_id, NULL);
	for(i = 0; i < CONSUMER_COUNT; i++)
	{
		pthread_join(consumer_id[i], NULL);
	}
	
	// 销毁信号量
	sem_destroy(&vacancy);
	sem_destroy(&goods);
	sem_destroy(&sem_mutex);
	// 销毁队列
	queue_destroy(queue);

	return 0;
}


void *consumer_routine(void *arg)
{	
	printf("thread consumer start...\n");
	QUEUE * queue = (QUEUE *)arg;

	while(1)
	{
		sem_wait(&goods);
		sem_wait(&sem_mutex);
		
		if(exit_flag == 1)
		{
			printf("thread consumer is exiting\n");
			break;
		}
		// 获得队列首
		printf("consume %d\n",queue_front(queue));
		// 弹出队首
		queue_out(queue);
	
		sem_post(&sem_mutex);
		sem_post(&vacancy);
		usleep(random() % 10000);
	}
	pthread_exit(NULL);
}

void *productor_routine(void *arg)
{
	printf("thread productor start...\n");
	QUEUE * queue = (QUEUE *)arg;

	while(1)
	{
		sem_wait(&vacancy);
		sem_wait(&sem_mutex);
		
		if(exit_flag == 1)
		{
			printf("thread productor is exiting\n");
			break;
		}
		// 创建节点到队列中
		int n = random() % 100;
		queue_in(queue, n);
		printf("product %d\n", n);

		sem_post(&sem_mutex);
		sem_post(&goods);
		usleep(random() % 10000);
	}
	pthread_exit(NULL);	
}

queue.h

#ifndef _QUEUE_H
#define _QUEUE_H
#define QUEUE_DEEP 10

extern int myerror;

typedef struct Node
{
	//	BINODE *pnode;
	int num;
	int count;
	struct Node *next;
}QNODE, QUEUE;

int queue_is_empty(QNODE *queue);
void queue_destroy(QNODE *queue);
int queue_out(QNODE *queue);
int queue_in(QNODE *queue, int n);
int queue_front(QNODE *queue);
int queue_is_full(QNODE *queue);
QNODE *queue_init();
#endif

queue.c

#include "queue.h"
#include <stdio.h>

int myerror = 0;

QNODE *queue_init()
{
	QNODE *head = malloc(sizeof(QNODE));
	if (head == NULL)
	{
		return NULL;
	}
	memset(head, sizeof(QNODE));
	return head;
}

int queue_is_empty(QNODE *queue)
{
	return queue->next == NULL;
}
void queue_destroy(QNODE *queue)
{
	while (queue != NULL)
	{
		QNODE *temp = queue->next;
		free(queue);
		queue = temp;
	}
}
/*
	取出队列首段的元素
*/
int queue_out(QNODE *queue)
{
	if (queue->next == NULL)
	{
		return -1;
	}
	QNODE * temp = queue->next;
	queue->next = temp->next;
	free(temp);
	return 0;
}
/*
	将元素添加到队尾
*/
int queue_in(QNODE *queue, int n)
{
	while (queue->next)
	{
		queue = queue->next;
	}
	QNODE * temp = malloc(sizeof(QNODE));
	if (temp == NULL)
	{
		return -1;
	}
	temp->next = NULL;
	temp->num = n;
	queue->next = temp;
	return 0;
}
int queue_front(QNODE *queue)
{
	if (queue->next == NULL)
	{
		myerror = 1;
		return -1;
	}
	return queue->next->num;
}