先來介紹一下生產者消費者模型，舉一個常見的例子：

生活中，我們會經常去超市買東西，這裡涉及到了三個事物：我們、超市、供貨商。很容易就可以想到，我們就相當於消費者，而供貨商就相當於生產者，那麼超市就算是一個交易場所了。

對於生產者消費者模型我們可以簡單的總結一下叫做 3 2 1原則

• 3：三種關係
• 2：兩個物件
• 1：一個交易場所

一個交易場所很容易理解，還有兩個物件當然就算生產者和消費者了，那麼三種關係是哪三種關係呢？

1、生產者和生產者之間的關係

還是剛剛的例子，供貨商相當於生產者。假設所有的供貨商都生產同一種商品，而超市只能由一家供貨商供貨，他們之間肯定互相競爭這個名額，那麼可想而知他們之間存在著互斥關係

2、消費者和消費者之間的關係

和生產者類似，如果現在超市中只有一件商品了，而所有的消費者就想去擁有這款商品，同樣他們就會去互相競爭，也就是存在著互斥關係

3、生產者和消費者之間的關係

再來看看生產者和消費者之間，想想一下如果超市的貨架上一件商品都沒有我們可以消費嗎？在想想超市的貨架上面商品都擺的滿滿的，供貨商還會繼續擺商品嗎？答案是不會，所以就可以想到，消費者必須等生產者生產出商品才能消費，而生產者必須等到消費者消費商品之後才能繼續生產，這是典型的同步關係。同樣的生產者和消費者之間還具有著互斥關係
在生產者把商品放到貨架上的時候消費者不能去消費，必須等生產者放完才可以消費，而在消費者把商品從貨架上拿走的時候生產者不能去放商品，必須等消費者完全拿走之後才可以放。

解釋清楚生產者消費者模型之後，我們來實現一下
我們可以把超市想象成單鏈表，生產者就是往連結串列裡插入資料，消費者就是從連結串列中刪除資料。我們再利用執行緒來模擬生產者和消費者(這裡實現的是單生產者單消費者模型，所以我們只需維護生產者和消費者之間的同步和互斥關係即可)，利用之前說過的互斥量和條件變數來維護互斥和同步（連結戳這裡）

好了，直接上程式碼

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <time.h>
 


//初始化互斥量
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
//初始化條件變數
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

//連結串列節點
typedef struct Node
{
    int data;
    struct Node* next;
}Node,*pNode,**ppNode;

//申請節點
pNode BuyNode(int x)
{
    pNode newNode=(pNode)malloc(sizeof(Node));
    if(newNode==NULL)
    {
        perror("malloc");
        exit(1);
    }
    newNode->data=x;
    newNode->next=NULL;
    return newNode;
}
//初始化連結串列
void InitList(ppNode head)
{
    *head=BuyNode(0);
}
//連結串列判空
int IsEmpty(pNode head)
{
    return head->next==NULL?1:0;
}
//往連結串列中插入元素---頭插
void pushList(pNode head,int x)
{
    pNode n=BuyNode(x);
    n->next=head->next;
    head->next=n;
}
//從連結串列中刪除元素---頭刪
int popList(pNode head,int* x)
{
    if(IsEmpty(head))
        return -1;

    pNode n=head->next;
    head->next=n->next;
    *x=n->data;
    free(n);
    return 0;
}

//消費者
void* runC(void* arg)
{
    pNode lhead = *((ppNode)arg);
    int d=0;
    while(1)
    {
        pthread_mutex_lock(&mutex);//加鎖
        while(lhead->next==NULL)//如果連結串列為空
        {
            printf("consumer need to wait...\n");
            pthread_cond_wait(&cond,&mutex);//消費者掛起等待
        }
        popList(lhead,&d);//不為空消費資料
        printf("consumer get data:%d\n",d);
        sleep(2);
    }

}

//生產者
void* runP(void* arg)
{
    pNode lhead = *((ppNode)arg);
    int d=0;
    while(1)
    {
        sleep(1);//為了演示消費者等待讓生產者先執行
        d = rand()%100+1;
        pushList(lhead,d);//生產資料
        printf("product data:%d\n",d);
        pthread_cond_signal(&cond);//喚醒消費者
        pthread_mutex_unlock(&mutex);//解鎖
        sleep(3);
    }
}

int main()
{
    pthread_t c,p;

    pNode head=NULL;
    InitList(&head);

    srand((unsigned long)time(NULL));

    pthread_create(&c,NULL,runC,(void*)&head);
    pthread_create(&p,NULL,runP,(void*)&head);

    pthread_join(c,NULL);
    pthread_join(p,NULL);

    return 0;
}

來看看結果：

剛剛我們實現的生產者消費者模型，是利用單鏈表模擬的交易場所實現的生產者消費者模型。當然我們還可以利用別的資料結構，來看看基於環形佇列的生產者消費者模型(直接談論多生產者多消費者模型)，首先來談談環形佇列的概念，上一張圖：

