windows多线程（十） 生产者与消费者问题

一、概述

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

二、实例

(一) 一个生产者，一个消费者，一个缓冲区。

要满足生产者与消费者关系，我们需要保证以下两点：

• 第一．从缓冲区取出产品和向缓冲区投放产品必须是互斥进行的。可以用关键段和互斥量来完成。
• 第二．生产者要等待缓冲区为空，这样才可以投放产品，消费者要等待缓冲区不为空，这样才可以取出产品进行消费。并且由于有二个等待过程，所以要用二个事件或信号量来控制。

代码实现如下:

//生产者消费者问题,一个生产者，一个消费者，一个缓冲区。
#include <iostream>
#include <windows.h>

using namespace std;

DWORD WINAPI ProducerThread(LPVOID);
DWORD WINAPI ConsumerThread(LPVOID);

const int PRODUCT_NUM = 10;  //总共生产10个产品
int g_Buffer = 0;		//缓冲区
CRITICAL_SECTION g_csVar;  //互斥锁
HANDLE g_hEventBufEmpty, g_hEventBufFull;


int main()
{
	InitializeCriticalSection(&g_csVar);
	g_hEventBufEmpty = CreateEvent(NULL, false, true, NULL);  //缓冲区为空事件
	g_hEventBufFull = CreateEvent(NULL, false, false, NULL);  //缓冲区满事件

	const int THREAD_NUM = 2;
	HANDLE handle[THREAD_NUM];
	handle[0] = CreateThread(NULL, 0, ProducerThread, NULL, 0, NULL); //生产者线程
	handle[1] = CreateThread(NULL, 0, ConsumerThread, NULL, 0, NULL); //消费者线程
	WaitForMultipleObjects(THREAD_NUM, handle, true, INFINITE);

	DeleteCriticalSection(&g_csVar);
	CloseHandle(handle[0]);
	CloseHandle(handle[1]);
	CloseHandle(g_hEventBufEmpty);
	CloseHandle(g_hEventBufFull);
	return 0;
}


DWORD WINAPI ProducerThread(LPVOID p)
{
	for (int i = 1; i <= PRODUCT_NUM; i++)
	{
		WaitForSingleObject(g_hEventBufEmpty, INFINITE);	//等待缓冲区为空
		EnterCriticalSection(&g_csVar);
		g_Buffer = i;
		cout << "生产者将数据 " << g_Buffer << " 放入缓冲区！" << endl;
		LeaveCriticalSection(&g_csVar);
		SetEvent(g_hEventBufFull);		//触发事件，缓冲区满
	}

	return 0;
}

DWORD WINAPI ConsumerThread(LPVOID p)
{
	for (int i = 1; i <= PRODUCT_NUM; i++)
	{
		WaitForSingleObject(g_hEventBufFull, INFINITE);  //等待缓冲区满
		EnterCriticalSection(&g_csVar);
		cout << "				消费者将数据 " << g_Buffer << " 从缓冲区取出！" << endl;
		LeaveCriticalSection(&g_csVar);
		SetEvent(g_hEventBufEmpty);   //触发事件，清空缓冲区
	}
	return 0;
}

运行结果如下，生产者等待缓冲区为空的时候才向缓冲区投放产品，消费者等待缓冲区满的时候才取走产品。

(二) 一个生产者，两个消费者，一个缓冲池（四个缓冲区）

相比于一个生产者，一个消费者，一个缓冲区，生产者由一个变成多个不难处理，多开线程就可以，需要注意的是缓冲区的变化，可以利用两个信号量就可以解决这种缓冲池有多个缓冲区的情况。用一个信号量A来记录为空的缓冲区个数，另一个信号量B记录非空的缓冲区个数，然后生产者等待信号量A，消费者等待信号量B就可以了。

代码实现如下:

// 一个生产者，两个消费者，一个缓冲池（四个缓冲区）

#include <iostream>
#include <windows.h>
using namespace std;

