Linux多线程实现生产者消费者进程（Linux+window代码）

原文链接：我的个人链接

Linux 常用多线程函数

pthread_create（）：创建一个线程
pthread_exit（）：退出一个线程
pthread_jion（）：阻塞当前线程，直到另一个线程执行结束
pthread_mutex_lock（）：互斥加锁
pthread_mutex_unlock（）：互斥锁解锁
pthread_cond_init（）：初始化条件变量
pthread_cond_signal（）：发送信号唤醒进程
pthread_cond_wait（）：等待条件变量的特殊事件发生

问题简介

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

条件变量+互斥锁

条件变量

条件变量是一种“事件通知机制”，它本身不提供、也不能够实现“互斥”的功能。因此，条件变量通常（也必须）配合互斥量来一起使用，其中互斥量实现对“共享数据”的互斥（即同步），而条件变量则去执行 “通知共享数据状态信息的变化”的任务。比如通知队列为空、非空，或任何其他需要由线程处理的共享数据的状态变化。实际开发中，生产者-消费者模型是经常被使用到的一个技巧，若干线程不断的往某个队列中生产数据进去，而其他若干线程不断的去队列中消费数据。这里“队列”是共享的数据，需要用互斥量来对其进行加锁，防止数据紊乱；细心的读者发现，去队列消费数据的线程怎么才能知道队列中何时有数据进来，总不能一直在那里傻傻的等待吧！如果一直等待，则需要互斥量加锁，那么生产者线程会加锁失败，会一直尝试去加锁。这无形中会导致性能的急剧下降。因此这时候是“条件变量”闪亮登场，展示它真正“技术”的时候了，它会通知对应的（或广播所有等待状态变化的）线程，告知它们这个“共享数据”的状态变化信息，以执行对应的处理。
在一个条件变量上等待会发生以下“原子操作”：释放相关的互斥量，等待其他线程发给该条件变量的信号（唤醒一个等待者-即等待线程）或广播该条件变量（唤醒所有等待者）。当等待条件变量时，互斥量必须始终锁住；当线程从条件变量等待中醒来时，它重新继续锁住互斥量。

思路

在生产者/消费者问题中，生产者线程必须在缓冲区中有可用空间后才能向其中放置内容，否则将阻塞(进入休眠状态)直到出现下一个可用的空位置。生产者线程可使用互斥量原子性地检查缓冲区，而不受其他线程干扰。当发现缓冲区已满后，生产者阻塞自己并在缓冲区变为非满时被唤醒，这可由条件变量实现。
消费者线程必须在生产者向缓冲区中写入之后才能从中提取内容。同理，可用互斥量和条件变量以无竞争的方式等待缓冲区由空变为非空。
此处，N按照产品数重新定义。互斥量mutex用于保护全局队列mutex和两个条件变量。条件变量notFull用于队列由满变为非满时通知(唤醒)生产者线程，而notEmpty用于队列由空变为非空时通知消费者线程。

代码

Linux代码

#include <stdio.h>
#include <pthread.h>
#include <unistd.h>
#include <stdlib.h>
#define N 100
#define true 1
#define producerNum  10//生产者数量
#define consumerNum  5//消费者数量
#define sleepTime 1
typedef int item;
item buffer[N] = {0};//产品缓冲区
int in = 0;
int out = 0;
int proCount = 0;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t notFull = PTHREAD_COND_INITIALIZER; //Full->Not Full
pthread_cond_t notEmpty = PTHREAD_COND_INITIALIZER; //Empty->Not Empty
//生产者进程
void * producer(void *a){while(true){pthread_mutex_lock(&mutex);//上锁 while((out+1)%N==in){pthread_cond_wait(&notFull,&mutex);//唤醒生产者进程 } proCount++;printf("生产一个产品ID%d,缓冲区位置为%d\n",proCount,in);buffer[in] = proCount;in = (in+1)%N;pthread_mutex_signal(&notEmpty);//此处必然处于缓冲区有数据状态，改变取数线程条件变量为signaledpthread_mutex_unlock(&mutex);//释放锁 sleep(sleepTime);}
}
//消费者进程
void * consumer(void *b){while(true){pthread_mutex_lock(&mutex);//上锁 while(in==out){pthread_cond_wait(&notEmpty,&mutex);}//消费产品 int nextc = buffer[out];buffer[out]=0;out = (out+1)%N;printf("消费一个产品ID%d,缓冲区位置为%d\n",nextc,out);pthread_cond_signal(&notFull);pthread_mutex_unlock(&mutex);//释放锁 sleep(sleepTime);}
}
int main(){pthread_t threadPool[producerNum+consumerNum];int i;for(i=0;i<producerNum;i++){pthread_t temp;if(pthread_create(&temp,NULL,producer,NULL)==-1){printf("ERROR,fail to create producer%d\n",i);exit(1);}threadPool[i] = temp;}//创建生产者进程放入线程池for(i=0;i<consumerNum;i++){pthread_t temp;if(pthread_create(&temp,NULL,consumer,NULL)==-1){printf("ERROR,fail to create consumer%d\n",i);exit(1);}threadPool[i+producerNum] = temp;}//创建消费者进程放入线程池void * result;for(i=0;i<producerNum+consumerNum;i++){if(pthread_join(threadPool[i],&result)==-1){printf("fail to recollect\n");exit(1);}}return 0;
}

window版本

代码来源于：博客

#include <stdio.h>
#include <pthread.h>
#include <windows.h>
#define N 100
#define true 1
#define producerNum  10
#define consumerNum  5
#define sleepTime 1000typedef int semaphore;
typedef int item;
item buffer[N] = {0};
int in = 0;
int out = 0;
int proCount = 0;
semaphore mutex = 1, empty = N, full = 0, proCmutex = 1;void * producer(void * a){while(true){while(proCmutex <= 0);proCmutex--;proCount++;printf("生产一个产品ID%d, 缓冲区位置为%d\n",proCount,in);proCmutex++;while(empty <= 0){printf("缓冲区已满！\n");}empty--;while(mutex <= 0);mutex--;buffer[in] = proCount;in = (in + 1) % N;mutex++;full++;Sleep(sleepTime);}
}void * consumer(void *b){while(true){while(full <= 0){printf("缓冲区为空！\n");}full--;while(mutex <= 0);mutex--;int nextc = buffer[out];buffer[out] = 0;//消费完将缓冲区设置为0out = (out + 1) % N;mutex++;empty++;printf("\t\t\t\t消费一个产品ID%d,缓冲区位置为%d\n", nextc,out);Sleep(sleepTime);}
}int main()
{pthread_t threadPool[producerNum+consumerNum];int i;for(i = 0; i < producerNum; i++){pthread_t temp;if(pthread_create(&temp, NULL, producer, NULL) == -1){printf("ERROR, fail to create producer%d\n", i);exit(1);}threadPool[i] = temp;}//创建生产者进程放入线程池for(i = 0; i < consumerNum; i++){pthread_t temp;if(pthread_create(&temp, NULL, consumer, NULL) == -1){printf("ERROR, fail to create consumer%d\n", i);exit(1);}threadPool[i+producerNum] = temp;}//创建消费者进程放入线程池void * result;for(i = 0; i < producerNum+consumerNum; i++){if(pthread_join(threadPool[i], &result) == -1){printf("fail to recollect\n");exit(1);}}//运行线程池return 0;
}