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); //消费者线程2WaitForMultipleObjects(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;
}