系统相关
首页 > 系统相关> > Linux系统编程---17(条件变量及其函数,生产者消费者条件变量模型,生产者与消费者模型(线程安全队列),条件变量优点,信号量及其主要函数,信号量与条件变量的区别,)

Linux系统编程---17(条件变量及其函数,生产者消费者条件变量模型,生产者与消费者模型(线程安全队列),条件变量优点,信号量及其主要函数,信号量与条件变量的区别,)

作者:互联网

条件变量

条件变量本身不是锁!但它也可以造成线程阻塞。通常与互斥锁配合使用。给多线程提供一个会合的场所。

主要应用函数:

  1. pthread_cond_init 函数
  2. pthread_cond_destroy 函数
  3. pthread_cond_wait 函数
  4. pthread_cond_timedwait 函数
  5. pthread_cond_signal 函数
  6. pthread_cond_broadcast 函数
  7. 以上 6 个函数的返回值都是:成功返回 0, 失败直接返回错误号。
  8. pthread_cond_t 类型 用于定义条件变量
  9. pthread_cond_tcond;

pthread_cond_init 函数

初始化一个条件变量
int pthread_cond_init(pthread_cond_t *restrictcond,const pthread_condattr_t *restrictattr);
参 2:attr 表条件变量属性,通常为默认值,传 NULL 即可
也可以使用静态初始化的方法,初始化条件变量:

pthread_cond_t	cond=PTHREAD_COND_INITIALIZER;

pthread_cond_destroy 函数

销毁一个条件变量

int	pthread_cond_destroy(pthread_cond_t		*cond);

pthread_cond_wait 函数

阻塞等待一个条件变量

int	pthread_cond_wait(pthread_cond_t	*restrictcond,pthread_mutex_t	*restrictmutex);
函数作用:
  1. 阻塞等待条件变量 cond(参 1)满足
  2. 释放已掌握的互斥锁(解锁互斥量)相当于 pthread_mutex_unlock(&mutex);
  3. 1.2.两步为一个原子操作。
  4. 当被唤醒,pthread_cond_wait 函数返回时,解除阻塞并重新申请获取互斥锁 pthread_mutex_lock(&mutex);

pthread_cond_timedwait 函数

限时等待一个条件变量

int pthread_cond_timedwait(pthread_cond_t	 *restrict cond, pthread_mutex_t *restrict mutex, const struct timespec *restrictabstime);

参 3: 参看 mansem_timedwait 函数,查看 struct timespec 结构体。

struct	timespec{
 time_t	tv_sec; /*seconds*/ 秒
  long tv_nsec; /*nanosecondes*/ 纳秒 
  } 

形参 abstime:绝对时间。
如:time(NULL)返回的就是绝对时间。
而 alarm(1)是相对时间,相对当前时间定时 1 秒钟。

struct	timespect={1,0}; 
pthread_cond_timedwait(&cond,&mutex,&t); 只能定时到 1970 年 1 月 1 日 00:00:01 秒(早已经过去) 

正确用法:

time_tcur=time(NULL); 获取当前时间。
 structtimespect; 定义 timespec 结构体变量 t
  t.tv_sec=cur+1; 定时 1 秒
   pthread_cond_timedwait(&cond,&mutex,&t); 传参 

setitimer 函数还有另外一种时间类型:

 struct	timeval
 {
  time_t tv_sec; /*seconds*/ 秒 
  suseconds_ttv_usec; /*microseconds*/ 微秒
   };

pthread_cond_signal 函数

唤醒至少一个阻塞在条件变量上的线程

int	pthread_cond_signal(pthread_cond_t*cond);

pthread_cond_broadcast 函数

唤醒全部阻塞在条件变量上的线程

int	pthread_cond_broadcast(pthread_cond_t*cond);

生产者消费者条件变量模型

线程同步典型的案例即为生产者消费者模型,而借助条件变量来实现这一模型,是比较常见的一种方法。假定 有两个线程,一个模拟生产者行为,一个模拟消费者行为。两个线程同时操作一个共享资源(一般称之为汇聚), 生产向其中添加产品,消费者从中消费掉产品。

在这里插入图片描述

/*借助条件变量模拟  生产者--消费者问题*/
#include<stdio.h>
#include<unistd.h>
#include<pthread.h>
#include<stdio.h>
#include<string.h>
/*链表作为共享数据,需要被互斥量保护*/
struct msg{
    struct msg *next;
    int num;
};

struct msg *head;
struct msg *mp;

