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 << "\t\t\t\t消費者將數據 " << 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 << "\t\t\t編號為 " << 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;
}

運行結果如下所示：

參考資料：秒殺多線程第十篇 生產者消費者問題