线程同步和互斥（条件变量控制生产者消费者模型）-编程语言-互联网-天达云

线程同步和互斥（条件变量控制生产者消费者模型）
更新：HHH 时间：2023-1-7

条件变量

有关函数：

当向条件变量发送一个信号时，如果没有线程等待在条件变量，那么该信号会丢失。

生产者消费者模型：

关系：

同步

生产者<—————>消费者

互斥

生产者<—————>生产者

互斥

消费者<—————>消费者

场所：

缓冲区，下文以链表方式实现

1.单个生产者，单个消费者，且生产者和消费者访问链表的顺序是LIFO的

代码实现：

#include<stdio.h>
#include<pthread.h>

pthread_mutex_t _mutex_lock=PTHREAD_MUTEX_INITIALIZER;

pthread_cond_t need_product=PTHREAD_COND_INITIALIZER;

typedef struct List List;
struct List
{
    int _var;
    List *_next;
}*head=NULL;
//head=NULL;

void* product(void *arg)
{
    while(1){
        pthread_mutex_lock(&_mutex_lock);
        List* p=(List*)malloc(sizeof(List));
        p->_var=(rand()%2000);
        p->_next=head;
        head=p;
        printf("call consumer! product success,val is :%d\n",p->_var);
        pthread_mutex_unlock(&_mutex_lock);
        sleep(rand()%3);
        pthread_cond_signal(&need_product);
    }
}

void* consumer(void *arg)
{
    while(1){
        pthread_mutex_lock(&_mutex_lock);
        if(head==NULL){
            pthread_cond_wait(&need_product,&_mutex_lock);
        }
        List *p=head;
        head=head->_next;
        p->_next=NULL;
        pthread_mutex_unlock(&_mutex_lock);
        printf("consumer has get protect:%d\n",p->_var);
        free((void*)p);
        p=NULL;
    }
}

int main()
{
    int err;
    pthread_t p;
    pthread_t c;

    err=pthread_create(&p,NULL,product,NULL);
    if(err!=0){
        printf("%s\n",strerror(err));
    }

    err=pthread_create(&c,NULL,consumer,NULL);
    if(err!=0){
        printf("%s\n",strerror(err));
    }

    void *status;
    pthread_join(p,&status);
     printf("%s\n",status);
    pthread_join(c,&status);
    printf("%s\n",status);

    pthread_cond_destroy(&need_product);

    pthread_mutex_destroy(&_mutex_lock);

    return 0;
}

运行结果：

==================================================================

2.单个生产者，单个消费者，且生产者和消费者访问链表的顺序是FIFO的

代码实现：

fifo_cond.c

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

pthread_mutex_t lock=PTHREAD_MUTEX_INITIALIZER;
//pthread_cond_t need_product=PTHREAD_COND_INITIALIZER;
pthread_cond_t need_product;

typedef int datatype;

typedef struct node{
    datatype _val;
    struct node *_next;
}node,*node_p,**node_pp;

typedef struct list_info
{
    node_p _head;
    node_p _tail;
}list_info,*list_info_p;

void init_list(node_pp head,node_pp last)
{
    *head=NULL;
    *last=*head;
}

node_p buy_node(datatype data)
{
    node_p p=(node_p)malloc(sizeof(node));
    p->_val=data;
    p->_next=NULL;

    if(p==NULL){
        return NULL;
    }
    return p;
}

int push(node_pp list,node_pp last,datatype data)
{
    node_p p=buy_node(data);
    if(p==NULL){
        return -1;
    }
    if(*last==NULL){
        *last=p;
        *list=p;
    }
    else{
        (*last)->_next=p;
        (*last)=(*last)->_next;
    }
    return data;
}

node_p destroy_node(node_pp list,node_pp last)
{
    if((*list)!=NULL){
        node_p tmp=*list;
        *list=(*list)->_next;
        return tmp;
    }
    *last=NULL;
    return NULL;
}

int pop(node_pp list,node_pp last)
{
    node_p p=destroy_node(list,last);
    if(p==NULL)
        return -1;
    int data=p->_val;
    free(p);
    p=NULL;
    return data;
}

int show_list(node_p list,node_p last)
{
    while(list!=last){
        printf("%d->",list->_val);
        list=list->_next;
    }
    if(last!=NULL)
        printf("%d\n",last->_val);
}

void *product(void *arg)
{
    while(1){
        datatype data=rand()%100;
        pthread_mutex_lock(&lock);
        push(&(((list_info_p)arg)->_head),&(((list_info_p)arg)->_tail),data);
        printf("Product success,val:%d\n",data);
        pthread_mutex_unlock(&lock);
        sleep(2);
        pthread_cond_signal(&need_product);
    }
}

void *consumer(void *arg)
{
    while(1){
        int ret=-1;
//        sleep(4);
        pthread_mutex_lock(&lock);
        while(-1==(ret=pop(&(((list_info_p)arg)->_head),&(((list_info_p)arg)->_tail)))){
            pthread_cond_wait(&need_product,&lock);
        }
        pthread_mutex_unlock(&lock);
        printf("consumer success,val:%d\n",ret);
     }
}