/*静态初始化  一个条件变量 和一个互斥量*/
pthread_cond_t has_product = PTHREAD_COND_INITIALIZER; 
pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;

void *consumer(void *p)
{
    for(;;){
        pthread_mutex_lock(&lock);  //头指针为空,说明没有结点 
        while(head == NULL){
            pthread_cond_wait(&has_product,&lock); //判断条件变量是否满足
        }
        mp = head;
        head = mp->next;        //模拟消费掉一个产品
        pthread_mutex_unlock(&lock);

        printf("---Consume ---%d\n",mp->num);
        free(mp);
        sleep(rand() %  5);
    }
}
void *producer()
{
    for(;;){
        mp = malloc(sizeof(struct msg));
        mp->num = rand() % 1000 + 1;    //模拟生产一个产品
        printf("--Produce ---%d\n",mp->num);

        pthread_mutex_lock(&lock);
        mp->next=head;              //头插法
        head = mp;
        pthread_mutex_unlock(&lock); //释放

        pthread_cond_signal(&has_product); //将等待在该条件变量上的一个线程>唤醒
        sleep(rand() % 5);
    }
}

int main(int argc,char *argv[])
{
    pthread_t pid,cid;  //pid生产者ID  cid消费者ID
    srand(time(NULL));

    pthread_create(&pid,NULL,producer,NULL); //生产者
    pthread_create(&cid,NULL,consumer,NULL); //消费者

    pthread_join(pid,NULL);
    pthread_join(cid,NULL);

    return 0;
}   

在这里插入图片描述

条件变量是搭配互斥锁一起使用的
  1. 因为条件变量实现同步只提供等待与唤醒功能,并没有提供条件判断的功能,因此条件判断需要用户实现,但是条件的操作是一个临界资源的操作,因此需要受保护,需要在条件判断之前加锁
  2. 如果加锁成功后,因为条件不满足而陷入休眠,就会导致卡死(因为另一方因为无法获取锁,而导致无法促使条件满足),因此需要在休眠之前解锁;并且解锁与休眠必须是原子操作
  3. 被唤醒之后,即将对临界资源进行操作,但是被操作前还要进行保护加锁
  4. 所以pthread_cond_wait集合了三步原子操作:解锁–>等待–>被唤醒后加锁

条件变量的优点

  1. 相较于 mutex 而言,条件变量可以减少竞争。
  2. 如直接使用 mutex,除了生产者、消费者之间要竞争互斥量以外,消费者之间也需要竞争互斥量,但如果汇聚 (链表)中没有数据,消费者之间竞争互斥锁是无意义的。有了条件变量机制以后,只有生产者完成生产,才会引 起消费者之间的竞争。提高了程序效率。

生产者与消费者模型(线程安全队列)

一个场所,两种角色,三种关系

功能:

  1. 解耦和(两个关系之间紧密)
  2. 支持忙闲不均
  3. 支持并发

三者关系

生产者–生产者:互斥
消费者–消费者:互斥
生产者–消费者:同步+互斥

/*生产者与消费者模型队列实现                                                                                                                                                                                     
 * 1.实现线程安全的队列,对外提供线程安全的数据入队和出队操作
 * 2.创建线程,分别作为生产者与消费者数据入队或数据出队
 */
 
#include<iostream>
#include<queue>
#include<pthread.h>

#define MAX_QUEUE 10
class BlockQueue
{
    public:
        BlockQueue(int cap = MAX_QUEUE):_capacity(cap){
            //初始化队列
            pthread_mutex_init(&_mutex,NULL);
            pthread_cond_init(&_cond_con,NULL);
            pthread_cond_init(&_cond_pro,NULL);
        }   
        ~BlockQueue(){
            pthread_mutex_destroy(&_mutex);
            pthread_cond_destroy(&_cond_con);
            pthread_cond_destroy(&_cond_pro);
        }   
        //入队
        void QueuePush(int data){
            QueueLock();
            while(QueueIsFull()){ //队列满了
                ProWait();    //生产者等待
            }   
            _queue.push(data);
            ConWakeUp();
            QueueUnLock();
        }   
        void QueuePop(int *data){
            QueueLock();
            while(QueueIsEmpty()){
                ConWait();
            }   
            *data = _queue.front();//获取队列头结点
            _queue.pop();//结点出队
            ProWakeUp();
            QueueUnLock();
        }


