#include<windows.h>
#include<fstream.h>
#include<stdio.h>
#include<string>
#include<conio.h>
//定義一些常量；
//本程序允許的最大臨界區(qū)數(shù)；
#define MAX_BUFFER_NUM 10
//秒到微秒的乘法因子；
#define INTE_PER_SEC 1000
//本程序允許的生產(chǎn)和消費(fèi)線程的總數(shù)；
#define MAX_THREAD_NUM 64
//定義一個(gè)結(jié)構(gòu)，記錄在測(cè)試文件中指定的每一個(gè)線程的參數(shù)
struct ThreadInfo
{
int serial; //線程序列號(hào)
char entity; //是P還是C
double delay; //線程延遲
int thread_request[MAX_THREAD_NUM]; //線程請(qǐng)求隊(duì)列
int n_request; //請(qǐng)求個(gè)數(shù)
};
//全局變量的定義
//臨界區(qū)對(duì)象的聲明,用于管理緩沖區(qū)的互斥訪問；
CRITICAL_SECTION PC_Critical[MAX_BUFFER_NUM];
int Buffer_Critical[MAX_BUFFER_NUM]; //緩沖區(qū)聲明，用于存放產(chǎn)品；
HANDLE h_Thread[MAX_THREAD_NUM]; //用于存儲(chǔ)每個(gè)線程句柄的數(shù)組；
ThreadInfo Thread_Info[MAX_THREAD_NUM]; //線程信息數(shù)組；
HANDLE empty_semaphore; //一個(gè)信號(hào)量；
HANDLE h_mutex; //一個(gè)互斥量；
DWORD n_Thread = 0; //實(shí)際的線程的數(shù)目；
DWORD n_Buffer_or_Critical; //實(shí)際的緩沖區(qū)或者臨界區(qū)的數(shù)目；
HANDLE h_Semaphore[MAX_THREAD_NUM]; //生產(chǎn)者允許消費(fèi)者開始消費(fèi)的信號(hào)量；
//生產(chǎn)消費(fèi)及輔助函數(shù)的聲明
void Produce(void *p);
void Consume(void *p);
bool IfInOtherRequest(int);
int FindProducePositon();
int FindBufferPosition(int);
int main(void)
{
//聲明所需變量；
DWORD wait_for_all;
ifstream inFile;
//初始化緩沖區(qū)；
for(int i=0;i< MAX_BUFFER_NUM;i++)
Buffer_Critical[i] = -1;
//初始化每個(gè)線程的請(qǐng)求隊(duì)列；
for(int j=0;j<MAX_THREAD_NUM;j++){
for(int k=0;k<MAX_THREAD_NUM;k++)
Thread_Info[j].thread_request[k] = -1;
Thread_Info[j].n_request = 0;
}
//初始化臨界區(qū)；
for(i =0;i< MAX_BUFFER_NUM;i++)
InitializeCriticalSection(&PC_Critical[i]);
//打開輸入文件，按照規(guī)定的格式提取線程等信息；
inFile.open("test.txt");
//從文件中獲得實(shí)際的緩沖區(qū)的數(shù)目；
inFile >> n_Buffer_or_Critical;
inFile.get();
printf("輸入文件是:\n");
//回顯獲得的緩沖區(qū)的數(shù)目信息；
printf("%d \n",(int) n_Buffer_or_Critical);
//提取每個(gè)線程的信息到相應(yīng)數(shù)據(jù)結(jié)構(gòu)中；
while(inFile){
inFile >> Thread_Info[n_Thread].serial;
inFile >> Thread_Info[n_Thread].entity;
inFile >> Thread_Info[n_Thread].delay;
char c;
inFile.get(c);
while(c!='\n'&& !inFile.eof()){
inFile>> Thread_Info[n_Thread].thread_request[Thread_Info[n_Thread].n_request++];
inFile.get(c);
}
n_Thread++;
}
//回顯獲得的線程信息，便于確認(rèn)正確性；
for(j=0;j<(int) n_Thread;j++){
int Temp_serial = Thread_Info[j].serial;
char Temp_entity = Thread_Info[j].entity;
double Temp_delay = Thread_Info[j].delay;
printf(" \n thread%2d %c %f ",Temp_serial,Temp_entity,Temp_delay);
int Temp_request = Thread_Info[j].n_request;
for(int k=0;k<Temp_request;k++)
printf(" %d ", Thread_Info[j].thread_request[k]);
cout<<endl;
}
printf("\n\n");
//創(chuàng)建在模擬過程中幾個(gè)必要的信號(hào)量
empty_semaphore=CreateSemaphore(NULL,n_Buffer_or_Critical,n_Buffer_or_Critical,
"semaphore_for_empty");
h_mutex = CreateMutex(NULL,FALSE,"mutex_for_update");
//下面這個(gè)循環(huán)用線程的ID號(hào)來為相應(yīng)生產(chǎn)線程的產(chǎn)品讀寫時(shí)所
//使用的同步信號(hào)量命名；
for(j=0;j<(int)n_Thread;j++){
std::string lp ="semaphore_for_produce_";
int temp =j;
while(temp){
char c = (char)(temp%10);
lp+=c;
temp/=10;
}
h_Semaphore[j+1]=CreateSemaphore(NULL,0,n_Thread,lp.c_str());
}
//創(chuàng)建生產(chǎn)者和消費(fèi)者線程；
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);
else
h_Thread[i]=CreateThread(NULL,0,(LPTHREAD_START_ROUTINE)(Consume),
&(Thread_Info[i]),0,NULL);
}
//主程序等待各個(gè)線程的動(dòng)作結(jié)束；
wait_for_all = WaitForMultipleObjects(n_Thread,h_Thread,TRUE,-1);
printf(" \n \nALL Producer and consumer have finished their work. \n");
printf("Press any key to quit!\n");
_getch();
return 0;
}
//確認(rèn)是否還有對(duì)同一產(chǎn)品的消費(fèi)請(qǐng)求未執(zhí)行；
bool IfInOtherRequest(int req)
{
for(int i=0;i<n_Thread;i++)
for(int j=0;j<Thread_Info[i].n_request;j++)
if(Thread_Info[i].thread_request[j] == req)
return TRUE;
return FALSE;
}
//找出當(dāng)前可以進(jìn)行產(chǎn)品生產(chǎn)的空緩沖區(qū)位置；
int FindProducePosition()
{
int EmptyPosition;
for (int i =0;i<n_Buffer_or_Critical;i++)
if(Buffer_Critical[i] == -1){
EmptyPosition = i;
//用下面這個(gè)特殊值表示本緩沖區(qū)正處于被寫狀態(tài)；
Buffer_Critical[i] = -2;
break;
}
return EmptyPosition;
}
//找出當(dāng)前所需生產(chǎn)者生產(chǎn)的產(chǎn)品的位置；
int FindBufferPosition(int ProPos)
{
int TempPos;
for (int i =0 ;i<n_Buffer_or_Critical;i++)
if(Buffer_Critical[i]==ProPos){
TempPos = i;
break;
}
return TempPos;
}
//生產(chǎn)者進(jìn)程
void Produce(void *p)
{
//局部變量聲明；
DWORD wait_for_semaphore,wait_for_mutex,m_delay;
int m_serial;
//獲得本線程的信息；
m_serial = ((ThreadInfo*)(p))->serial;
m_delay = (DWORD)(((ThreadInfo*)(p))->delay *INTE_PER_SEC);
Sleep(m_delay);
//開始請(qǐng)求生產(chǎn)
printf("Producer %2d sends the produce require.\n",m_serial);
//確認(rèn)有空緩沖區(qū)可供生產(chǎn)，同時(shí)將空位置數(shù)empty減1；用于生產(chǎn)者和消費(fèi)者的同步；
wait_for_semaphore = WaitForSingleObject(empty_semaphore,-1);
//互斥訪問下一個(gè)可用于生產(chǎn)的空臨界區(qū)，實(shí)現(xiàn)寫寫互斥；
wait_for_mutex = WaitForSingleObject(h_mutex,-1);
int ProducePos = FindProducePosition();
ReleaseMutex(h_mutex);
//生產(chǎn)者在獲得自己的空位置并做上標(biāo)記后，以下的寫操作在生產(chǎn)者之間可以并發(fā)；
//核心生產(chǎn)步驟中，程序?qū)⑸a(chǎn)者的ID作為產(chǎn)品編號(hào)放入，方便消費(fèi)者識(shí)別;
printf("Producer %2d begin to produce at position %2d.\n",m_serial,ProducePos);
Buffer_Critical[ProducePos] = m_serial;
printf("Producer %2d finish producing :\n ",m_serial);
printf(" position[ %2d ]:%3d \n" ,ProducePos,Buffer_Critical[ProducePos]);
//使生產(chǎn)者寫的緩沖區(qū)可以被多個(gè)消費(fèi)者使用，實(shí)現(xiàn)讀寫同步；
ReleaseSemaphore(h_Semaphore[m_serial],n_Thread,NULL);
}
//消費(fèi)者進(jìn)程
void Consume(void * p)
{
//局部變量聲明；
DWORD wait_for_semaphore,m_delay;
int m_serial,m_requestNum; //消費(fèi)者的序列號(hào)和請(qǐng)求的數(shù)目；
int m_thread_request[MAX_THREAD_NUM];//本消費(fèi)線程的請(qǐng)求隊(duì)列；
//提取本線程的信息到本地；
m_serial = ((ThreadInfo*)(p))->serial;
m_delay = (DWORD)(((ThreadInfo*)(p))->delay *INTE_PER_SEC);
m_requestNum = ((ThreadInfo *)(p))->n_request;
for (int i = 0;i<m_requestNum;i++)
m_thread_request[i] = ((ThreadInfo*)(p))->thread_request[i];
Sleep(m_delay);
//循環(huán)進(jìn)行所需產(chǎn)品的消費(fèi)
for(i =0;i<m_requestNum;i++){
//請(qǐng)求消費(fèi)下一個(gè)產(chǎn)品
printf("Consumer %2d request to consume %2d product\n",m_serial,m_thread_request[i]);
//如果對(duì)應(yīng)生產(chǎn)者沒有生產(chǎn)，則等待；如果生產(chǎn)了,允許的消費(fèi)者數(shù)目-1；實(shí)現(xiàn)了讀寫同步；
wait_for_semaphore=WaitForSingleObject(h_Semaphore[m_thread_request[i]],-1);
//查詢所需產(chǎn)品放到緩沖區(qū)的號(hào)
int BufferPos=FindBufferPosition(m_thread_request[i]);
//開始進(jìn)行具體緩沖區(qū)的消費(fèi)處理，讀和讀在該緩沖區(qū)上仍然是互斥的；
//進(jìn)入臨界區(qū)后執(zhí)行消費(fèi)動(dòng)作；并在完成此次請(qǐng)求后，通知另外的消費(fèi)者本處請(qǐng)求已
//經(jīng)滿足；同時(shí)如果對(duì)應(yīng)的產(chǎn)品使用完畢，就做相應(yīng)處理；并給出相應(yīng)動(dòng)作的界面提
//示；該相應(yīng)處理指將相應(yīng)緩沖區(qū)清空，并增加代表空緩沖區(qū)的信號(hào)量；
EnterCriticalSection(&PC_Critical[BufferPos]);
printf("Consumer%2d begin to consume %2d product \n",m_serial,m_thread_request[i]);
((ThreadInfo*)(p))->thread_request[i] =-1;
if(!IfInOtherRequest(m_thread_request[i])){
Buffer_Critical[BufferPos] = -1;//標(biāo)記緩沖區(qū)為空；
printf("Consumer%2d finish consuming %2d:\n ",m_serial,m_thread_request[i]);
printf(" position[ %2d ]:%3d \n" ,BufferPos,Buffer_Critical[BufferPos]);
ReleaseSemaphore(empty_semaphore,1,NULL);
}
else{
printf("Consumer %2d finish consuming product %2d\n ",m_serial,m_thread_request[i]);
}
//離開臨界區(qū)
LeaveCriticalSection(&PC_Critical[BufferPos]);
}
}
看函數(shù)就行
