producerConsumer

producerConsumer.c

/*************************************************************************
    > File Name: producerConsumer.c
      > Author: Netcan
      > Blog: http://www.netcan666.com
      > Mail: 1469709759@qq.com
      > Created Time: 2016-09-24 六 20:04:04 CST
 ************************************************************************/
#include <stdio.h> // 标准输入输出
#include <stdlib.h> // 标准库
#include <errno.h> // 错误码相关处理
#include <pthread.h> // 多线程函数相关
#include <semaphore.h> // 信号量函数相关
#include <unistd.h> // usleep函数相关
#include <sys/stat.h> // 配置文件读取相关
#include <fcntl.h> // 文件控制相关

// 缓冲区大小
int bufferSize;
// 生产/消费的类型
typedef int item;
// 环形队列
int in = 0, out = 0;
// 缓冲区
item *buffer;
// 互斥、空缓冲区、满缓冲区信号量
sem_t mutex, empty, full;

// 相关变量
int producerNum, consumerNum,
	producerIntervl, consumerIntervl;

// 配置文件内容
const char conf[] = "bufferSize = 10\n"
					"producerNum = 10\n"
					"consumerNum = 10\n"
					"producerIntervl = 100\n"
					"consumerIntervl = 100\n";

// 生产者消费者接口
void* producer(void* id);
void* consumer(void* id);

void initAndRun() {
	// 读取配置文件
	int fd = open("producerConsumer.conf", O_RDWR);
	if(fd == -1) {
		// 读取失败，创建默认配置文件
		if((fd = open("producerConsumer.conf", O_WRONLY|O_CREAT, S_IRUSR|S_IWUSR)) == -1) {
			perror("producerConsumer.conf");
			exit(-1);
		}
		write(fd, conf, sizeof(conf));
		close(fd);
		if((fd = open("producerConsumer.conf", O_RDWR)) == -1) {
			perror("producerConsumer.conf");
			exit(-1);
		}
	}

	// 加载配置文件
	char buf[100];
	buf[read(fd, buf, sizeof(buf))] = 0;
	sscanf(buf, "bufferSize = %d\n"
			"producerNum = %d\n"
			"consumerNum = %d\n"
			"producerIntervl = %d\n"
			"consumerIntervl = %d",
			&bufferSize, &producerNum,
			&consumerNum, &producerIntervl,
			&consumerIntervl);

	printf("bufferSize = %d\n"
			"producerNum = %d\n"
			"consumerNum = %d\n"
			"producerIntervl = %d\n"
			"consumerIntervl = %d\n",
			bufferSize, producerNum,
			consumerNum, producerIntervl,
			consumerIntervl);

	// 生产/消费时间转换为毫秒
	producerIntervl *= 1000;
	consumerIntervl *= 1000;

	// 缓冲区初始化
	buffer = (item*)malloc(sizeof(item) * bufferSize);

	// 初始化信号量
	if(sem_init(&mutex, 0, 1) == -1)
		perror("sem_init");
	if(sem_init(&empty, 0, bufferSize - 1) == -1)
		perror("sem_init");
	if(sem_init(&full, 0, 0) == -1)
		perror("sem_init");

    // 创建生产者，消费者
	pthread_t *p = (pthread_t*)malloc(sizeof(pthread_t) * producerNum);
	pthread_t *c = (pthread_t*)malloc(sizeof(pthread_t) * consumerNum);

	// 为生产者，消费者命名
	char **pName = (char**)malloc(sizeof(char*) * producerNum);
	char **cName = (char**)malloc(sizeof(char*) * consumerNum);

	// 多线程运行
	for (int i = 0; i < producerNum; ++i) {
		pName[i] = (char*)malloc(sizeof(char) * 50);
		sprintf(pName[i], "生产者%d", i);
		if(pthread_create(&p[i], NULL, producer, (void*)pName[i]))
			perror("pthread_create");
	}

	for (int i = 0; i < consumerNum; ++i) {
		cName[i] = (char*)malloc(sizeof(char) * 50);
		sprintf(cName[i], "消费者%d", i);
		if(pthread_create(&c[i], NULL, consumer, (void*)cName[i]))
			perror("pthread_create");
		/* printf("i = %d\n", i); */
		/* printf("id = %s\n", id); */
	}

	// 等待多线程运行结束
	for (int i = 0; i < producerNum; ++i)
		if(pthread_join(p[i], NULL)) perror("pthread_join");

	for (int i = 0; i < consumerNum; ++i)
		if(pthread_join(c[i], NULL)) perror("pthread_join");

	// 释放资源
	close(fd);
	free(buffer);
	free(p);
	free(c);
}

// 显示缓冲区内容
void showBuffer() {
//    printf("in=%d, out=%d\n", in, out);
    printf("[");
    if(out < in)
        for(int i=out; (i % bufferSize)<in; ++i)
            printf("%2d", buffer[i]);
    else if(out > in){
        for(int i=out; i< bufferSize; ++i)
            printf("%2d", buffer[i]);
        for(int i=0; i<in; ++i)
            printf("%2d", buffer[i]);
    }
    printf(" ]\n");
}

void showCurrentStat(const char *id, const item nextc) {
	// 计算缓冲区空满数量
	int ful = (in - out + bufferSize) % bufferSize;
	int emp = bufferSize - ful - 1; // 缓存区有一个元素不存放物品
	/* sem_getvalue(&empty, &emp); */
	/* sem_getvalue(&full, &ful); */
	time_t t = time(NULL);
	struct tm tm = *localtime(&t);
	printf("======%s======\n"
			"now:  %d-%02d-%02d %02d:%02d:%02d\n"
			"缓冲区大小：%d\n"
			"生产/消费的物品：%d\n"
			"空缓冲区数量：%d\n"
			"满缓冲区数量：%d\n",
			(char*)id,
			tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday,
			tm.tm_hour, tm.tm_min, tm.tm_sec,
			bufferSize, nextc, emp, ful);
}

// 生产者
void* producer(void* id) {
    while(1) {
		// 生产者休眠时间
		usleep(producerIntervl);

        item nextp = rand() % 10;
        sem_wait(&empty);
        sem_wait(&mutex);
		// 临界区开始
        buffer[in] = nextp;
        in = (++in)%bufferSize;
		showCurrentStat((char*)id, nextp);
        showBuffer();
		// 临界区结尾
        sem_post(&mutex);
        sem_post(&full);
    }
}
// 消费者
void* consumer(void* id) {
    while(1) {
		// 消费者休眠时间
		usleep(consumerIntervl);

        sem_wait(&full);
        sem_wait(&mutex);
		// 临界区开始
        item nextc = buffer[out];
        out = (++out)%bufferSize;

		showCurrentStat((char*)id, nextc);
        showBuffer();
		// 临界区结尾
        sem_post(&mutex);
        sem_post(&empty);
    }
}

int main() {
	initAndRun();

    return 0;
}