大学

生产者与消费者问题算法实现

by 曾经沧海
316 阅读

《生产者与消费者问题算法实现》

 设计思想

    因为有多个缓冲区，所以生产者线程没有必要在生成新的数据之前等待最后一个数据被消费者线程处理完毕。同样，消费者线程并不一定每次只能处理一个数据。在多缓冲区机制下，线程之间不必互相等待形成死锁，因而提高了效率。

　　多个缓冲区就好像使用一条传送带替代托架，传送带上一次可以放多个产品。生产者在缓冲区尾加入数据，而消费者则在缓冲区头读取数据。当缓冲区满的时候，缓冲区就上锁并等待消费者线程读取数据；每一个生产或消费动作使得传送带向前移动一个单位，因而，消费者线程读取数据的顺序和数据产生顺序是相同的。

可以引入一个count计数器来表示已经被使用的缓冲区数量。用hNotEmptyEvent 和hNotFullEvent 来同步生产者和消费者线程。每当生产者线程发现缓冲区满（ count=BufferSize ），它就等待hNotEmptyEvent 事件。同样，当消费者线程发现缓冲区空，它就开始等待hNotEmptyEvent。生产者线程写入一个新的数据之后，就立刻发出hNotEmptyEvent 来唤醒正在等待的消费者线程；消费者线程在读取一个数据之后，就发出hNotFullEvent 来唤醒正在等待的生产者线程。

程序的设计思想大致为：设置一while循环，pi生产者访问临界区，得到权限访问缓冲区，如果缓冲区满的，则等待，直到缓冲区非满；访问互斥锁，当得到互斥锁且缓冲区非满时，跳出while循环，开始产生新数据，并把数据存放于Buffer缓冲区中，当数据存放结束则结束临界区；接着唤醒消费者线程；ci消费者访问临界区，得到权限访问缓冲区，如果缓冲区为空，没有可以处理的数据，则释放互斥锁且等待，直到缓冲区非空；当等到缓冲区非空时，跳出while循环；消费者获得数据，并根据所获得的数据按类别消费（当消费者获得的数据为大写字母时，则把大写字母转换成小写字母，并显示；当消费者获得的数据为小写字母时，则把小写字母转换成大写字母，并显示；当消费者获得的数据为字符0、1、2、……8、9时，把这些字符直接显示到屏幕；当消费者获得的数据为符号（＋、－、*、\……）时，把这些符号打印成7行7列的菱形）；处理完数据后，结束临界区；接着唤醒生产者线程。     

＃i nclude<stdio.h>
＃i nclude< iostream.h> 
＃i nclude< windows.h> 
#define BufferSize 15
char Buffer[BufferSize];
int head,tail=0;//Buffer数组下标
int count;//被使用的缓冲区数量 
HANDLE hMutex;
HANDLE hNotFullEvent, hNotEmptyEvent;//用来同步生产者和消费者线程 
////////缓冲区存储情况
display(char a[15])
{
 int i;
 cout<<"缓冲区存储情况为:"<<endl;
 for (i=14;i>=0;i–){
  cout<<"\t|—-"<<a[i]<<"—-|"<<endl;
 }
}

