【Linux】条件变量及生产者消费者模型

为什么要将这两者放在一起进行呢?
主要是因为生产消费与条件变量关系密切,正好相辅相成。

目录

  • 条件变量:
    • 条件变量的引出:
    • 条件变量的解释与接口:
    • 测试代码:
  • 生产者消费者模型:
    • 概念:
    • 代码实现Blocking Queue:
    • 完整代码:
    • 测试结果:
    • 一些问题:

条件变量:

条件变量的引出:

我们假设有一个自习室,这个自习室每次只能有一个人进入(使用挂在自习室门前钥匙),其他的人都要排队(排队之前都去试图找钥匙)。

今天小A起了个大早去拿钥匙,一直学到中午,此时饿得受不了了,于是想出去吃饭,他就走到门前打算归还钥匙,可是刚挂到门前就后悔了,因为她不想排队那么久才能继续学习,于是又拿着钥匙将门打开,但是又很饿,就这样循环往复,自己没有得到知识,外面的人也只能干排队等待。

与之对应:
自习室就相当于临界资源,钥匙就是锁,当其中一个线程拿到锁就可以访问临界资源,其他的线程都陷入阻塞状态。
由于线程A继续访问临界资源已经没有意义,但是竞争锁的能力很强,因为距离自己很近,造成了其他线程饥饿问题

虽然线程A这样做合法,但是不道德。
我们此时需要制定规则,保证过程具有一定顺序性,也就是同步。
而条件变量就是确保这个顺序性的!

条件变量的解释与接口:

我们知道了条件变量的作用,但是还是并不清楚条件变量是什么。

我们再来一个例子进行解释。

假设有一个游戏,一个蒙眼拿盘子中的苹果,一个睁眼将苹果放入盘子。

在这里插入图片描述

现在我们就可以对应一下了。
铃铛就是条件变量,他是一个队列,可以让线程进行等待
他的唤醒有两种策略,全部唤醒与单个唤醒。

我们也就可以对应一下条件变量的接口了。
有些接口与锁很类似。
在这里插入图片描述
其中init是条件变量初始化的函数,与锁一致。
signal与broadcast就是通知接口,signal是一次通知一个,boardcast就是全部通知
wait就是去铃铛下等待,也就是去条件变量下等待。
timedwait我们不管。
desory就是销毁条件。

关于这里还有一个细节,为什么条件变量要把锁业传入?
后面会进行解释。

测试代码:

我们目前可以浅浅的使用一下条件变量,熟悉一下接口。

我们现在要实现一个场景,在这里插入图片描述
主要逻辑是先创建一批线程并进行管理,每个创建好的线程都去执行各自的任务,任务是个死循环,每次进入在条件变量下等待被唤醒。

#include <iostream>
#include <pthread.h>
#include <unistd.h>const int N = 5;pthread_mutex_t g_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t g_cond = PTHREAD_COND_INITIALIZER;void *Run(void *args)
{while (true){pthread_mutex_lock(&g_mutex);pthread_cond_wait(&g_cond, &g_mutex);std::cout << "i am new thread-" << reinterpret_cast<long long>(args) << std::endl;pthread_mutex_unlock(&g_mutex);}
}int main()
{pthread_t thds[N];for (int i = 0; i < N; i++){pthread_create(&thds[i], nullptr, Run, reinterpret_cast<void*>(i));}// 唤醒while (true){sleep(1);pthread_cond_signal(&g_cond);}for (auto &val : thds){pthread_join(val, nullptr);}return 0;
}

现象:
在这里插入图片描述

生产者消费者模型:

概念:

条件变量暂时说到这里,我们先来谈一谈这个模型。

我们依旧是拿具体场景进行解释:
这是最简单的一个图示
在这里插入图片描述
实际上,我们的生产者代表的就是供货商,超市是一块缓存,顾客就是消费者。

这个模型有3大优点:

  1. 协调忙闲不均:
    当生产慢时而消费又多,我们就可以提前生产一批产品到超市;
    当生产快时而消费又慢,我们就可以提前让消费者来到超市;
  2. 效率高:
    你在买东西时,供应商在生产,达到了消费生产并发
    你在处理你买的东西时,供应商把商品放入超市,达到放入处理并发
  3. 解耦:
    生产者与消费者互不影响:比如一共有10个供货商,其中一个倒闭了,并不影响你买商品;当你吃泡面撒掉了,也不会影响供应商供货。