    private:
        //队列加锁
        void QueueLock(){
            pthread_mutex_lock(&_mutex);
        }
        //队列解锁
        void QueueUnLock(){
            pthread_mutex_unlock(&_mutex);
        }
        //消费者等待
        void ConWait(){
            pthread_cond_wait(&_cond_con,&_mutex);
        }
        //消费者唤醒
        void ConWakeUp(){
            pthread_cond_signal(&_cond_con);
        }
        //生产者等待
        void ProWait(){
            pthread_cond_wait(&_cond_pro,&_mutex);
        }
        //生产者唤醒
        void ProWakeUp(){
            pthread_cond_signal(&_cond_pro);
        }
        //判断队列是否为空
        bool QueueIsFull(){
            return (_capacity == _queue.size());
        }
        //队列是否是满的
        bool QueueIsEmpty(){
            return _queue.empty();
        }
    private:
        std::queue<int>_queue;//创建队列
        int _capacity;//队列结点最大数量                          
        //线程安全实现成员
        pthread_mutex_t _mutex;
        pthread_cond_t _cond_pro;
        pthread_cond_t _cond_con;
};


void *thr_consumer(void *arg){
    BlockQueue *q = (BlockQueue *)arg;
    while(1){
        int data;
        q->QueuePop(&data);
        std::cout<<"consumer"<<pthread_self() <<" get data:"<< data <<std::endl;
    }
    return NULL;
}

int i = 0; //必须受保护
pthread_mutex_t mutex;

void *thr_productor(void *arg){
    BlockQueue *q = (BlockQueue *)arg;
    while(1){
        pthread_mutex_lock(&mutex);
        q->QueuePush(i++);
        pthread_mutex_unlock(&mutex);
        std::cout<<"productor:" <<pthread_self() <<"put data:"<< i <<std::endl;
    }
    return NULL;
}


int main(int argc,char *argv[])
{
    BlockQueue q;

    pthread_t ctid[4],ptid[4];
    int i,ret;

    pthread_mutex_init(&mutex,NULL);
    for(i = 0;i < 4; i++){        
        ret = pthread_create(&ctid[i],NULL,thr_consumer,(void *)&q);
        if(ret != 0){
            std::cout<<"pthread create error\n";
            return -1;
        }
    }
    for(i = 0;i < 4; i++){
        ret = pthread_create(&ptid[i],NULL,thr_productor,(void *)&q);
        if(ret != 0){
            std::cout<<"pthread create error\n";
            return -1;
        }
    }
    for(i = 0;i < 4; i++){
        pthread_join(ctid[i],NULL);
    } 
    for(i = 0; i < 4;i++){
        pthread_join(ptid[i],NULL);
    }

    return 0;
}

在这里插入图片描述

信号量

进化版的互斥锁(1–>N)
由于互斥锁的粒度比较大,如果我们希望在多个线程间对某一对象的部分数据进行共享,使用互斥锁是没有办 法实现的,只能将整个数据对象锁住。这样虽然达到了多线程操作共享数据时保证数据正确性的目的,却无形中导 致线程的并发性下降。线程从并行执行,变成了串行执行。与直接使用单进程无异。
信号量,是相对折中的一种处理方式,既能保证同步,数据不混乱,又能提高线程并发。

计数器+等待队列+等待与唤醒功能
  1. 通过自身的计数器实现条件判断,当前条件满足时则直接返回并且计数-1.当条件并不满足时则阻塞
  2. 当产生资源后,通过信号量的唤醒功能唤醒等待并且计数+1

信号量和条件变量实现同步的区别

  1. 信号量的条件判断由自身来完成,而条件变量的条件判断由用户完成
  2. 信号量并不搭配互斥锁使用,而条件变量需要搭配互斥锁一起使用保护条件的改变

sem_init 函数

初始化一个信号量
int sem_init(sem_t *sem,int pshared,unsigned int value);
参 1:sem 信号量
参 2:pshared 取 0 用于线程间;取非 0(一般为 1)用于进程间
参 3:value 指定信号量初值

sem_destroy 函数

销毁一个信号量

int	sem_destroy(sem_t	*sem);

sem_wait 函数

给信号量加锁 ,对计数进行判断,计数<=0则阻塞;否则立即返回流程继续,计数-1

int	sem_wait(sem_t	*sem);

sem_post 函数

给信号量解锁 ,对计数进行+1,并且唤醒等到的线程

 int	sem_post(sem_t	*sem);

sem_trywait 函数

尝试对信号量加锁
(与 sem_wait 的区别类比 lock 和 trylock)

 int	sem_trywait(sem_t	*sem);

