C++线程同步消费者生产者实例.docx

C+线程同步成产者消费者举例1. 测试用例文件说明：本用例是在VS2010（中文版） 环境下进行。项目创建流程（只供参考）：文件 新建 Visual C+ Win32 项目 输入项目名（ThreadPC） 确定 下一步 选择控制台应用程序 完成测试用例文件用于描述各线程的有关信息，该文件内容及格式如下(将下列内容复制到test.txt文件中)：31P32P43C414P25C3124说明： 第一行给出的是程序中设置的临界区个数；其余各行是各进程信息。 每行中的数据之间用Tab键分隔。 第一列(除第一行外)：线程号（共5个线程）。 第二列：P生产者，C消费者。 第三列：线程在生产和消费前的休眠时间，单位为秒。 第四及以后各列：消费的产品所对应的生产者线程号。2. 数据结构(1) 用整型数组Buffer_Critical表示缓冲区。/本程序允许的最大临界区数#define MAX_BUFFER_NUM 10int Buffer_CriticalMAX_BUFFER_NUM;/缓冲区声明，用于存放产品(2) 用自定义结构ThreadInfo记录一条线程信息，多个线程对应一个ThreadInfo数组。struct ThreadInfoint serial;/线程序列号char entity;/是P还是Cdouble delay;/线程延迟int thread_requestMAX_THREAD_NUM;/线程请求队列int n_request;/请求个数;(3) 通过如下同步对象实现互斥：l 设一个互斥量h_mutex，实现生产者在查询和保留缓冲区的下一个空位置时进行互斥。HANDLE h_mutex;/一个互斥量l 设置h_SemaphoreMAX_THREAD_NUM信号量数组表示相应产品已经生产，实现生产者与消费者之间的同步。同时，用表示空缓冲区数目的信号量empty_semephore指示是否存在空位置，实现类似的同步，以便开始下一个产品的生产。/本程序允许的生产和消费线程的总数#define MAX_THREAD_NUM 64HANDLE h_SemaphoreMAX_THREAD_NUM;/生产者允许消费者开始消费的信号量l 设置临界区对象数组PC_CriticalMAX_BUFFER_NUM实现每个缓冲区上消费者之间的互斥。CRITICAL_SECTION PC_CriticalMAX_BUFFER_NUM;/临界区对象的声明,用于管理缓冲区的互斥访问3. 程序流程为了方便，程序结构用如下的文字予以描述。 (1) 主函数 (2) 初始化缓冲区、消费请求队列及部分同步对象 (3) 提取线程信息(到test.txt文件中提取) (4) 完成线程相关同步对象的初始化 (5) 创建线程，模拟生产者和消费者 (6) 等待所有线程结束 (7) 程序结束 (8) 消费者 (9) 有无消费请求？有，则继续(10)；无，则转(16) (10) 此请求可满足？可满足，转(11)；否，则阻塞，再转(10) (11) 确定产品位置 (12) 此产品正被消费？是，则阻塞，再转(12)；否，则转(13) (13) 进入临界区(请求同一产品的消费者之间互斥) (14) 消费产品，并判断是否应该释放产品所占缓冲区 (15) 退出临界区，转(9) (16) 结束消费者线程 (17) 生产者 (18) 存在空缓冲区？有，则继续(19)；无，则阻塞，再转(18) (19) 另一生产者在写？否，则转(20)；是，则阻塞，再转(19) (20) 进入临界区(请求同一产品的生产者之间互斥) (21) 在缓冲区中为本线程产品分配空间 (22) 退出临界区 (23) 写入产品到分配的缓冲区空间中 (24) 结束生产者线程4. 代码实现/ ThreadPC.cpp : 定义控制台应用程序的入口点#include stdafx.h#include#include#include#include#include#includeusing namespace std;/定义一些常量/本程序允许的最大临界区数#define MAX_BUFFER_NUM 10/秒到微秒的乘法因子#define INTE_PER_SEC 1000/本程序允许的生产和消费线程的总数#define MAX_THREAD_NUM 64/定义一个结构，记录在测试文件中指定的每一个线程的参数struct ThreadInfoint serial;/线程序列号char entity;/是P还是Cdouble delay;/线程延迟int thread_requestMAX_THREAD_NUM;/线程请求队列int n_request;/请求个数;/全局变量的定义/struct _RTL_CRITICAL_SECTION;/typedef _RTL_CRITICAL_SECTION RTL_CRITICAL_SECTION;/typedef RTL_CRITICAL_SECTION CRITICAL_SECTION;CRITICAL_SECTION PC_CriticalMAX_BUFFER_NUM;/临界区对象的声明,用于管理缓冲区的互斥访问int Buffer_CriticalMAX_BUFFER_NUM;/缓冲区声明，用于存放产品HANDLE h_ThreadMAX_THREAD_NUM;/用于存储每个线程句柄的数组ThreadInfo Thread_InfoMAX_THREAD_NUM;/线程信息数组HANDLE empty_semaphore;/一个信号量HANDLE h_mutex;/一个互斥量DWORD n_Thread=0;/实际的线程的数目DWORD n_Buffer_or_Critical;/实际的缓冲区或者临界区的数目HANDLE h_SemaphoreMAX_THREAD_NUM;/生产者允许消费者开始消费的信号量/生产消费及辅助函数的声明void Produce(void*p);void Consume(void*p);bool IfInOtherRequest(int);int FindProducePosition();int FindBufferPosition(int);int main(void)/声明所需变量int i = 0, j = 0;DWORD wait_for_all;ifstream inFile;/输入文件/初始化缓冲区，即产品缓冲区，存储生产的产品，产品为正整数，-1表示空for(int i=0;iMAX_BUFFER_NUM;i+)Buffer_Criticali=-1;/初始化每个线程的请求队列for(int j=0;jMAX_THREAD_NUM;j+)for(int k=0;kMAX_THREAD_NUM;k+)Thread_Infoj.thread_requestk=-1;/-1表示无请求，正整数表示生产者请求生产的产品或消费者请求消费的产品Thread_Infoj.n_request=0;/初始化临界区对象列表，使用临界区是必须初始化for(i=0;in_Buffer_or_Critical;/=3inFile.get();/读取nprintf(输入文件是:n);/回显获得的缓冲区的数目信息printf(缓冲区的数目是：%dn,(int)n_Buffer_or_Critical);/提取每个线程的信息到相应数据结构中while(inFile)inFileThread_Infon_Thread.serial;/线程号inFileThread_Infon_Thread.entity;/P生产者，C消费者inFileThread_Infon_Thread.delay;/线程在生产和消费前的休眠时间，单位为秒char c;inFile.get(c);while(c!=n&!inFile.eof()/c!=n说明delay列后面还有数据inFileThread_Infon_Thread.thread_requestThread_Infon_Thread.n_request+;inFile.get(c);n_Thread+;/输出获得的线程信息，便于确认正确性cout从文件中获得的线程信息endl;coutserialtentitytdelaytrequestt.endl;for(j=0;j(int)n_Thread;j+)int Temp_serial=Thread_Infoj.serial;char Temp_entity=Thread_Infoj.entity;double Temp_delay=Thread_Infoj.delay;printf(n线程%2dt%ct%ft,Temp_serial,Temp_entity,Temp_delay);int Temp_request=Thread_Infoj.n_request;for(int k=0;kTemp_request;k+)printf(%dt,Thread_Infoj.thread_requestk);coutendl;printf(nn);/创建信号量empty_semaphore控制同一时刻访问空缓冲区域的线程数目empty_semaphore = CreateSemaphore(NULL,n_Buffer_or_Critical,n_Buffer_or_Critical,(LPCWSTR)semaphore_for_empty);/互斥量h_mutex，实现生产者在查询和保留缓冲区的下一个空位置时进行互斥h_mutex = CreateMutex(NULL,FALSE,(LPCWSTR)mutex_for_update);/下面这个循环用线程的ID号来为相应生产线程的产品读写时所使用的同步信号量命名for(j=0;j(int)n_Thread;j+)std:string lp = semaphore_for_produce_;int temp=j;while(temp)/将tmp转化为字符串char c=(char)(temp%10)+0;lp+=c;temp/=10;/信号量h_Semaphorei控制被生产者线程h_Threadi生产的线程可以被多少个消费线程使用h_Semaphorej+1 = CreateSemaphore(NULL,0,(LONG)n_Thread,(LPCWSTR)lp.c_str();/创建生产者和消费者线程for(i=0;i(int)n_Thread;i+)if(Thread_Infoi.entity=P)h_Threadi=CreateThread(NULL,0,(LPTHREAD_START_ROUTINE)(Produce), &(Thread_Infoi),0,NULL);elseh_Threadi=CreateThread(NULL,0,(LPTHREAD_START_ROUTINE)(Consume), &(Thread_Infoi),0,NULL);/主程序等待各个线程的动作结束wait_for_all=WaitForMultipleObjects(n_Thread,h_Thread,TRUE,-1);printf(nn所有的生产者和消费者已经完成他们的工作.n);printf(按任意键退出！n);_getch();return 0;/确认是否还有对同一产品的消费请求未执行bool IfInOtherRequest(int req)for(unsigned int i=0;in_Thread;i+)for(int j=0;jThread_Infoi.n_request;j+)if(Thread_Infoi.thread_requestj=req)return TRUE;return FALSE;/找出当前可以进行产品生产的空缓冲区位置int FindProducePosition()int EmptyPosition;for(unsigned int i=0;in_Buffer_or_Critical;i+)if(Buffer_Criticali=-1)EmptyPosition = i;/用下面这个特殊值表示本缓冲区正处于被写状态Buffer_Criticali = -2;break;return EmptyPosition;/找出当前所需生产者生产的产品的位置int FindBufferPosition(int ProPos)int TempPos;for(unsigned int i=0;iserial;/本线程序号m_delay = (DWORD)(ThreadInfo*)(p)-delay*INTE_PER_SEC);/开始生产或消费之前的等待时间Sleep(m_delay);/开始请求生产printf(生产者%2d发送生产请求！n,m_serial);/确认有空缓冲区可供生产，同时将控制访问空位置线程数目的信号量empty_semaphore的计数器减wait_for_semaphore = WaitForSingleObject(empty_semaphore,-1);/等待互斥量，互斥访问下一个可用于生产的空临界区，实现写互斥wait_for_mutex = WaitForSingleObject(h_mutex,-1);/生产者获取生产位置，有互斥量可保证不同生产这可以找到不同的生产位置，同时信号量empty_semaphore可以保证生产者线程数不超过空位置数int ProducePos = FindProducePosition();/释放互斥量ReleaseMutex(h_mutex);/生产者在获得自己的空位置并做上标记后，以下的写操作在生产者之间可以并发在核心生产步骤中，程序将生产者的ID作为产品编号放入，方便消费者识别printf(生产者%2d开始在位置%2d上生产！n, m_serial, ProducePos);/生产者生产中Buffer_CriticalProducePos = m_serial;printf(生产者%2d完成生产；n ,m_serial);printf(位置%2d：%3dn, ProducePos, Buffer_CriticalProducePos);/使生产者写的缓冲区可以被多个消费者使用，实现读写同步ReleaseSemaphore(h_Semaphorem_serial,n_Thread,NULL);/消费者进程void Consume(void*p)/局部变量声