基于单链表和基于环形队列的生产者单消费者模型

清疚 2022-05-26 23:59 151阅读 0赞

先来介绍一下生产者消费者模型，举一个常见的例子：

生活中，我们会经常去超市买东西，这里涉及到了三个事物：我们、超市、供货商。很容易就可以想到，我们就相当于消费者，而供货商就相当于生产者，那么超市就算是一个交易场所了。

对于生产者消费者模型我们可以简单的总结一下叫做 **3 2 1原则**

*  3：三种关系
 *  2：两个对象
 *  1：一个交易场所

一个交易场所很容易理解，还有两个对象当然就算生产者和消费者了，那么三种关系是哪三种关系呢？

###### 1、生产者和生产者之间的关系 ######

还是刚刚的例子，供货商相当于生产者。假设所有的供货商都生产同一种商品，而超市只能由一家供货商供货，他们之间肯定互相竞争这个名额，那么可想而知他们之间存在着互斥关系

###### 2、消费者和消费者之间的关系 ######

和生产者类似，如果现在超市中只有一件商品了，而所有的消费者就想去拥有这款商品，同样他们就会去互相竞争，也就是存在着互斥关系

###### 3、生产者和消费者之间的关系 ######

再来看看生产者和消费者之间，想想一下如果超市的货架上一件商品都没有我们可以消费吗？在想想超市的货架上面商品都摆的满满的，供货商还会继续摆商品吗？答案是不会，所以就可以想到，消费者必须等生产者生产出商品才能消费，而生产者必须等到消费者消费商品之后才能继续生产，这是典型的同步关系。同样的生产者和消费者之间还具有着互斥关系  
在生产者把商品放到货架上的时候消费者不能去消费，必须等生产者放完才可以消费，而在消费者把商品从货架上拿走的时候生产者不能去放商品，必须等消费者完全拿走之后才可以放。

--------------------

解释清楚生产者消费者模型之后，我们来实现一下  
我们可以把超市想象成单链表，生产者就是往链表里插入数据，消费者就是从链表中删除数据。我们再利用线程来模拟生产者和消费者(这里实现的是单生产者单消费者模型，所以我们只需维护生产者和消费者之间的同步和互斥关系即可)，利用之前说过的互斥量和条件变量来维护互斥和同步（[链接戳这里][Link 1]）

好了，直接上代码

#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;
    }

来看看结果：

![这里写图片描述][70]

--------------------

刚刚我们实现的生产者消费者模型，是利用单链表模拟的交易场所实现的生产者消费者模型。当然我们还可以利用别的数据结构，来看看基于环形队列的生产者消费者模型(直接谈论多生产者多消费者模型)，首先来谈谈环形队列的概念，上一张图：  
![这里写图片描述][70 1]  
可以看到，环形队列是一段连续的空间，我们很容易就可以想到数组也是一段连续的空间，所以利用数组和模运算就很容易模拟环形队列了。

想想我们用单链表实现的生产者消费者模型维护的哪几种关系，用环形队列同样要维护，而且我们还必须遵守几条规则  
![这里写图片描述][70 2]

解释一下这几条规则：

最开始的时候，我们的队列里面没有元素，所以必须让生产者生产数据，然后消费者跟在生产者的后面消费数据，但是当生产者不能把消费者套圈，如果套圈就会覆盖消费者还没有消费的数据。如果生产者一直生产消费者没有消费的时候，这个队列就会被生产满，那么此时生产者就不能再生产了，只能等消费者消费之后露出空格，再去生产数据。

我们可以发现的是环形队列只需要一个类似于计数器的东西来标记数据数量和空格数量就可以实现生产者和消费者之间的同步和互斥机制了，很容易就可以想到利用信号量来实现。

接下来我们来介绍一下**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;
    }

来看一下结果：

![这里写图片描述][70 3]

[Link 1]: https://blog.csdn.net/qq_34021920/article/details/79822982
[70]: /images/20220527/fd0deee0f29f4f10b731fb34a843db0e.png
[70 1]: /images/20220527/f11a2afb617e46fd9c4def7bed0fe24b.png
[70 2]: /images/20220527/1a97eb29af5b42f1b450c12f474afbe5.png
[70 3]: /images/20220527/b0914dbc993d46ffac65a6bc501124cc.png