高度提炼一下可以简记为:1 2 3 原则。
1:一个交易场所(一段内存空间)
2:两种角色(生产角色,消费角色)
3:三种关系(生产与生产,消费与消费,生产与消费)
其中这三种关系分别为互斥,互斥,互斥&&同步。

相信前两个都很好理解,第三个呢?

比如供应商生产好商品还没录入数据就肯定不能被顾客拿走,这就是互斥。
这就像你在写数据,还没写完就被读走了一样,他们之间也是需要互斥的。
同样,如果超市没货了,我们要要依靠超市通知供应商,等有货了在反过来通知顾客。这就是同步。

代码实现Blocking Queue:

为什么写单对单?因为他简单!为什么不吃牛,因为他善!在这里插入图片描述
不过为什么单对单简单,因为我们的1 2 3原则中有三个关系,生产与生产,消费与消费,生产与消费,单对单就不需要考虑前两种,自然而言的简单了许多。
这里我们要解决一个历史问题:为什么wait要传入锁?

在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。
其与普通的队列区别在于,当队列为空时,从队列获取元素的操作将会被阻塞,直到队列中被放入了元素;
当队列满时,往队列里存放元素的操作也会被阻塞,直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的,线程在对阻塞队列进程操作时会被阻塞)
在这里插入图片描述

那么我们现在就要有一个大概的轮廓:

生产者一直写,消费者一直读即可,很基础的框架,有了框架我们就知道如何下手实现主要功能了。