DWORD WINAPI ProducerThread(LPVOID);
DWORD WINAPI ConsumerThread(LPVOID);	// 两个消费者，开两个线程就行了



const int PRODUCT_NUM = 16; //产品总数
const int BUFFER_SIZE = 4;  //缓冲区大小
int g_Buffer[BUFFER_SIZE];
CRITICAL_SECTION g_csVar; // 互斥锁
HANDLE g_hEventBufEmpty, g_hEventBufFull;
int g_i = 0, g_j = 0;


int main()
{
	InitializeCriticalSection(&g_csVar);
	g_hEventBufEmpty = CreateSemaphore(NULL, 4, 4, NULL);   //记录空缓冲区个数信号量
	g_hEventBufFull = CreateSemaphore(NULL, 0, 4, NULL);    //记录满缓冲区个数信号量
	const int THREAD_NUM = 3; //线程数
	HANDLE handle[THREAD_NUM];
	memset(g_Buffer, 0, sizeof(g_Buffer));	//缓冲池清零
	handle[0] = CreateThread(NULL, 0, ProducerThread, NULL, 0, NULL); //生产者线程
	handle[1] = CreateThread(NULL, 0, ConsumerThread, NULL, 0, NULL); //消费者线程1
	handle[2] = CreateThread(NULL, 0, ConsumerThread, NULL, 0, NULL); //消费者线程2

	WaitForMultipleObjects(THREAD_NUM, handle, true, INFINITE);
	for (int i = 0; i<THREAD_NUM; i++)
	{
		CloseHandle(handle[i]);
	}
	CloseHandle(g_hEventBufEmpty);
	CloseHandle(g_hEventBufFull);
	DeleteCriticalSection(&g_csVar);

	return 0;
}


DWORD WINAPI ProducerThread(LPVOID p)
{
	for (int i = 1; i <= PRODUCT_NUM; i++)
	{
		WaitForSingleObject(g_hEventBufEmpty, INFINITE);  //生产者等待空缓冲区
		EnterCriticalSection(&g_csVar);
		g_Buffer[g_i] = i;
		cout << "生产者在第 " << g_i << " 个缓冲池中放入数据 " << g_Buffer[g_i] << endl;
		g_i = (g_i + 1) % BUFFER_SIZE;   //g_i自增，并实现在缓冲池中循环
		LeaveCriticalSection(&g_csVar);
		ReleaseSemaphore(g_hEventBufFull, 1, NULL);   //生产完产品后，记录满缓冲区个数信号量加一，即记录现有产品数
	}
	cout << "生产者完成任务，线程结束运行!" << endl;
	return 0;
}


DWORD  WINAPI ConsumerThread(LPVOID p)
{
	for (int i = 1; i <= PRODUCT_NUM; i++)
	{
		WaitForSingleObject(g_hEventBufFull, INFINITE);  //消费者等待缓冲区有产品(不为空)
		EnterCriticalSection(&g_csVar);
		cout << "			编号为 " << GetCurrentThreadId() << " 的消费者在第 " << g_j << " 个缓冲池中取走数据 " << g_Buffer[g_j] << endl;

		if (g_Buffer[g_j] == PRODUCT_NUM)  //最后一个产品已经被取走，此时需要退出消费者线程。
		{
			LeaveCriticalSection(&g_csVar);
			ReleaseSemaphore(g_hEventBufFull, 1, NULL);  //这里信号量加一，通知其它消费者有数据了(实际没有),使其它消费者执行这里的if语句，结束线程。
			break;
		}
		g_j = (g_j + 1) % BUFFER_SIZE;  //g_i自增，并实现在缓冲池中循环
		LeaveCriticalSection(&g_csVar);
		ReleaseSemaphore(g_hEventBufEmpty, 1, NULL);
	}
	cout << "编号为 " << GetCurrentThreadId() << " 的消费者结束运行！ " << endl;
	return 0;
}

运行结果如下所示：

参考资料：秒杀多线程第十篇 生产者消费者问题