sem_timedwait 函数

限时尝试对信号量加锁

int	sem_timedwait(sem_t	*sem,const	struct	timespec	*abs_timeout); 

参 2:abs_timeout 采用的是绝对时间。
定时 1 秒:

time_tcur=time(NULL); 获取当前时间。 
structtimespect; 定义 timespec 结构体变量 t 
t.tv_sec=cur+1; 定时 1 秒 
t.tv_nsec=t.tv_sec+100; 
sem_timedwait(&sem,&t); 传参

使用信号量实现生产者与消费者模型

在这里插入图片描述

/*使用信号量实现生产者与消费者模型
 *
 */

#include<iostream>
#include<queue>
#include<pthread.h>
#include<semaphore.h>

class RingQueue
{
    public:
        RingQueue(int cap = 10):_capacity(cap),_queue(cap){
            //1.信号量变量
            //2.参数取值    0:用于线程间同步与互斥
            //              非0:用于进程间同步与互斥
            //3.信号量初值
            sem_init(&_sem_lock,0,1);//互斥锁初始值只给1
            sem_init(&_sem_data,0,0);//初始数据资源数据为0
            sem_init(&_sem_space,0,cap);//初始空闲空间计数
        }   
        ~RingQueue(){
            sem_destroy(&_sem_lock);
            sem_destroy(&_sem_data);
            sem_destroy(&_sem_space);
        }   
        void QueuePush(int data){

            // ProWait();//空闲空间计数判断是否有空闲空间,若有返回,否则等待
            // 因为已经通过_sem_space的空闲空间计数知道是否有空闲空间
            sem_wait(&_sem_space);//添加数据之后,空闲空间计数-1

            sem_wait(&_sem_lock);//锁计数初始为1,一旦进入-1加锁
            _queue[_step_write]=data;                                                                                                                                                                            
            _step_write = ( _step_write + 1) % _capacity;
            sem_post(&_sem_lock);//数据添加完毕后解锁,数据资源计数+1

            sem_post(&_sem_data);//数据添加完毕后,数据资源计数+1
            //ConWakeUp();

        }   
    void QueuePop(int *data){

            sem_wait(&_sem_data);//取数据的时候,数据资源计数-1

            sem_wait(&_sem_lock);//锁最好仅仅保护临界区

            *data = _queue[_step_read];
            _step_read = (_step_read + 1) % _capacity;
            sem_post(&_sem_lock);

            sem_post(&_sem_space);//取数据之后,空闲空间计数+!

        }
    private:
        std::vector<int>_queue;
        int _capacity; //队列最大数量
        int _step_write;//当前写到哪里的下标
        int _step_read;//当前读到哪里了的下标

        sem_t _sem_lock;//实现互斥锁
        sem_t _sem_space;//空闲空间计数
        sem_t _sem_data;//数据资源计数
        
/*
        //队列加锁
        void QueueLock(){
            pthread_mutex_lock(&_mutex);
        }
        //队列解锁
        void QueueUnLock(){
            pthread_mutex_unlock(&_mutex);
        }
         
  */  
};
void *thr_productor(void *arg){                                                                                                                                                                                  
    RingQueue *q = (RingQueue*)arg;
    int i=0;
    while(1){
        q->QueuePush(i);
        std::cout<<"thread:"<<pthread_self()<<"put data"<<i++<<"\n";
    }
    return NULL;
}

void *thr_consumer(void *arg){
    RingQueue *q = (RingQueue*)arg;
    while(1){
        int data;
        q->QueuePop(&data);
        std::cout<<"thread:"<<pthread_self()<<"get data"<<data<<"\n";
    }
    return NULL;
}

int main(int argc,char *argv[])
{
    RingQueue q;
    pthread_t ptid,ctid[4];
    int i ,ret;

    ret = pthread_create(&ptid,NULL,thr_productor,(void *)&q);
    if(ret != 0){
        std::cout<<"thread create error\n";
        return -1;
    }

    for(i = 0;i < 4;i++){
        ret = pthread_create(&ctid[i],NULL,thr_consumer,(void *)&q);
        if(ret != 0){
            std::cout<<"thread create error\n";
            return -1;
        }
    }

    for(i = 0; i < 4; i++){
        pthread_join(ctid[i],NULL);
    }

    pthread_join(ptid,NULL);

    return 0;
}

标签:变量,int,条件,&_,信号量,cond,pthread,mutex,sem
来源: https://blog.csdn.net/liuyuchen282828/article/details/100581699