#include "BlockingQueue.hpp"void *Consumer(void *args)
{BlockingQueue *bq = static_cast<BlockingQueue*>(args);while (true){// 接收数据bp.Pop();// 处理数据}
}void *Producer(void *args)
{BlockingQueue *bq = static_cast<BlockingQueue*>(args);while (true){// 构建数据bp.Push();}
}int main()
{BlockingQueue pc;pthread_t t1, t2;pthread_create(&t1, nullptr, Producer, static_cast<void*>(&pc));pthread_create(&t2, nullptr, Consumer, static_cast<void*>(&pc));pthread_join(t1, nullptr);pthread_join(t2, nullptr);return 0;
}

这是我们的阻塞队列,里面有对于一些点的详细注释,
比如cond为什么要传入锁?唤醒在哪个位置进行唤醒?
而又因为我们也不确定用户使用的是什么类型,设计为模板即可。

#pragma once#include <iostream>
#include <pthread.h>
#include <queue>
#include <unistd.h>template <class T>
class BlockingQueue
{
public:BlockingQueue(int cap = 5) : _cap(cap){pthread_mutex_init(&_mutex, nullptr);pthread_cond_init(&_cond_p, nullptr);pthread_cond_init(&_cond_c, nullptr);}void Push(const T &val){pthread_mutex_lock(&_mutex);if (IsFull()){pthread_cond_wait(&_cond_p, &_mutex);// 所以这里的wait为什么要传锁就一目了然了// 因为如果你带着锁去wait,那么别的线程只会一直阻塞,不会拿到锁// 所以wait还要进行解锁,等被唤醒在继续抢锁。}// 出来了就代表肯定不是满的,此时push数据即可。_q.push(val);pthread_mutex_unlock(&_mutex);pthread_cond_signal(&_cond_c);// 注意这里的唤醒放在解锁的上或下都可以,当放在下时,我们还是以单对单为例// 假设我们此时队列中还没有数据,消费者在wait中解锁阻塞,生产者生产完数据进行解锁再唤醒消费者,// 而这时消费者会和生产者再次抢锁,因为消费者会从wait中被唤醒需要抢锁,而生产者在lock中抢锁。// 如果消费者抢到了就罢了,直接进行写入数据;但是如果生产者抢到了,那就继续生产数据,// 我们假设生产者一直将队列写满,那么他就会在wait中进行阻塞,让消费者写入数据在解锁唤醒。// 同理,实际上因为锁的钳制,在解锁前或后唤醒都是可以的}void Pop(T *val){pthread_mutex_lock(&_mutex);if (IsEmpty()){pthread_cond_wait(&_cond_c, &_mutex);}// 出来了就代表此时数据肯定不为空,可以写给消费者了。*val = _q.front();_q.pop();pthread_mutex_unlock(&_mutex);pthread_cond_signal(&_cond_p);}~BlockingQueue(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_cond_p);pthread_cond_destroy(&_cond_c);}private:bool IsFull(){return _cap == _q.size();}bool IsEmpty(){return _q.empty();}private:int _cap;std::queue<T> _q;pthread_mutex_t _mutex;pthread_cond_t _cond_p;pthread_cond_t _cond_c;
};

但是这里还有一个细节,我们判断空或满时使用了if,这在某些情况下是会出问题的。

比如当前队列中没有数据。
有一个生产者,两个消费者。
当生产者刚刚生产完,进行了broadcast唤醒,导致两个消费者线程都被唤醒我们假设生产者此时不会抢到锁,所以目前只会有一个消费者线程抢到锁,当其中一个A抢到后,进行写入数据并pop,这都是没有问题的,但是如果A解锁后被另一个消费者B抢到,那么B又会继续写入并进行pop。
问题就出现了,因为唯一的一个数据已经被A拿走了,此时就会出现err,所以我们要对if进行一下修改,改为while,这样就避免了因为B抢到锁而继续pop造成的err。

完整代码:


```cpp
#pragma once#include <iostream>
#include <pthread.h>
#include <queue>
#include <unistd.h>template <class T>
class BlockingQueue
{
public:BlockingQueue(int cap = 5) : _cap(cap){pthread_mutex_init(&_mutex, nullptr);pthread_cond_init(&_cond_p, nullptr);pthread_cond_init(&_cond_c, nullptr);}void Push(const T &val){pthread_mutex_lock(&_mutex);while (IsFull()){pthread_cond_wait(&_cond_p, &_mutex);// 所以这里的wait为什么要传锁就一目了然了// 因为如果你带着锁去wait,那么别的线程只会一直阻塞,不会拿到锁// 所以wait还要进行解锁,等被唤醒在继续抢锁。}// 出来了就代表肯定不是满的,此时push数据即可。_q.push(val);pthread_mutex_unlock(&_mutex);pthread_cond_signal(&_cond_c);// 注意这里的唤醒放在解锁的上或下都可以,当放在下时,我们还是以单对单为例// 假设我们此时队列中还没有数据,消费者在wait中解锁阻塞,生产者生产完数据进行解锁再唤醒消费者,// 而这时消费者会和生产者再次抢锁,因为消费者会从wait中被唤醒需要抢锁,而生产者在lock中抢锁。// 如果消费者抢到了就罢了,直接进行写入数据;但是如果生产者抢到了,那就继续生产数据,// 我们假设生产者一直将队列写满,那么他就会在wait中进行阻塞,让消费者写入数据在解锁唤醒。// 同理,实际上因为锁的钳制,在解锁前或后唤醒都是可以的}void Pop(T *val){pthread_mutex_lock(&_mutex);while (IsEmpty()){pthread_cond_wait(&_cond_c, &_mutex);}// 出来了就代表此时数据肯定不为空,可以写给消费者了。*val = _q.front();_q.pop();pthread_mutex_unlock(&_mutex);pthread_cond_signal(&_cond_p);}~BlockingQueue(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_cond_p);pthread_cond_destroy(&_cond_c);}private:bool IsFull(){return _cap == _q.size();}bool IsEmpty(){return _q.empty();}private:int _cap;std::queue<T> _q;pthread_mutex_t _mutex;pthread_cond_t _cond_p;pthread_cond_t _cond_c;
};

main函数代码:

#include "BlockingQueue.hpp"#include <random>void *Consumer(void *args)
{BlockingQueue<int> *bq = static_cast<BlockingQueue<int>*>(args);while (true){sleep(2);// 接收数据int val;bq->Pop(&val);// 处理数据std::cout << "Consumer receive a data: " << val << std::endl;}
}void *Producer(void *args)
{BlockingQueue<int> *bq = static_cast<BlockingQueue<int>*>(args);while (true){// 构建数据int val = rand() % 10 + 1;// [1, 10]bq->Push(val);std::cout << "Producer produce a data: " << val << std::endl;}
}int main()
{srand(time(nullptr));BlockingQueue<int> pc;pthread_t t1, t2;pthread_create(&t1, nullptr, Producer, static_cast<void*>(&pc));pthread_create(&t2, nullptr, Consumer, static_cast<void*>(&pc));pthread_join(t1, nullptr);pthread_join(t2, nullptr);return 0;
}

测试结果:

在这里插入图片描述
观察到生产者生产顺序为1 9 6 6 6…而消费者也正好是1 9 6 6 6…

注意:我们的生产者消费者不仅仅只能传递最简单的内置类型,也可以进行传递自定义类型!可调用对象也是可以的!

一些问题:

我们的多生产多消费不需要修改代码,因为锁已经帮我们搞定了生产与生产,消费与消费之间的关系。

为什么明明一次只能有一个线程访问队列,但还是说生产消费模型效率高?
因为我们不能只关注生产与消费在缓存的时间!
我们的生产任务或数据需要时间,处理任务或数据需要时间。
也就是说:当其中一个生产者在放任务,其他的生产者在生产任务;
其中一个消费者在拿任务,其他消费者在处理任务。
这种并发才让我们的效率变高!

为什么要在锁之后再进行条件变量wait?
这是因为我们push或pop之前一定会临街资源,临界资源一定是被锁保护起来的,所以设计者才会这样设计接口,我们也才要这样使用。

完~

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/bicheng/49533.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

opengaussdb在oepnEuler上安装

安装前提&#xff1a; 软件环境&#xff1a;openEuler 20.03LTS 个人开发者最低配置2核4G&#xff0c;推荐配置4核8G 数据库版本&#xff1a;openGauss-5.0.2-openEuler-64bit-all.tar.gz 数据库下载地址&#xff1a; https://docs-opengauss.osinfra.cn/zh/docs/5.0.0/docs/In…

OpenSNN推文:百度沈抖:深度拥抱人工智能+,加速发展新质生产力,共创智能时代新未来

在中国联通合作伙伴大会上&#xff0c;百度集团执行副总裁、百度智能云事业群总裁沈抖发表了一场题为“深度拥抱人工智能&#xff0c;加快发展新质生产力”的精彩演讲&#xff0c;深入探讨了大模型技术在当前科技浪潮中的核心地位及其对企业生产力的深远影响。 沈抖指出&#…

【LeetCode】86.分割链表

1. 题目 2. 分析 这题没有太大难度&#xff0c;主要是熟悉代码。 3. 代码 # Definition for singly-linked list. # class ListNode: # def __init__(self, val0, nextNone): # self.val val # self.next next class Solution:def partition(self, he…

Linux系统编程-文件系统

目录 什么是Linux文件系统 文件系统的职责 存储介质抽象 inode&#xff1a;文件系统的核心 文件分配策略 目录结构 文件系统布局 日志和恢复机制 目录权限 粘滞位(t位)&#xff1a; 硬链接和符号链接 硬链接的特点&#xff1a; 创建硬链接&#xff1a; 符号链接的…

MySQL补充性文件

数据库专属单词 authentication #身份验证 delimiter #分隔符 character #字符集 collate #整理。 指定字符集的排序规则 unicode #统一码 flush #刷新 privileges #特权 string #串 set #设置 use #使用 zerofill #修饰符。0可以填补输出的值 unsigned #修饰符。无符…

德国云手机:企业移动办公解决方案

在现代商业环境中&#xff0c;移动办公已经成为一种趋势。德国云手机作为一种高效的解决方案&#xff0c;为企业提供了强大的支持。本文将探讨德国云手机如何优化企业的移动办公环境。 一、德国云手机的主要优势 高灵活性 德国云手机具有高度的灵活性&#xff0c;能够根据用户需…

Elasticsearch:Golang ECS 日志记录 - Logrus

ECS 记录器是你最喜欢的日志库的格式化程序/编码器插件。它们可让你轻松地将日志格式化为与 ECS 兼容的 JSON。 编码器以 JSON 格式记录&#xff0c;内部依赖于默认的 logrus.JSONFormatter。它还处理 ECS 错误格式的错误字段记录。 默认情况下&#xff0c;会添加以下字段&am…

【学习笔记】无人机系统(UAS)的连接、识别和跟踪(三)-架构模型和概念

引言 3GPP TS 23.256 技术规范&#xff0c;主要定义了3GPP系统对无人机&#xff08;UAV&#xff09;的连接性、身份识别、跟踪及A2X&#xff08;Aircraft-to-Everything&#xff09;服务的支持。 3GPP TS 23.256 技术规范&#xff1a; 【免费】3GPPTS23.256技术报告-无人机系…

js-toLocaleString()方法的使用(根据本地规则将对象转换为字符串-千分位或百分比等)

1.应用场景 toLocaleString() 方法是 JavaScript 中的一个内置方法&#xff0c;它可以根据本地规则将对象转换为字符串。但主要被用于 Date、Number 和 Array 对象上。 2.具体应用 2.1 date对象中使用 Date 对象上调用 toLocaleString() 方法时&#xff0c;它会根据运行代码的…

基于 Electron+Vite+Vue3+Sass 框架搭建

技术参考 技术描述Electron一个使用 JavaScript、HTML 和 CSS 构建桌面应用程序的框架。嵌入 Chromium 和 Node.jsElectron Forge用于打包和分发 Electron 应用程序的一体化工具。英文地址在此Vite前端构建工具Vue3用于构建用户界面的 JavaScript 框架vitejs/plugin-vueVite 插…

PlantUML 语法、图标和示例

基本语法 关键字 声明参与者的几个关键字 actor、boundary、control、entity、database、collections、participant 箭头样式 我们可以通过&#xff0c;修改箭头样式&#xff0c;来表达不一样的意思&#xff1a; 表示一条丢失的消息&#xff1a;末尾加 x让箭头只有上半部…

Redis在SpringBoot中遇到的问题:预热,雪崩,击穿,穿透

缓存预热 预热即在产品上线前&#xff0c;先对产品进行访问或者对产品的Redis中存储数据。 原因&#xff1a; 1. 请求数量较高 2. 主从之间数据吞吐量较大&#xff0c;数据同步操作频度较高,因为刚刚启动时&#xff0c;缓存中没有任何数据 解决方法&#xff1a; 1. 使用脚…

Kafka - 生产者

生产者消息对象 public class ProducerRecord<K, V> {private final String topic; // 主题private final Integer partition; //分区号private final Headers headers; //消息头部private final K key; //键private final V value; //值private final Long timestamp; …

opencv 按键开启连续截图,并加载提示图片

背景图小图 键盘监听使用的是pynput 库 保存图片时使用了年月日时分秒命名 原图&#xff1a; from pynput import keyboard import cv2 import time# 键盘监听 def on_press(key):global jieglobal guanif key.char a:jie Trueelif key.char d:jie Falseelif key.char…

配置三个Spring Boot应用并通过Nginx进行反向代理 讨论中

要配置三个Spring Boot应用并通过Nginx进行反向代理&#xff0c;你可以按照以下步骤操作&#xff1a; 步骤 1: 准备Spring Boot应用 确保每个Spring Boot应用都有不同的端口号。例如&#xff0c;你可以设置第一个应用监听8080端口&#xff0c;第二个监听8081端口&#xff0c;…

FPGA JTAG最小系统 EP2C5T144C8N

FPGA的文档没有相应的基础还真不容易看懂&#xff0c;下面是B站上对FPGA文档的解读(本文非对文档解读&#xff0c;只是为个人记录第三期&#xff1a;CycloneIV E最小系统板设计&#xff08;一&#xff09;从Datasheet上获取FPGA的基本参数_哔哩哔哩_bilibili 电源部份 核心电…

TS config

moduleResolution 是 TypeScript 编译器中的一个选项&#xff0c;用于控制如何解析模块导入。这个选项影响着 TypeScript 如何查找和解析 import 和 export 声明中指定的模块。 {"compilerOptions": {"moduleResolution": "Node"//小写也没问题…

SELinux的 getenforce setenforce 配置文件/etc/selinux/config的 SELINUX和SELINUXTYPE

SELinux&#xff08;Security-Enhanced Linux&#xff09;是一个为Linux系统提供访问控制安全策略的安全模块。它是Linux内核的一个功能强大的安全子系统&#xff0c;旨在提供访问控制安全策略机制&#xff0c;以限制程序中特定代码段的权限。SELinux超越了传统的UNIX权限模型&…

AI学习记录 -使用react开发一个网页,对接chatgpt接口,附带一些英语的学习prompt

实现了如下功能&#xff08;使用react实现&#xff0c;原创&#xff09; 实现功能&#xff1a; 1、对接gpt35模型问答&#xff0c;并实现了流式传输&#xff08;在java端&#xff09; 2、在实际使用中&#xff0c;我们的问答历史会经常分享给他人&#xff0c;所以下图的 copy …

Python酷库之旅-第三方库Pandas(042)

目录 一、用法精讲 141、pandas.Series.agg(regate)方法 141-1、语法 141-2、参数 141-3、功能 141-4、返回值 141-5、说明 141-6、用法 141-6-1、数据准备 141-6-2、代码示例 141-6-3、结果输出 142、pandas.Series.transform方法 142-1、语法 142-2、参数 142…