可以看到，環形佇列是一段連續的空間，我們很容易就可以想到陣列也是一段連續的空間，所以利用陣列和模運算就很容易模擬環形隊列了。

想想我們用單鏈表實現的生產者消費者模型維護的哪幾種關係，用環形佇列同樣要維護，而且我們還必須遵守幾條規則

解釋一下這幾條規則：

最開始的時候，我們的佇列裡面沒有元素，所以必須讓生產者生產資料，然後消費者跟在生產者的後面消費資料，但是當生產者不能把消費者套圈，如果套圈就會覆蓋消費者還沒有消費的資料。如果生產者一直生產消費者沒有消費的時候，這個佇列就會被生產滿，那麼此時生產者就不能再生產了，只能等消費者消費之後露出空格，再去生產資料。

我們可以發現的是環形佇列只需要一個類似於計數器的東西來標記資料數量和空格數量就可以實現生產者和消費者之間的同步和互斥機制了，很容易就可以想到利用訊號量來實現。

接下來我們來介紹一下POSIX訊號量，之前學習程序間通訊的時候談論過system V版本的訊號量，那個訊號量只能用於程序，而POSIX訊號量既可以用於程序也可以用於執行緒，來看看它的相關函式

初始化訊號量

#include <semaphore.h>

int sem_init(sem_t *sem, int pshared, unsigned int value);
//sem：要初始化的訊號量
//pshared：填1表示程序間共享，填0表示執行緒間共享
//value：要初始化的值
//返回值：成功返回0，失敗返回-1

等待資源(P操作)

#include <semaphore.h>

int sem_wait(sem_t *sem);
//相當於對sem訊號量進行-1操作
//返回值：成功返回0，失敗返回-1

釋放資源(V操作)

#include <semaphore.h>

int sem_post(sem_t *sem);
//相當於對訊號量進行+1操作
//返回值：成功返回0，失敗返回-1

銷燬訊號量

#include <semaphore.h>

int sem_destroy(sem_t *sem);
//sem：要銷燬的訊號量
//返回值：成功返回0，失敗返回-1

由於我們編寫的是多生產者多消費者模型，所以我們還需要維持生產者與生產者之間的互斥關係，還有消費者與消費者之間的互斥關係，同樣的我們需要用到互斥量

好了，直接上程式碼：

#include <stdio.h>
#include <stdlib.h>
#include <semaphore.h>
#include <time.h>
#include <pthread.h>

#define M 10
#define C 3
#define P 3
int ring[M];//環形佇列，利用陣列和%運算實現

sem_t sem_data;//資料量
sem_t sem_blank;//空格子量

pthread_mutex_t lock1=PTHREAD_MUTEX_INITIALIZER;//消費者之間的互斥
pthread_mutex_t lock2=PTHREAD_MUTEX_INITIALIZER;//生產者之間的互斥

//消費者
void* runC(void* arg)
{
    static int i=0;
    int d;
    while(1)
    {
        pthread_mutex_lock(&lock1);//加鎖，必須給臨界區的所以程式碼加鎖
        sem_wait(&sem_data);//消費者申請資料資源
        d=ring[i];
        printf("consumer data:%d\n",d);
        i++;
        i%=M;//模運算，防止下標越界
        sem_post(&sem_blank);//消費資料後對空格資源V操作
        pthread_mutex_unlock(&lock1);
        sleep(3);
    }
}
//生產者
void* runP(void* arg)
{
    int data=0;
    static int i=0;
    while(1)
    {
        pthread_mutex_lock(&lock2);
        data=rand()%100+1;
        sem_wait(&sem_blank);//生產者申請空格資源
        ring[i]=data;
        printf("product data:%d\n",data);
        i++;
        i%=M;//模運算。防止下標越界
        sem_post(&sem_data);//生產者生產完後對資料資源V操作
        pthread_mutex_unlock(&lock2);
        sleep(2);
    }
}

int main()
{
    srand((unsigned long)time(NULL));

    pthread_t consumer[C];
    pthread_t product[P];
    int i=0;
    for(i=0;i<C;i++)//建立多個消費者執行緒
    {
        pthread_create(&consumer[i],NULL,runC,NULL);
    }
    for(i=0;i<P;i++)//建立多個生產者執行緒
    {
        pthread_create(&product[i],NULL,runP,NULL);
    }
    sem_init(&sem_data,0,0);
    sem_init(&sem_blank,0,M);

    for(i=0;i<C;i++)
    {
        pthread_join(consumer[i],NULL);
    }
    for(i=0;i<P;i++)
    {
        pthread_join(product[i],NULL);
    }

    sem_destroy(&sem_data);
    sem_destroy(&sem_blank);

    pthread_mutex_destroy(&lock1);
    pthread_mutex_destroy(&lock2);

    return 0;
}

來看一下結果：