生产者消费者问题

       生产者消费者问题是一个著名的线程同步问题，该问题描述如下：有一个生产者在生产产品，这些产品将提供给若干个消费者去消费，为了使生产者和消费者能并发执行，在两者之间设置一个具有多个缓冲区的缓冲池，生产者将它生产的产品放入一个缓冲区中，消费者可以从缓冲区中取走产品进行消费，显然生产者和消费者之间必须保持同步，即不允许消费者到一个空的缓冲区中取产品，也不允许生产者向一个已经放入产品的缓冲区中再次投放产品。

 

 

先假设生产者和消费者都只有一个，且缓冲区也只有一个。这样情况就简便多了。

    第一   从缓冲区取出产品和向缓冲区投放产品必须是互斥进行的。可以用关键段和互斥量来完成。

    第二   生产者要等待缓冲区为空（或者有空位），这样才可以投放产品，消费者要等待缓冲区不为空，这样才可以取出产品进行消费。并且由于有二个等待过程，所以要用二个事件或信号量来控制。

 

#include <iostream>
#include <windows.h>
#include <process.h>

using namespace std;

const int NUM = 2, BUFFER_SIZE = 10;
int buffer;

HANDLE buffer_full, buffer_empty;
HANDLE thread[NUM];
CRITICAL_SECTION sc;

unsigned int __stdcall produce(PVOID pm) {
	int i = 0;
	for(;i < BUFFER_SIZE; i++) {
		WaitForSingleObject(buffer_empty, INFINITE);
		
		EnterCriticalSection(&sc);
		buffer = i;
		cout << "生产者生产了" << buffer << endl; 
		LeaveCriticalSection(&sc);

		SetEvent(buffer_full);
	}
	return 0;
}

unsigned int __stdcall consume(PVOID pm) {
	bool flag = true;
	while(flag){
		WaitForSingleObject(buffer_full, INFINITE);
		EnterCriticalSection(&sc);
		cout << "      消费者消费了" << buffer << endl;
		if(buffer == BUFFER_SIZE) {
			flag = false;
		}
		LeaveCriticalSection(&sc);
		SetEvent(buffer_empty);
	}
	return 0;
}

int main() {

	buffer_full = CreateEvent(NULL, FALSE, FALSE, NULL);
	buffer_empty = CreateEvent(NULL, FALSE, TRUE, NULL);
	InitializeCriticalSection(&sc);

	thread[0] = (HANDLE) _beginthreadex(NULL, 0, produce, NULL, 0, NULL);
	thread[1] = (HANDLE) _beginthreadex(NULL, 0, consume, NULL, 0, NULL);
	WaitForMultipleObjects(NUM, thread, TRUE, INFINITE);

	DeleteCriticalSection(&sc);

	getchar();
	return 0;
}

采用事件进行同步控制，运行效果如下：

 

 

接下来，我们加大缓冲区，增加消费者的个数。

需要注意：

1. 在生产者中，信号量个数的释放限制了消费者的并发数；

2. 缓冲区的大小和信号量的个数相等。 在生产者的实际操作中就不用判断缓冲区是否已满，用信号量进行限制。（在消费者中释放了多少个信号量，说明在生产者中就可以生产多少个数据）

 

代码如下：

 

#include <iostream>
#include <windows.h>
#include <process.h>

using namespace std;

const int NUM = 4, BUFFER_SIZE = 5,	END_FLAG_NUM = 20;
int buffer_p, buffer_c, p, c;
int buffer[BUFFER_SIZE];

HANDLE buffer_full, buffer_empty;
HANDLE thread[NUM];
CRITICAL_SECTION sc, oc;

unsigned int __stdcall produce(PVOID pm) {
	int i = 0;
	for(; i<END_FLAG_NUM; i++) {
		WaitForSingleObject(buffer_empty, INFINITE);
		
 		EnterCriticalSection(&sc);
		buffer[buffer_p] = i;
		cout << "生产者生产了" << buffer[buffer_p] << endl; 
		buffer_p = (i + 1) % BUFFER_SIZE;
		LeaveCriticalSection(&sc);

		ReleaseSemaphore(buffer_full, 1, NULL);	
	}
	return 0;
}

unsigned int __stdcall consume(PVOID pm) {
	bool flag = true;
	while(flag) {
		
		if(c == END_FLAG_NUM) {
			flag = false;
		} else {
			WaitForSingleObject(buffer_full, 5000);

			EnterCriticalSection(&sc);
			// 再次检测
			if(c == END_FLAG_NUM) {
				cout << "      超时或者生产者结束生产" << endl;
				flag = false;
			} else {
				cout << "      消费者消费了" << buffer[buffer_c] << endl;
				buffer_c = (++c) % BUFFER_SIZE;
			}
			LeaveCriticalSection(&sc);
			ReleaseSemaphore(buffer_empty, 1, NULL);
		}
	}
	EnterCriticalSection(&oc);
	cout << "消费者线程结束运行,线程号为：" << GetCurrentThreadId() << endl;
	LeaveCriticalSection(&oc);
	return 0;
}

int main() {

	buffer_full = CreateSemaphore(NULL, 0, BUFFER_SIZE, NULL);
	buffer_empty = CreateSemaphore(NULL, BUFFER_SIZE, BUFFER_SIZE, NULL);
	InitializeCriticalSection(&sc);
	InitializeCriticalSection(&oc);

	thread[0] = (HANDLE) _beginthreadex(NULL, 0, produce, NULL, 0, NULL);

	thread[1] = (HANDLE) _beginthreadex(NULL, 0, consume, NULL, 0, NULL);
	thread[2] = (HANDLE) _beginthreadex(NULL, 0, consume, NULL, 0, NULL);
	thread[3] = (HANDLE) _beginthreadex(NULL, 0, consume, NULL, 0, NULL);
	thread[4] = (HANDLE) _beginthreadex(NULL, 0, consume, NULL, 0, NULL);
	WaitForMultipleObjects(NUM, thread, TRUE, INFINITE);

	DeleteCriticalSection(&sc);
	DeleteCriticalSection(&oc);

	CloseHandle(buffer_full);
	CloseHandle(buffer_empty);

	cout << "运行完毕" << endl;

	getchar();
	return 0;
}

 

 

       我们假设了四个消费者，在消费者中，我们用到了双重检验法来进行互斥操作，并且其中的等待函数必须设置超时时间，否则如果设置成无限等待，可能出现有的消费线程结束，有的消费线程无限等待的情况。

       对数组下角标采取求余的方式，就可以将缓冲区看成是一个循环的模型，并且一定要保证的是消费者跟在生产者的屁股后面取数据，释放该信号量--->生产者捕获该信号量生产一个数据。

 

运行效果如下：