//p1
void p1_Producer()
{
int i;
char ch;
char p1[]={'a','A','b','B','c','C','D','d','E','e'};
if(tail<15){
for(i=0;i<10;i++){
while(1) {

WaitForSingleObject(hMutex,INFINITE);
if(count==BufferSize){ //缓冲区满
ReleaseMutex(hMutex);
//等待直到缓冲区非满
WaitForSingleObject(hNotFullEvent,INFINITE);
continue;
}
//得到互斥锁且缓冲区非满,跳出while循环
break;

}
if (tail>14){
 cout<<"缓冲区已满，不能再存入数据！"<<endl;
 ReleaseMutex(hMutex); //结束临界区
    PulseEvent(hNotEmptyEvent); //唤醒消费者线程
}
else{
//得到互斥锁且缓冲区非满,开始产生新数据
cout<<"Producer p1:\t"<<p1[i]<<endl;
Buffer[tail]=p1[i];
//tail=(tail+1)%BufferSize;///存放于缓冲区的位置
display(Buffer);
tail++;
count++;
cout<<"按ENTER继续…."<<endl;
ch=getchar(); 
ReleaseMutex(hMutex); //结束临界区
PulseEvent(hNotEmptyEvent); //唤醒消费者线程
}
}
}
}
//////////////////////////////////////////////////////////////////
//p2
void p2_Producer()
{
int i;
char ch;
char p2[]={'0','1','2','3','4','5','6','7','8','9'};
if(tail<15){
for(i=0;i<10;i++){
while(1) {
 ch=getchar();
WaitForSingleObject(hMutex,INFINITE);
if(count==BufferSize){ // 缓冲区满
ReleaseMutex(hMutex);
// 等待直到缓冲区非满
WaitForSingleObject(hNotFullEvent,INFINITE);
continue;
}
// 得到互斥锁且缓冲区非满,跳出while循环
break;
}
if (tail>14){
 cout<<"缓冲区已满，不能再存入数据！程序结束！"<<endl;
 ReleaseMutex(hMutex); //结束临界区
    PulseEvent(hNotEmptyEvent); //唤醒消费者线程
}
else{
// 得到互斥锁且缓冲区非满,开始产生新数据
cout<<"Producer p2:\t"<<p2[i]<<endl;
Buffer[tail]=p2[i];
//tail=(tail+1)%BufferSize;
display(Buffer);
tail++;
count++;
cout<<"按ENTER继续…."<<endl;
ch=getchar();
ReleaseMutex(hMutex); // 结束临界区
PulseEvent(hNotEmptyEvent); // 唤醒消费者线程
}
}
}
}
//////////////////////////////////////////////////////////////////
//p3
void p3_Producer()
{
int i;
char ch;
char p3[]={'!','#','$','%','&','*','+','-','.','/'};
if(tail<15){
for(i=0;i<10;i++){
while(1) {
 ch=getchar();
WaitForSingleObject(hMutex,INFINITE);
if(count==BufferSize){ // 缓冲区满
ReleaseMutex(hMutex);
// 等待直到缓冲区非满
WaitForSingleObject(hNotFullEvent,INFINITE);
continue;
}
// 得到互斥锁且缓冲区非满,跳出while循环
break;
}
if (tail>14){
 cout<<"缓冲区已满，不能再存入数据！程序结束！"<<endl;
 ReleaseMutex(hMutex); //结束临界区
    PulseEvent(hNotEmptyEvent); //唤醒消费者线程
}
else{
// 得到互斥锁且缓冲区非满,开始产生新数据
cout<<"Producer p3:\t"<<p3[i]<<endl;
Buffer[tail]=p3[i];
//tail=(tail+1)%BufferSize;
display(Buffer);
tail++;
count++;
cout<<"按ENTER继续…."<<endl;
ch=getchar();
ReleaseMutex(hMutex); // 结束临界区
PulseEvent(hNotEmptyEvent); // 唤醒消费者线程
}
}
}
}
//////////////////////////////////////////////////////////////////
//c1
void c1_Consumer()
{
int i,j,k;
char result,ch;
while(1){
  ch=getchar(); 
WaitForSingleObject(hMutex,INFINITE);
if(count==0){ // 没有可以处理的数据
ReleaseMutex(hMutex); // 释放互斥锁且等待
// 等待直到缓冲区非空
WaitForSingleObject(hNotEmptyEvent,INFINITE);
}
else {i
f(Buffer[head]==0) {
cout<<"Consumer 0: 缓冲区的数据已全消费过一次，消费完毕！"<<endl;
ReleaseMutex(hMutex); // 结束临界区
ExitThread(0);
}
else { // 获得互斥锁且缓冲区有数据,开始处理
result=Buffer[head];
if(result>64&&result<70){
 result=result+32;
 cout<<"Consumer c1:(大写－>小写)\t "<<result<<endl;
 Buffer[head]='^';//  '^'表示数据已被消费
 cout<<"'^'表示数据已被消费"<<endl;
 display(Buffer);
}

if(result>96&&result<102){
 result=result-32;
 cout<<"Consumer c1:(小写－>大写)\t "<<result<<endl;
 Buffer[head]='^';
 cout<<"'^'表示数据已被消费"<<endl;
 display(Buffer);
}
if(result>47&&result<58){
 cout<<"Consumer c1:(显示字符)\t "<<result<<endl;
 Buffer[head]='^';
 cout<<"'^'表示数据已被消费"<<endl;
 display(Buffer);}
if(result>32&&result<48){
 cout<<"Consumer c1:(用符号打印出菱形) "<<endl;
 for(i=1;i<=(9+1)/2;i++)
 {
  for(j=1;j<=40-i;j++)
   cout<<" ";
  for(k=1;k<=2*i-1;k++)
   cout<<result;
   cout<<endl;
 }
 for(i=1;i<=9/2;i++)
 {
  for(j=1;j<=40-(9+1)/2+i;j++)
   cout<<" ";
  for(k=1;k<=9-2*i;k++)
   cout<<result;
   cout<<endl;
 }
 Buffer[head]='^';
 cout<<"'^'表示数据已被消费"<<endl;
 display(Buffer);
}

head=(head+1)%BufferSize;
count–;
cout<<"按ENTER继续…."<<endl;
ch=getchar();
ReleaseMutex(hMutex); // 结束临界区
PulseEvent(hNotFullEvent); // 唤醒生产者线程
}
}
}
}
//////////////////////////////////////////////////////////////////
//c2
void c2_Consumer()
{
int i,j,k;
char result,ch;
while(1){
WaitForSingleObject(hMutex,INFINITE);
if(count==0){ // 没有可以处理的数据
ReleaseMutex(hMutex); // 释放互斥锁且等待
// 等待直到缓冲区非空
WaitForSingleObject(hNotEmptyEvent,INFINITE);
}
else {if(Buffer[head]==0) {
cout<<"Consumer 0:缓冲区的数据已全消费过一次，消费完毕！"<<endl;
ReleaseMutex(hMutex); // 结束临界区
ExitThread(0);
}
else { // 获得互斥锁且缓冲区有数据,开始处理
result=Buffer[head];
if(result>64&&result<90){
 result=result+32;
 cout<<"Consumer c2:(大写－>小写)\t "<<result<<endl;
 Buffer[head]='^';
 cout<<"'^'表示数据已被消费"<<endl;
 display(Buffer);
}

if(result>96&&result<102){
 result=result-32;
 cout<<"Consumer c2:(小写－>大写)\t "<<result<<endl;
 Buffer[head]='^';
 cout<<"'^'表示数据已被消费"<<endl;
 display(Buffer);}
if(result>47&&result<58){
 cout<<"Consumed c2:(显示字符)\t "<<result<<endl;
 Buffer[head]='^';
 cout<<"'^'表示数据已被消费"<<endl;
 display(Buffer);}
if(result>32&&result<48){
 cout<<"Consumer c2:(用符号打印出菱形) "<<endl;
 for(i=1;i<=(9+1)/2;i++)
 {
  for(j=1;j<=40-i;j++)
   cout<<" ";
  for(k=1;k<=2*i-1;k++)
   cout<<result;
   cout<<endl;
 }
 for(i=1;i<=9/2;i++)
 {
  for(j=1;j<=40-(9+1)/2+i;j++)
   cout<<" ";
  for(k=1;k<=9-2*i;k++)
   cout<<result;
   cout<<endl;
 }
 Buffer[head]='^';
 cout<<"'^'表示数据已被消费"<<endl;
 display(Buffer);
}

head=(head+1)%BufferSize;
count–;
cout<<"按ENTER继续…."<<endl;
ch=getchar();
ReleaseMutex(hMutex); // 结束临界区
PulseEvent(hNotFullEvent); // 唤醒生产者线程
}
}
}
}
//////////////////////////////////////////////////////////////////
//c3
void c3_Consumer()
{
int i,j,k;
char result,ch;
while(1){
WaitForSingleObject(hMutex,INFINITE);
if(count==0){ // 没有可以处理的数据
ReleaseMutex(hMutex); // 释放互斥锁且等待
// 等待直到缓冲区非空
WaitForSingleObject(hNotEmptyEvent,INFINITE);
}
else {if(Buffer[head]==0) {
cout<<"Consumer 0: 缓冲区的数据已全消费过一次，消费完毕！"<<endl;
ReleaseMutex(hMutex); // 结束临界区
ExitThread(0);
}
else { // 获得互斥锁且缓冲区有数据,开始处理
result=Buffer[head];
if(result>64&&result<70){
 result=result+32;
 cout<<"Consumer c3:(大写－>小写)\t "<<result<<endl;
 Buffer[head]='^';
 cout<<"'^'表示数据已被消费"<<endl;
 display(Buffer);}

if(result>96&&result<102){
 result=result-32;
 cout<<"Consumer c3:(小写－>大写)\t "<<result<<endl;
 Buffer[head]='^';
 cout<<"'^'表示数据已被消费"<<endl;
 display(Buffer);}
if(result>47&&result<58){
 cout<<"Consumer c1:(显示字符)\t "<<result<<endl;
 Buffer[head]='^';
 cout<<"'^'表示数据已被消费"<<endl;
 display(Buffer);
}
if(result>32&&result<48){
 cout<<"Consumer c3:(用符号打印出菱形) "<<endl;
 for(i=1;i<=(7+1)/2;i++)
 {
  for(j=1;j<=40-i;j++)
   cout<<" ";
  for(k=1;k<=2*i-1;k++)
   cout<<result;
   cout<<endl;
 }
 for(i=1;i<=7/2;i++)
 {
  for(j=1;j<=40-(7+1)/2+i;j++)
   cout<<" ";
  for(k=1;k<=7-2*i;k++)
   cout<<result;
   cout<<endl;
 }
 Buffer[head]='^';
 cout<<"'^'表示数据已被消费"<<endl;
 display(Buffer);
}
head=(head+1)%BufferSize;
count–;
cout<<"按ENTER继续…."<<endl;
ch=getchar();
ReleaseMutex(hMutex); // 结束临界区
PulseEvent(hNotFullEvent); // 唤醒生产者线程
}
}
}
}
//////////////////////////////////////////////////////////////////
//主函数
void main()< br/>{
HANDLE hThreadVector[6];
DWORD ThreadID;
count = 0;
head = 0;
tail = 0;
hMutex=CreateMutex(NULL,FALSE,NULL);
hNotFullEvent=CreateEvent(NULL,TRUE,FALSE,NULL);
hNotEmptyEvent=CreateEvent(NULL,TRUE,FALSE,NULL);
hThreadVector[0]=CreateThread (NULL, 0,(LPTHREAD_START_ROUTINE) p1_Producer,NULL, 0, (LPDWORD)&ThreadID);
hThreadVector[1]=CreateThread (NULL, 0,(LPTHREAD_START_ROUTINE) c1_Consumer,NULL, 0, (LPDWORD)&ThreadID);
hThreadVector[3]=CreateThread (NULL, 0,(LPTHREAD_START_ROUTINE) p2_Producer,NULL, 0, (LPDWORD)&ThreadID);
hThreadVector[4]=CreateThread (NULL, 0,(LPTHREAD_START_ROUTINE) c2_Consumer,NULL, 0, (LPDWORD)&ThreadID);
hThreadVector[5]=CreateThread (NULL, 0,(LPTHREAD_START_ROUTINE) p3_Producer,NULL, 0, (LPDWORD)&ThreadID);
hThreadVector[5]=CreateThread (NULL, 0,(LPTHREAD_START_ROUTINE) c3_Consumer,NULL, 0, (LPDWORD)&ThreadID);
WaitForMultipleObjects(2,hThreadVector,TRUE,INFINITE);
//cout<<"**********************Finish*************************"<<endl;

1 评论

hyioi 2006年10月27日 - 10:27

thank u.

发表评论