生产者消费者模型

r囧r小猫 2022-06-13 07:57 298阅读 0赞

生产者消费者模型遵循三二一原则  
三种关系：生产者和生产者 ：互斥  
消费者和消费者：互斥  
生产者和消费者：同步与互斥  
两种角色  
一个交易场所

条件变量的作用：判断临界资源中生产消费的条件是否成熟

通常情况下互斥锁和条件变量搭配使用

对单链表的操作
    #include<stdio.h>
    #include<pthread.h>
    #include <stdlib.h>
    typedef struct Node
    {
        int data;
       struct Node* next;
    }Node,**ppNode;
    static pthread_mutex_t lock=PTHREAD_MUTEX_INITIALIZER;
    static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
    Node*  pHead=NULL;
    Node*GetNode(int val)
    {
         Node*tmp=(Node*)malloc(sizeof(Node));
         if(tmp==NULL)
         {
              perror("malloc");
              return NULL;
         }
         tmp->data=val;
         tmp->next=NULL;
         return tmp;
    }
    void InitList(ppNode p)
    {
         *p=GetNode(0);
    }
    void PushFront(Node* pHead,const int data)
    {
       Node*tmp=GetNode(data);
       tmp->next=pHead->next;
       pHead->next=tmp;
    }
    void PopFront(Node* pHead,int*out)
    {
         Node*tmp=pHead->next;
         if(tmp)
         {
             Node*next=tmp->next;
             pHead->next=next;
              *out=tmp->data;
             free(tmp);
              tmp=NULL;
         }
    }
    void *product(void *arg)
    {
         while(1)
         {
            int data=rand()%1000;
            pthread_mutex_lock(&lock);
            PushFront(pHead,data);
            pthread_mutex_unlock(&lock);
            pthread_cond_signal(&cond);
            printf("product done:%d\n!",data);
            sleep(1);
         }
    }
    void *consume(void*arg)
    {
         while(1)
         {
              int c=-1;
          pthread_mutex_lock(&lock);
           while(pHead->next==NULL)
           {
                printf("consume is waiting!\n");
                pthread_cond_wait(&cond,&lock);
           }
           PopFront(pHead,&c);
           pthread_mutex_unlock(&lock);
           printf("consume is done :%d\n",c);
         }
    }
    int main()
    {
        InitList(&pHead);
         pthread_t tproduct;
         pthread_t tconsume;
         pthread_create(&tproduct,NULL,product,NULL);
         pthread_create(&tconsume,NULL,consume,NULL);
         pthread_join(tproduct,NULL);
         pthread_join(tconsume,NULL);
         pthread_mutex_destroy(&lock);
         pthread_cond_destroy(&cond);
       return 0;
    }

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

![这里写图片描述][SouthEast 1]

我们可以通过信号量来实现  
sem\_wait()可以获得资源（P操作）,使semaphore的值减1,如果调⽤用sem\_wait()时 semaphore的值 已经是0,则挂起等待。如果不希望挂起等待,可以调⽤用sem\_trywait() 。调⽤用 sem\_post() 可以释放资源（V操作）,使semaphore 的值加1,同时唤醒挂起等待的线程。总的来说也是通过信号量来实现同步  
初始化一个线程使用sem\_init

#include <semaphore.h>
     int sem_init(sem_t *sem, int pshared, unsigned int value);
    #include <semaphore.h>
    int sem_wait(sem_t *sem);
    int sem_trywait(sem_t *sem);

#include<stdio.h>
    #include<pthread.h>
    #include <semaphore.h>
    #define SIZE 10
    sem_t data_sem;
    sem_t blank_sem;
    int arr[SIZE]={
       0};
    void* product(void*argv)
    {
         int ret=0;
         int i=0;
        while(1)
         {
              sem_wait(&blank_sem);
              arr[i++]=ret;
              sem_post(&data_sem);
              printf("product done:%d!\n",ret++);
              i%=SIZE;
         }
    }
    void* consume(void*argv)
    {
         int i=0;
         while(1)
         {
          sem_wait(&data_sem);
           int ret=arr[i++];
           sem_post(&blank_sem);
           printf("consume done:%d\n",ret);
           i%=SIZE;
           sleep(1);
         }
    }
    int main()
    {
         pthread_t p;
         pthread_t c;
        sem_init(&data_sem,0,0);
         sem_init(&blank_sem,0,SIZE);
         pthread_create(&p,NULL,product,NULL);
         pthread_create(&c,NULL,consume,NULL);
         pthread_join(p,NULL);
         pthread_join(c,NULL);
         sem_destroy(&data_sem);
         sem_destroy(&blank_sem);
         return 0;
    }

![这里写图片描述][SouthEast 2]  
针对多生产者多消费者的模型：  
pthread\_mutex\_t mutex = PTHREAD\_MUTEX\_INITIALIZER;  
必须要加锁  
生产者和生产者是互斥的  
消费者和消费者互斥的  
生产者和消费者要满足不套圈，消费者不能超过生产者

#include<stdio.h>
    #include<pthread.h>
    #include <semaphore.h>
    #define SIZE 10
    sem_t data_sem;
    sem_t blank_sem;
    static int i=0;
    static int ret=0;
    pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
    int arr[SIZE]={
       0};
    void* product(void*argv)
    {
        while(1)
        {
            sem_wait(&blank_sem);
            pthread_mutex_lock(&mutex);
            arr[i++]=ret;
            sem_post(&data_sem);
            printf("product done:ret=%d!\n",i,ret++);
            i%=SIZE;
            pthread_mutex_unlock(&mutex);
            sleep(1);
        }
    }
    void* consume(void*argv)
    {
        int ret=0;
       static int i=0;
        while(1)
        {
          sem_wait(&data_sem);
          pthread_mutex_lock(&mutex);
          int ret=arr[i++];
          sem_post(&blank_sem);
          printf("consume done:ret=%d\n",i,ret);
          i%=SIZE;
          pthread_mutex_unlock(&mutex);
          //sleep(1);
        }
    }
    int main()
    {
        pthread_t p1;
        pthread_t p2;
        pthread_t c1;
        pthread_t c2;
        sem_init(&data_sem,0,0);
        sem_init(&blank_sem,0,SIZE);
        pthread_create(&p1,NULL,product,NULL);
        pthread_create(&p2,NULL,product,NULL);
        pthread_create(&c1,NULL,consume,NULL);
        pthread_create(&c2,NULL,consume,NULL);
        pthread_join(p1,NULL);
        pthread_join(p2,NULL);
        pthread_join(c1,NULL);
        pthread_join(c2,NULL);
        sem_destroy(&data_sem);
        sem_destroy(&blank_sem);
        return 0;
    }

[SouthEast]: /images/20220613/71d47984a6d347779739c182b55414fc.png
[SouthEast 1]: /images/20220613/f1ce63135f9348da95702267552d766e.png
[SouthEast 2]: /images/20220613/c54b83b6e7624fd995e53e5f235b8919.png