线程同步和互斥（条件变量控制生产者消费者模型）

条件变量

生产者消费者模型：

关系：

同步

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

互斥

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

互斥

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

场所：

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

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个消费者生产者生产出的数据放入同一链表中，消费者也都从该链表取数据，任何一刻对象对改链表进行操作时，别的对象都不能对该链表进行操作，实现了互斥功能。

时间： 2024-11-08 20:25:17

线程同步和互斥（条件变量控制生产者消费者模型）的相关文章

Linux多线程实践(8) --Posix条件变量解决生产者消费者问题

Posix条件变量 int pthread_cond_init(pthread_cond_t *cond, pthread_condattr_t *cond_attr); int pthread_cond_destroy(pthread_cond_t *cond); int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex); int pthread_cond_timedwait(pthread_cond_t *cond

实现同步与互斥的一个实例--生产者消费模型

1.基础知识 1).生产者消费模型中存在3种关系: a.生产者与生产者之间是互斥的: b.消费者与消费者之间是互斥的: c.生产者与消费者之间是同步与互斥的: 2).生产者与消费者的场所,这儿我们选择单链表. 2.内容:多生产者生产一个结构体串在链表的表尾上,多消费者从表头取走结构体. 3.代码实现 1 #include<stdio.h> 2 #include<stdlib.h> 3 #include<pthread.h> 4 5 typedef

POSIX 使用互斥量和条件变量实现生产者/消费者问题

boost的mutex,condition_variable非常好用.但是在Linux上,boost实际上做的是对pthread_mutex_t 和pthread_cond_t的一系列的封装.因此通过对原生态的POSIX 的mutex,cond的生成者,消费者的实现,我们可以再次体会boost带给我们的便利. 1. 什么是互斥量互斥量从本质上说是一把锁,在访问共享资源前对互斥量进行加锁,在访问完成后释放互斥量上的锁.对互斥量进行加锁以后,任何其他试图再次对互斥量加锁的线程将会被阻塞直到当前线

线程同步与互斥之条件·变量

条件变量(condition variable) 线程间的同步与互斥技术,主要以互斥锁和条件变量为主,条件变量和互斥所的配合使用可以很好的处理对于条件等待的线程间的同步问题.举个例子:消费者和生产者问题. 消费者与生产者最基本的关系是服务与被服务的关系,但是在线程同步与互斥中强调的是两者访问资源的关系.首先生产与消费的关系为:同步与互斥,生产与生产的关系为:互斥,消费与消费的关系为:互斥.所以维护这三种关系的有两类人:生产者与消费者.并且生产数据与消费数据必须有场所. 所以将其简述为三种关系两类

LinuxC线程pthread线程同步进程同步-互斥量、信号量、条件变量、读写锁、文件锁

1. 同步概念 2. 线程同步 3. 进程同步 4. 生产者消费者模型 5. 哲学家问题原文地址:https://www.cnblogs.com/yongfengnice/p/12116954.html

linux多线程-互斥&条件变量与同步

多线程代码问题描述我们都知道,进程是操作系统对运行程序资源分配的基本单位,而线程是程序逻辑,调用的基本单位.在多线程的程序中,多个线程共享临界区资源,那么就会有问题: 比如 #include <pthread.h> #include <unistd.h> #include <stdio.h> #include <stdlib.h> int g_val = 10; void * test1(void* args) { g_val = 20; printf(&

linux线程间通信之条件变量和互斥量

一.条件变量定义有的时候仅仅依靠锁住共享资源来使用它是不够的.有时候共享资源只有某些状态的时候才能够使用.比方说,某个线程如果要从堆栈中读取数据,那么如果栈中没有数据就必须等待数据被压栈.这种情况下的同步使用互斥锁是不够的.另一种同步的方式--条件变量,就可以使用在这种情况下.条件变量(Condition Variable)是线程间的一种同步机制,提供给两个线程协同完成任务的一种方法,使用条件变量可以以原子方式阻塞线程,直到某个特定条件为真为止.条件变量的测试一般是用互斥量来保护的,用来确保每

线程的同步与互斥---生产者消费者模型

生产者与消费者模型生产者与消费者模型是一种描述进程间同步与互斥的一个方式,在这个模式下有两类人,一个是不停产生数据的生产者,一个是不停获取数据的消费者,为了效率最高,就必须保持两者之间的同步与互斥. 为了在生产者与消费者使用mutex保持互斥的前提下,posix版本下还有另外一个函数cond它的作用是在生产者生产出来一个数据时便提醒消费者,而消费者在没东西消费时可以使用cond将自己保持在一个等待生产者信号和关闭锁的状态,当收到信号时将锁打开然后消费. 这是cond的创建与销毁:

python并发编程之多进程(二)：互斥锁（同步锁）&进程其他属性&进程间通信（queue）&生产者消费者模型

一,互斥锁,同步锁进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的, 竞争带来的结果就是错乱,如何控制,就是加锁处理 part1:多个进程共享同一打印终端 #并发运行,效率高,但竞争同一打印终端,带来了打印错乱 from multiprocessing import Process import os,time def work(): print('%s is running' %os.getpid()) time.sleep(2) print('