int main()
{
    pthread_cond_init(&need_product,NULL);

    node_p head,tail;
    init_list(&head,&tail);

    list_info _list_info;
    _list_info._head=head;
    _list_info._tail=tail;

    pthread_t tid1,tid2;
    pthread_create(&tid1,NULL,product,(void*)(&_list_info));
    pthread_create(&tid2,NULL,consumer,(void*)(&_list_info));

    pthread_join(tid1,NULL);
    pthread_join(tid2,NULL);
    
    pthread_cond_destroy(&need_product);
    pthread_mutex_destroy(&lock);
//  int data=0;
//  while(data<10){
//      push(&head,&last,data);
//      show_list(head,last);
//      data++;
//  }
//
//  while(data>0){
//      pop(&head,&last);
//      show_list(head,last);
//      data--;
//  }
    return 0;
}

运行结果：

consumer()有sleep(4);

运行结果：

consumer()函数中没有sleep(4)这条语句

从以上两次结果可以看出消费者是按找生产产品的顺序来消费的，如果生产者生产的慢，消费者会等待

==================================================================

3.多个生产者，多个消费者

实现代码：

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

pthread_mutex_t lock=PTHREAD_MUTEX_INITIALIZER;
//pthread_cond_t need_product=PTHREAD_COND_INITIALIZER;
pthread_cond_t need_product;

typedef int datatype;

typedef struct node{
    datatype _val;
    struct node *_next;
}node,*node_p,**node_pp;

node_p head,tail;

typedef struct list_info
{
    node_p _head;
    node_p _tail;

    int _flag;
}list_info,*list_info_p;

void init_list(node_pp head,node_pp last)
{
    *head=NULL;
    *last=*head;
}

node_p buy_node(datatype data)
{
    node_p p=(node_p)malloc(sizeof(node));
    p->_val=data;
    p->_next=NULL;

    if(p==NULL){
        return NULL;
    }
    return p;
}

int push(node_pp list,node_pp last,datatype data)
{
    node_p p=buy_node(data);
    if(p==NULL){
        return -1;
    }
    if(*last==NULL){
        *last=p;
         *list=p;
    }
    else{
        (*last)->_next=p;
        (*last)=(*last)->_next;
    }
    return data;
}

node_p destroy_node(node_pp list,node_pp last)
{
    if((*list)!=NULL){
        node_p tmp=*list;
        *list=(*list)->_next;
//      if(*list==NULL)  //###########error###############
//          *last==NULL;
        return tmp;
    }
    *last=NULL;
    return NULL;
}

int pop(node_pp list,node_pp last)
{
    node_p p=destroy_node(list,last);
    if(p==NULL)
        return -1;
    int data=p->_val;
    free(p);
    p=NULL;
    return data;
}

int show_list(node_p list,node_p last)
{
    while(list!=last){
        printf("%d->",list->_val);
        list=list->_next;
    }
    if(last!=NULL)
        printf("%d\n",last->_val);
}

void *product(void *arg)
{
    while(1){
        datatype data=rand()%100;
        pthread_mutex_lock(&lock);
        push(&head,&tail,data);
        printf("Product%d put success,val:%d\n",(int)arg,data);
        pthread_mutex_unlock(&lock);
        sleep(1);
//        pthread_cond_signal(&need_product);
        pthread_cond_broadcast(&need_product);
    }
}

void *consumer(void *arg)
{
     while(1){
         int ret=-1;
         sleep(2);
         pthread_mutex_lock(&lock);
         while(-1==(ret=pop(&head,&tail))){
             pthread_cond_wait(&need_product,&lock);
         }
         pthread_mutex_unlock(&lock);
         sleep(1);
         printf("consumer%d take success,val:%d\n",(int)arg,ret);
     }
}

int main()
{
     init_list(&head,&tail);

       pthread_t tid1,tid2;
       pthread_create(&tid1,NULL,product,(void*)1);
       pthread_create(&tid2,NULL,product,(void*)2);

       pthread_t tid3,tid4,tid5;
       pthread_create(&tid3,NULL,consumer,(void*)3);
       pthread_create(&tid4,NULL,consumer,(void*)4);
       pthread_create(&tid5,NULL,consumer,(void*)5);

       pthread_join(tid1,NULL);
       pthread_join(tid2,NULL);
       
       pthread_join(tid3,NULL);
       pthread_join(tid4,NULL);
       pthread_join(tid5,NULL);

       pthread_cond_destroy(&need_product);
       pthread_mutex_destroy(&lock);

     return 0;
}

运行结果：

以上有2个生产者，3个消费者生产者生产出的数据放入同一链表中，消费者也都从该链表取数据，任何一刻对象对改链表进行操作时，别的对象都不能对该链表进行操作，实现了互斥功能。


返回编程语言教程...

新手上路

产品管理

支付方式

关于我们