多线程生产者消费者问题

    最近在学习多线程，自己尝试写了下生产者消费者问题，

    思路：

    1、使用两个信号量表示缓冲区中可以取的数据个数和可以放数据的位子个数

    2、生产者和消费者的存取数据时，使用关键段达到缓冲区互斥访问的目的

    代码记录如下：

    

//生产者消费者问题

#include<stdio.h>
#include<process.h>
#include<Windows.h>

#define PRODUCE_NUM 1            //生产者数量
#define CUST_NUM 2              //消费者数量
#define BUFFER_LENGTH 8         //缓冲区长度
#define SOURCE_LENGTH 15        //源数据长度
#define END_FLAG -1              //数据结束标志

typedef struct
{
    int aBuffer[BUFFER_LENGTH];
    int iHeadIndex;
    int iEndIndex;
}Buffer;
Buffer g_Buffer;
CRITICAL_SECTION g_csBuffer;    //缓冲区关键段
CRITICAL_SECTION g_csSource;    //源数据关键段
CRITICAL_SECTION g_csPinrt;     //打印数据关键段，此关键段仅用于打印时显示不同的颜色
HANDLE g_hSemaphoreEmpty;       //缓冲区空位信号量
HANDLE g_hSemaphoreFull;        //缓冲区有数据信号量
int g_Source[SOURCE_LENGTH];    //源数据，各生产者线程从这里获取数据
int g_SourceIndex;

//从源数据获取数据，若没数据，返回END_FLAG
int GetDataFromSource()
{
    int data = 0;
    EnterCriticalSection(&g_csSource);
    if(g_SourceIndex == SOURCE_LENGTH)
    {
        data = END_FLAG;
    }
    else
    {
        data = g_Source[g_SourceIndex++];
    }
    LeaveCriticalSection(&g_csSource);
    return data;
}

//向缓冲区发送数据
void SendDataToBuffer(int data)
{
    EnterCriticalSection(&g_csBuffer);
    g_Buffer.aBuffer[g_Buffer.iHeadIndex] = data;
    g_Buffer.iHeadIndex = (g_Buffer.iHeadIndex + 1) % BUFFER_LENGTH;
    LeaveCriticalSection(&g_csBuffer);
    ReleaseSemaphore(g_hSemaphoreFull,1,NULL);
}

//从缓冲区取数据
int GetDataFromBuffer()
{
    int data = 0;

    EnterCriticalSection(&g_csBuffer);
    data = g_Buffer.aBuffer[g_Buffer.iEndIndex];
    g_Buffer.iEndIndex = (g_Buffer.iEndIndex + 1) % BUFFER_LENGTH;
    LeaveCriticalSection(&g_csBuffer);
    ReleaseSemaphore(g_hSemaphoreEmpty,1,NULL);
    return data;
}

//设置打印字体颜色
BOOL SetConsoleColor(WORD wAttributes)
{
    HANDLE hConsole = GetStdHandle(STD_OUTPUT_HANDLE);
    if(hConsole == NULL)
    {
        return FALSE;
    }
    return SetConsoleTextAttribute(hConsole,wAttributes);
}

//生产者线程函数
unsigned int __stdcall ProducerThreadFun(void *pM)
{
    int data = 0;
    BOOL bSourceFlag = TRUE;
    while(bSourceFlag)
    {
        //等待缓冲区为空
        WaitForSingleObject(g_hSemaphoreEmpty,INFINITE);

        data = GetDataFromSource();
        if(data == END_FLAG)
        {
            SendDataToBuffer(data);
            bSourceFlag = FALSE;
        }
        else
        {
            SendDataToBuffer(data);
            EnterCriticalSection(&g_csPinrt);
            printf("生产者%d 将数据%d放入缓冲区\n",GetCurrentThreadId(),data);
            LeaveCriticalSection(&g_csPinrt);
        }
    }
    printf("生产者线程%d结束\n",GetCurrentThreadId());
    return 0;
}

//消费者线程函数
unsigned int __stdcall ConcumerThreadFun(void *pM)
{
    volatile BOOL flag = TRUE;
    int data = 0;
    while(flag)
    {
        //等待缓冲区中有数据
        WaitForSingleObject(g_hSemaphoreFull,INFINITE);
        data = GetDataFromBuffer(); 
        if(data == END_FLAG)
        {
            SendDataToBuffer(data);
            flag = FALSE;
        }
        else
        {
            EnterCriticalSection(&g_csPinrt);
            SetConsoleColor(FOREGROUND_GREEN);
            printf("  消费者%d 从缓冲区中取数据%d\n",GetCurrentThreadId(),data);
            SetConsoleColor(FOREGROUND_RED | FOREGROUND_GREEN | FOREGROUND_BLUE); 
            LeaveCriticalSection(&g_csPinrt);
        }
        Sleep(10);
    }

    printf("消费者线程%d结束\n",GetCurrentThreadId());
    return 0;
}

//初始化数据源
void InitialSourceData()
{
    int i = 0;
    for(i = 0;i < SOURCE_LENGTH;i++)
    {
        g_Source[i] = i;
    }
    g_SourceIndex = 0;
}

int main()
{
    HANDLE hProduceThread[PRODUCE_NUM];
    HANDLE hCustThread[CUST_NUM];
    int i = 0;
    
    //初始化关键段
    InitializeCriticalSection(&g_csBuffer);
    InitializeCriticalSection(&g_csSource);
    InitializeCriticalSection(&g_csPinrt);
    
    //初始化信号量
    g_hSemaphoreEmpty = CreateSemaphore(NULL,BUFFER_LENGTH,BUFFER_LENGTH,NULL);
    g_hSemaphoreFull = CreateSemaphore(NULL,0,BUFFER_LENGTH,NULL);
    
    //初始化源数据
    InitialSourceData();

    //创建消费者线程
    for(i = 0;i < CUST_NUM;i++)
    {
        hCustThread[i] = (HANDLE)_beginthreadex(NULL,0,ConcumerThreadFun,NULL,0,NULL);
    }

    //创建生产者线程
    for(i = 0;i < PRODUCE_NUM;i++)
    {
        hProduceThread[i] = (HANDLE)_beginthreadex(NULL,0,ProducerThreadFun,NULL,0,NULL);
    }
    WaitForMultipleObjects(PRODUCE_NUM,hProduceThread,TRUE,INFINITE);
    WaitForMultipleObjects(CUST_NUM,hCustThread,TRUE,INFINITE);
    //关闭各内核对象，省略
    system("pause");
    return 0;
}

生产者2个，消费者4个，缓冲区长度8，数据源长度15时，结果如下