生产者-消费者算法

#includewindows.h#includefstream.h#includestdio.h#includestring#includeconio.h//定义一些常量；//本程序允许的最大临界区数；#defineMAX_BUFFER_NUM10//秒到微秒的乘法因子；#defineINTE_PER_SEC1000//本程序允许的生产和消费线程的总数；#defineMAX_THREAD_NUM64//定义一个结构，记录在测试文件中指定的每一个线程的参数structThreadInfo{intserial;//线程序列号charentity;//是P还是Cdoubledelay;//线程延迟intthread_request[MAX_THREAD_NUM];//线程请求队列intn_request;//请求个数};//全局变量的定义//临界区对象的声明,用于管理缓冲区的互斥访问；CRITICAL_SECTIONPC_Critical[MAX_BUFFER_NUM];intBuffer_Critical[MAX_BUFFER_NUM];//缓冲区声明，用于存放产品；HANDLEh_Thread[MAX_THREAD_NUM];//用于存储每个线程句柄的数组；ThreadInfoThread_Info[MAX_THREAD_NUM];//线程信息数组；HANDLEempty_semaphore;//一个信号量；HANDLEh_mutex;//一个互斥量；DWORDn_Thread=0;//实际的线程的数目；DWORDn_Buffer_or_Critical;//实际的缓冲区或者临界区的数目；HANDLEh_Semaphore[MAX_THREAD_NUM];//生产者允许消费者开始消费的信号量；//生产消费及辅助函数的声明voidProduce(void*p);voidConsume(void*p);boolIfInOtherRequest(int);intFindProducePositon();intFindBufferPosition(int);intmain(void){//声明所需变量；DWORDwait_for_all;ifstreaminFile;//初始化缓冲区；for(inti=0;iMAX_BUFFER_NUM;i++)Buffer_Critical[i]=-1;//初始化每个线程的请求队列；for(intj=0;jMAX_THREAD_NUM;j++){for(intk=0;kMAX_THREAD_NUM;k++)Thread_Info[j].thread_request[k]=-1;Thread_Info[j].n_request=0;}//初始化临界区；for(i=0;iMAX_BUFFER_NUM;i++)InitializeCriticalSection(&PC_Critical[i]);//打开输入文件，按照规定的格式提取线程等信息；inFile.open(test.txt);//从文件中获得实际的缓冲区的数目；inFilen_Buffer_or_Critical;inFile.get();printf(输入文件是:\n);//回显获得的缓冲区的数目信息；printf(%d\n,(int)n_Buffer_or_Critical);//提取每个线程的信息到相应数据结构中；while(inFile){inFileThread_Info[n_Thread].serial;inFileThread_Info[n_Thread].entity;inFileThread_Info[n_Thread].delay;charc;inFile.get(c);while(c!='\n'&&!inFile.eof()){inFileThread_Info[n_Thread].thread_request[Thread_Info[n_Thread].n_request++];inFile.get(c);}n_Thread++;}//回显获得的线程信息，便于确认正确性；for(j=0;j(int)n_Thread;j++){intTemp_serial=Thread_Info[j].serial;charTemp_entity=Thread_Info[j].entity;doubleTemp_delay=Thread_Info[j].delay;printf(\nthread%2d%c%f,Temp_serial,Temp_entity,Temp_delay);intTemp_request=Thread_Info[j].n_request;for(intk=0;kTemp_request;k++)printf(%d,Thread_Info[j].thread_request[k]);coutendl;}printf(\n\n);//创建在模拟过程中几个必要的信号量empty_semaphore=CreateSemaphore(NULL,n_Buffer_or_Critical,n_Buffer_or_Critical,semaphore_for_empty);h_mutex=CreateMutex(NULL,FALSE,mutex_for_update);//下面这个循环用线程的ID号来为相应生产线程的产品读写时所//使用的同步信号量命名；for(j=0;j(int)n_Thread;j++){std::stringlp=semaphore_for_produce_;inttemp=j;while(temp){charc=(char)(temp%10);lp+=c;temp/=10;}h_Semaphore[j+1]=CreateSemaphore(NULL,0,n_Thread,lp.c_str());}//创建生产者和消费者线程；for(i=0;i(int)n_Thread;i++){if(Thread_Info[i].entity=='P')h_Thread[i]=CreateThread(NULL,0,(LPTHREAD_START_ROUTINE)(Produce),&(Thread_Info[i]),0,NULL);elseh_Thread[i]=CreateThread(NULL,0,(LPTHREAD_START_ROUTINE)(Consume),&(Thread_Info[i]),0,NULL);}//主程序等待各个线程的动作结束；wait_for_all=WaitForMultipleObjects(n_Thread,h_Thread,TRUE,-1);printf(\n\nALLProducerandconsumerhavefinishedtheirwork.\n);printf(Pressanykeytoquit!\n);_getch();return0;}//确认是否还有对同一产品的消费请求未执行；boolIfInOtherRequest(intreq){for(inti=0;in_Thread;i++)for(intj=0;jThread_Info[i].n_request;j++)if(Thread_Info[i].thread_request[j]==req)returnTRUE;returnFALSE;}//找出当前可以进行产品生产的空缓冲区位置；intFindProducePosition(){intEmptyPosition;for(inti=0;in_Buffer_or_Critical;i++)if(Buffer_Critical[i]==-1){EmptyPosition=i;//用下面这个特殊值表示本缓冲区正处于被写状态；Buffer_Critical[i]=-2;break;}returnEmptyPosition;}//找出当前所需生产者生产的产品的位置；intFindBufferPosition(intProPos){intTempPos;for(inti=0;in_Buffer_or_Critical;i++)if(Buffer_Critical[i]==ProPos){TempPos=i;break;}returnTempPos;}//生产者进程voidProduce(void*p){//局部变量声明；DWORDwait_for_semaphore,wait_for_mutex,m_delay;intm_serial;//获得本线程的信息；m_serial=((ThreadInfo*)(p))-serial;m_delay=(DWORD)(((ThreadInfo*)(p))-delay*INTE_PER_SEC);Sleep(m_delay);//开始请求生产printf(Producer%2dsendstheproducerequire.\n,m_serial);//确认有空缓冲区可供生产，同时将空位置数empty减1；用于生产者和消费者的同步；wait_for_semaphore=WaitForSingleObject(empty_semaphore,-1);//互斥访问下一个可用于生产的空临界区，实现写写互斥；wait_for_mutex=WaitForSingleObject(h_mutex,-1);intProducePos=FindProducePosition();ReleaseMutex(h_mutex);//生产者在获得自己的空位置并做上标记后，以下的写操作在生产者之间可以并发；//核心生产步骤中，程序将生产者的ID作为产品编号放入，方便消费者识别;printf(Producer%2dbegintoproduceatposition%2d.\n,m_serial,ProducePos);Buffer_Critical[ProducePos]=m_serial;printf(Producer%2dfinishproducing:\n,m_serial);printf(position[%2d]:%3d\n,ProducePos,Buffer_Critical[ProducePos]);//使生产者写的缓冲区可以被多个消费者使用，实现读写同步；ReleaseSemaphore(h_Semaphore[m_serial],n_Thread,NULL);}//消费者进程voidConsume(void*p){//局部变量声明；DWORDwait_for_semaphore,m_delay;intm_serial,m_requestNum;//消费者的序列号和请求的数目；intm_thread_request[MAX_THREAD_NUM];//本消费线程的请求队列；//提取本线程的信息到本地；m_serial=((ThreadInfo*)(p))-serial;m_delay=(DWORD)(((ThreadInfo*)(p))-delay*INTE_PER_SEC);m_requestNum=((ThreadInfo*)(p))-n_request;for(inti=0;im_requestNum;i++)m_thread_request[i]=((ThreadInfo*)(p))-thread_request[i];Sleep(m_delay);//循环进行所需产品的消费for(i=0;im_requestNum;i++){//请求消费下一个产品printf(Consumer%2drequesttoconsume%2dproduct\n,m_serial,m_thread_request[i]);//如果对应生产者没有生产，则等待；如果生产了,允许的消费者数目-1；实现了读写同步；wait_for_semaphore=WaitForSingleO