有一個生產(chǎn)者在生產(chǎn)產(chǎn)品岔擂，這些產(chǎn)品將提供給若干個消費者去消費七冲，為了使生產(chǎn)者和消費者能并發(fā)執(zhí)行蝉稳，在兩者之間設(shè)置一個具有多個緩沖區(qū)的緩沖池，生產(chǎn)者將它生產(chǎn)的產(chǎn)品放入一個緩沖區(qū)中，消費者可以從緩沖區(qū)中取走產(chǎn)品進(jìn)行消費朋截，顯然生產(chǎn)者和消費者之間必須保持同步拗慨，即不允許消費者到一個空的緩沖區(qū)中取產(chǎn)品剧蹂，也不允許生產(chǎn)者向一個已經(jīng)放入產(chǎn)品的緩沖區(qū)中再次投放產(chǎn)品。

代碼如下：

#include <stdio.h>
#include <windows.h>
#include <time.h>
#include <iostream>
#include <thread>
#include <condition_variable>
#include <mutex>
#include <vector>
#include <deque>
#include <assert.h>
#include <set>
using namespace std;

#define POOL_SIZE      4        // 緩沖池含有的緩沖區(qū)的個數(shù)
#define BUF_SIZE       1        // 緩沖區(qū)能夠緩沖元素的個數(shù)
#define PRODUCT_NUM    1000     // 生產(chǎn)的總產(chǎn)品數(shù)目
#define PRODUCER_NUM   1        // 生產(chǎn)者的數(shù)目
#define CONSUMER_NUM   2        // 消費者的數(shù)目

vector<deque<int> > pool(POOL_SIZE, deque<int>());   // 緩沖池定義
int g_num = 0;

struct s_buf_info
{
    s_buf_info(int n) : nSize(n), bLock(false) {}
    int nSize;
    bool bLock;
};
vector<s_buf_info> poolStatus(POOL_SIZE, s_buf_info(BUF_SIZE));   // 緩沖池使用狀態(tài)
set<int> InitSetWrite() {
    set<int> s;
    for (int i = 0; i < POOL_SIZE; ++i) {
        s.insert(i);
    }
    return s;
}
set<int> setWrite(InitSetWrite());
set<int> setRead;

mutex mtxPoolStatus;
condition_variable conWrite;
condition_variable conRead;

mutex mtx;  // for output data

BOOL SetConsoleColor(WORD wAttributes)
{
    HANDLE hConsole = GetStdHandle(STD_OUTPUT_HANDLE);
    if (hConsole == INVALID_HANDLE_VALUE)
        return FALSE;

    return SetConsoleTextAttribute(hConsole, wAttributes);
}

int GetWriteID()
{
    if (setWrite.empty()) return -1;

    auto itr = setWrite.begin();
    int nID = *itr;

    setWrite.erase(itr);

    if (!setRead.empty()) {
        auto itr = setRead.find(nID);
        if (itr != setRead.end())
            setRead.erase(itr);
    }

    return nID;
}

int GetReadID()
{
    if (setRead.empty()) return -1;

    auto itr = setRead.begin();
    int nID = *itr;

    setRead.erase(itr);

    if (!setWrite.empty()) {
        auto itr = setWrite.find(nID);
        if (itr != setWrite.end())
            setWrite.erase(itr);
    }

    return nID;
}

int ProductLock()
{
    unique_lock<mutex> lck(mtxPoolStatus);

    int nID = GetWriteID();
    while (nID == -1) {
        conWrite.wait(lck);
        nID = GetWriteID();
    }

    poolStatus[nID].bLock = true;

    return nID;
}

void ProductUnLock(int nID)
{
    unique_lock<mutex> lck(mtxPoolStatus);

    poolStatus[nID].bLock = false;
    poolStatus[nID].nSize -= 1;

    if (poolStatus[nID].nSize < BUF_SIZE) {
        setRead.insert(nID);
        conRead.notify_all();
    }

    if (poolStatus[nID].nSize > 0) {
        setWrite.insert(nID);
        conWrite.notify_all();
    }
}

int ConsumeLock()
{
    unique_lock<mutex> lck(mtxPoolStatus);

    int nID = GetReadID();
    while (nID == -1) {
        conRead.wait(lck);
        nID = GetReadID();
    }

    poolStatus[nID].bLock = true;

    return nID;
}

void ConsumeUnLock(int nID)
{
    unique_lock<mutex> lck(mtxPoolStatus);

    poolStatus[nID].bLock = false;
    poolStatus[nID].nSize += 1;

    if (poolStatus[nID].nSize < BUF_SIZE) {
        setRead.insert(nID);
        conRead.notify_all();
    }

    if (poolStatus[nID].nSize > 0) {
        setWrite.insert(nID);
        conWrite.notify_all();
    }

}

void ProducerThreadFun(int nProductNum)
{   
    for (int i=0 ; i < nProductNum; ++i) {
        int nID = ProductLock();
        pool[nID].push_back(g_num++);
        mtx.lock();
        printf("編號為%d生產(chǎn)者在緩沖池第%d個緩沖區(qū)中投放數(shù)據(jù)%d\n", GetCurrentThreadId(), nID, g_num - 1);
        mtx.unlock();
        ProductUnLock(nID);
    }
}

void ConsumerThreadFun(int nConsumNum)
{
    for (int i = 0; i < nConsumNum; ++i) {
        int nID = ConsumeLock();
        int nVal = pool[nID].front();
        pool[nID].pop_front();
        ConsumeUnLock(nID);
        mtx.lock();
        SetConsoleColor(FOREGROUND_GREEN);
        printf("  編號為%d的消費者從緩沖池中第%d個緩沖區(qū)取出數(shù)據(jù)%d\n", GetCurrentThreadId(), nID, nVal);
        SetConsoleColor(FOREGROUND_RED | FOREGROUND_GREEN | FOREGROUND_BLUE);
        mtx.unlock();
    }

    SetConsoleColor(FOREGROUND_GREEN);
    printf("  編號為%d的消費者收到通知瑞眼，線程結(jié)束運行\(zhòng)n", GetCurrentThreadId());
    SetConsoleColor(FOREGROUND_RED | FOREGROUND_GREEN | FOREGROUND_BLUE);
}

int main(int argc, char ** argv)
{
    time_t t1 = time(NULL);

    vector<thread> pro(PRODUCER_NUM);
    vector<thread> con(CONSUMER_NUM);

    assert(PRODUCT_NUM % PRODUCER_NUM == 0);
    assert(PRODUCT_NUM % CONSUMER_NUM == 0);

    for (auto &th : pro) th = thread(ProducerThreadFun, PRODUCT_NUM / PRODUCER_NUM);
    for (auto &th : con) th = thread(ConsumerThreadFun, PRODUCT_NUM / CONSUMER_NUM);

    for (auto &th : pro) th.join();
    for (auto &th : con) th.join();

    time_t t2 = time(NULL);
    
    cout << t2 - t1 << " s" << endl;
    return 0;
}