本文收录于「流浪」的系列专栏Linux系统⚙️C数据结构与算法PythonLangChain LangGraph️MySQL 数据库Git 工具计算机网络LLM大厂面试、八股学习筑基专栏 博客主页流浪 原创首发于 CSDN前言线程十把条件变量讲到了底唤醒机制加等待队列wait 的两段阻塞也拆完了。本篇把互斥锁和条件变量组合成完整的多线程协作模型——生产者消费者。先摆清两种角色和一个交易场所再逐一推演三种关系为什么存在最后回答一个最容易想当然的问题进出交易场所明明是串行的高效到底从哪来。一、两种角色和一个交易场所1.1 两种角色由线程承担1. 角色怎么分一种角色负责生产数据把数据放进交易场所叫生产者一种角色负责从交易场所取出数据、处理数据叫消费者角色描述的是执行流干的事落到实现里就是由线程来承担2. 角色不绑定线程个数一个线程扮演一个角色是最常见的写法一个生产线程配一个消费线程也可以开多个生产者线程、多个消费者线程角色仍然只有两种角色按干什么划分不按线程数量划分线程数只影响并发度1.2 交易场所是一块特定结构的内存1. 它长什么样交易场所不是一句空话它是以特定结构构成的一种内存空间最常见的结构是阻塞队列和环形缓冲区容量有限、先进先出本篇以阻塞队列为例打个比方就是超市生产者把商品摆上货架消费者从货架上取走超市就是那个交易场所2. 它同时是临界资源所有的写入和读取都落在这一块空间上多方共享三种关系全部围绕它产生它是整个模型的交汇点整个模型就三样东西生产者、消费者、交易场所后面所有的关系和规则都是从这三样东西里长出来的。二、三种关系逐一推演为什么会存在2.1 生产者和生产者之间竞争互斥1. 竞争从哪来多个生产者往同一个交易场所放数据而场所只有一份容量有限谁先放谁占位置这就是竞争关系2. 为什么必须互斥放数据不只是放进一个格还要改场所的内部结构队尾指针、计数都会跟着动两个生产者同时改这些数据执行交错就会把结构写坏改指针和改计数都不是一步完成中间可以被切走切回来后写的是自己看到的旧值对方的改动被覆盖这正是线程八里 ticket-- 三步被切走的非原子问题场景换了病根没换所以生产者之间必须互斥一次只允许一个生产者动场所竞争是抢位置互斥是抢到了就独占完成两个词说的是同一件事的两面。2.2 消费者和消费者之间互斥竞争1. 为什么是互斥关系一份数据只能被一个消费者拿走被拿两次就是重复消费取数据同样要改队列结构并发地取一样会把结构写坏2. 竞争从哪来多个消费者去同一个交易场所取数据而场所只有一份容量有限谁先取谁占位置这就是竞争关系2. 这条关系什么时候显现只有一个消费者时这条关系看不见对手盘不存在消费者一多它和生产者之间那条一样是硬约束躲不掉单线程时看不出的问题不代表约束不存在只代表还没人跟它抢。2.3 生产者和消费者之间互斥加同步1. 互斥的那一半生产者写、消费者读读写的是同一块空间同时读写读到的可能是改到一半的半成品所以必须互斥2. 同步的两个方向场所满了生产者必须停手等待消费者先消费腾出位置再把生产者唤醒场所空了消费者必须等待生产者先生产有货了再把消费者唤醒两个方向合起来就是让该等的等、该走的走访问有节奏3. 没有同步会怎样满了还塞没消费的数据被覆盖空了还取取到的是无效数据互斥保证安全同步保证顺序和节奏缺一个模型都不成立生产者和消费者之间是最复杂的一条既要在同一块空间上互斥又要靠满和空两个条件同步线程十的条件变量就是给这一条准备的。2.4 三种关系对照关系性质针对什么生产者 vs 生产者竞争、互斥同一个场所的写入资格消费者 vs 消费者竞争、互斥同一份数据的取走资格生产者 vs 消费者互斥、同步同一块空间的读写加满空的节奏三、为什么要有生产者消费者模型3.1 解耦1. 不用模型时有多耦合生产线程直接调用消费线程就必须认识具体的消费方消费侧改个函数签名、换个参数类型生产侧跟着一起改以用户提交任务为例提交的一方和处理的一方直接调用处理方一动提交方就动两边绑死在一起2. 用模型之后双方只认识交易场所互相不见面生产者只管往场所里放消费者只管从场所里取换一个消费者、加十个消费者生产者一行代码不用动3.2 支持忙闲不均1. 速度不一致是常态生产者时快时慢消费者的处理速度也有限直接对接时快的那一阵数据直接把处理端打爆慢的那一阵处理端干等2. 场所吸收两侧的速度差生产快的那一阵多出来的数据先寄存在场所里相当于削峰生产慢下来的空档消费者回头把存货慢慢消化相当于填谷短时高峰不再等于系统过载处理能力不够就用空间来换时间3.3 提高效率解耦让两侧可以独立开发、独立变化忙闲不均让瞬时高峰不压垮处理端但提高效率这个词最容易想当然它到底高效在哪第四章单独掰开四、高效不在进出场所在获取与处理并发4.1 进出场所这一段是串行的生产和消费本身是互斥的往场所里放、从场所里取两步天生要排队同一时刻只能有一方在动交易场所这是互斥关系决定的所以别指望从搬进搬出这一段上找到效率这段只有串行4.2 消费者拿到商品之后还有一整个处理过程1. 拿到不等于干完消费者拿到商品只是获取动作完成后面还有对商品的具体处理假设拿到一个商品只要 1ms处理这个商品却要 1s处理的这一秒里消费者根本不占用交易场所场所对它是空闲的2. 这一段被大多数直觉漏掉直觉里的生产和消费只盯着进出场所那一瞬间真正的大头在场所之外的处理过程上那才是耗时的主体4.3 并发从哪冒出来消费者拿到任务、转身去处理的那一秒生产者可以继续往队列里 push 数据填充动作发生在交易场所上很短处理动作发生在交易场所之外很长一边填充一边处理获取任务和处理任务是并发的消费者不止一个时处理端还能横向叠加吞吐随消费者个数扩展提高效率不体现在入交易场所和出交易场所上而在于未来获取任务和处理具体任务是并发的。五、交易场所落地成阻塞队列5.1 阻塞队列和普通队列的差别普通队列空了还取、满了还塞要么报错要么未定义行为阻塞队列把节奏写进了结构队列空时 pop 阻塞直到有数据进来队列满时 push 阻塞直到有数据被取走差别的本质就是把三种关系做成了数据结构自身的行为5.2 一把锁加两个条件变量1. 一把互斥锁管所有互斥生产者之间、消费者之间、生产者和消费者之间的互斥全由这一把锁保证不管多少个线程动场所之前先拿锁一次只有一个在动2. 两个条件变量分管两个方向的唤醒not_full 给生产者等队列满时生产者在这上面等待not_empty 给消费者等队列空时消费者在这上面等待为什么要两个两边等的是相反的条件唤醒信号混着发被叫醒的线程一看条件还是不成立白醒一趟只用一个变量两个方向的线程挂在同一条等待队列上醒来后发现自己的条件仍不成立只能重新睡回去还多一次抢锁消费完唤醒 not_full 那头的生产者生产完唤醒 not_empty 那头的消费者3. 判条件必须用 while唤醒不等于条件成立伪唤醒和被抢先消费都会让条件再次变假线程十讲的 while 再判一次在这里原样适用用 if 单次判断迟早出事5.3 从单生产单消费到多生产多消费单生产单消费时互斥关系依然存在只是对手盘各只有一个人换成多生产多消费代码结构一行不用改生产者之间的互斥被那把锁顺带保证消费者之间同理变的只是线程数量不变的是模型的骨架六、demo拆解6.1 task任务classTask{public:Task(){}Task(intx,inty):_x(x),_y(y){}voidExecute(){_result_x_y;}intX(){return_x;}intY(){return_y;}intResult(){return_result;}private:int_x;int_y;int_result;};6.2 主函数逻辑void*consumer(void*mes){BlockTask*bq(BlockTask*)mes;while(true){sleep(1);Task tbq-Pop();std::cout我是客户端我拿到了一份数据x y t.Result()std::endl;}returnnullptr;}void*productor(void*mes){BlockTask*bq(BlockTask*)mes;intx1;inty1;while(true){sleep(1);std::cout我是服务端我生产了一份数据x y ?std::endl;Taskt(x,y);t.Execute();bq-Push(t);x;y;}returnnullptr;}intmain(){BlockTask*bqnewBlockTask();pthread_t p,c;pthread_create(c,nullptr,consumer,(void*)bq);pthread_create(p,nullptr,productor,(void*)bq);pthread_join(p,nullptr);pthread_join(c,nullptr);return0;}6.3 业务实现1 成员变量std::queueT_q;int_cap;//消费者数量pthread_mutex_t _block;//互斥锁pthread_cond_t _empty_cond;// 条件变量pthread_cond_t _full_cond;// 条件变量int_c_wait;//消费者等待个数int_p_wait;//生产者等待个数2 构造析构Block(){_c_wait0;_p_wait0;_capclientnum;pthread_mutex_init(_block,nullptr);pthread_cond_init(_empty_cond,nullptr);pthread_cond_init(_full_cond,nullptr);}~Block(){_cap0;pthread_mutex_destroy(_block);pthread_cond_destroy(_empty_cond);pthread_cond_destroy(_full_cond);}3 简单接口boolIsEmpty(){return_q.empty();//判断}boolIsfull(){return_q.size()_cap;}4 生产者入货voidPush(constT in){pthread_mutex_lock(_block);//原子操作 先加锁//用while而不用if如果多个生产者在等待时 同时被唤醒//(broad_cast)会同时执行后面逻辑本来只有一个拿到锁//能push但是大家都插入 导致重复重复插入。//如果采用while循环 生产者被唤醒后 再次进行判断不满足// 条件的继续阻塞等待while(Isfull()){_p_wait;pthread_cond_wait(_full_cond,_block);_p_wait--;//等待的生产者拿到锁之后数量自动减一}_q.push(in);//已经push超时肯定有货了再判断有没有等待的客户if(_c_wait0){pthread_cond_signal(_empty_cond);std::cout唤醒客户端std::endl;}pthread_mutex_unlock(_block);//释放锁}4 消费者拿货//逻辑和生产者入货一样TPop(){pthread_mutex_lock(_block);while(IsEmpty()){_c_wait;pthread_cond_wait(_empty_cond,_block);_c_wait--;}T data_q.front();_q.pop();if(_p_wait0){pthread_cond_signal(_full_cond);std::cout唤醒服务端std::endl;}pthread_mutex_unlock(_block);returndata;}完整demo#includeiostream#includevector#includepthread.h#includemutex#includequeueintclientnum5;namespaceBlockQueue{templateclassTclassBlock{private:boolIsEmpty(){return_q.empty();}boolIsfull(){return_q.size()_cap;}public:Block(){_c_wait0;_p_wait0;_capclientnum;pthread_mutex_init(_block,nullptr);pthread_cond_init(_empty_cond,nullptr);pthread_cond_init(_full_cond,nullptr);}voidPush(constT in){pthread_mutex_lock(_block);while(Isfull()){_p_wait;pthread_cond_wait(_full_cond,_block);_p_wait--;}_q.push(in);if(_c_wait0){pthread_cond_signal(_empty_cond);std::cout唤醒客户端std::endl;}pthread_mutex_unlock(_block);}TPop(){pthread_mutex_lock(_block);while(IsEmpty()){_c_wait;pthread_cond_wait(_empty_cond,_block);_c_wait--;}T data_q.front();_q.pop();if(_p_wait0){pthread_cond_signal(_full_cond);std::cout唤醒服务端std::endl;}pthread_mutex_unlock(_block);returndata;}~Block(){_cap0;pthread_mutex_destroy(_block);pthread_cond_destroy(_empty_cond);pthread_cond_destroy(_full_cond);}private:std::queueT_q;int_cap;pthread_mutex_t _block;pthread_cond_t _empty_cond;pthread_cond_t _full_cond;int_c_wait;int_p_wait;};}六、全篇总结1. 模型的三样东西两种角色生产者和消费者由线程承担一个交易场所以特定结构构成的内存空间同时是临界资源2. 三种关系生产者之间竞争关系落实为互斥消费者之间互斥生产者和消费者之间互斥加同步满和空各管一个方向3. 为什么要有这个模型解耦双方只认识交易场所互不绑定忙闲不均场所作缓冲削峰填谷提高效率效率的真正来源要看下一章的结论4. 高效的真正来源进出场所是串行的效率不在这段上拿任务 1ms、处理任务 1s处理期间场所空着生产者此时继续 push获取任务和处理任务是并发的5. 落地形态阻塞队列空了取阻塞、满了塞阻塞一把互斥锁加两个条件变量判条件用 while单生产单消费换多生产多消费骨架不变七、文末面试题7.1 推导题1. 生产者和生产者之间为什么是竞争、互斥关系答推导交易场所只有一份、容量有限多个生产者都往里放数据放的位置要靠抢这就是竞争关系而放数据要改场所的内部结构队尾指针和计数这些操作不是原子的两个生产者同时改交错执行就会把结构写坏和 ticket-- 三步被切走是同一个病根。所以必须互斥一次只允许一个生产者动场所。2. 消费者和消费者之间的互斥什么时候才会显现答推导只有一个消费者时这条关系看不见因为不存在对手盘消费者一多同一份数据只能被一个消费者拿走否则就是重复消费而且取数据同样要改队列结构并发地取一样写坏。所以这条关系是模型自带的硬约束不是线程多起来才新加的规则只是单线程时没机会暴露。3. 生产者和消费者之间为什么既要互斥又要同步答推导互斥这一半来自读写同一块空间同时读写会读到改到一半的半成品同步这一半来自满和空两个条件场所满时生产者继续塞会覆盖没消费的数据场所空时消费者继续取会拿到无效数据所以满时生产者等消费者腾位置、空时消费者等生产者放货。互斥保证安全同步保证节奏缺一个模型都不成立。4. 生产和消费明明是互斥的高效到底从哪来答推导高效不在入交易场所和出交易场所上这两段受互斥约束天生串行。真正的大头在场所之外消费者拿到任务只是获取动作完成假设拿到只要 1ms、处理却要 1s处理这一秒里消费者根本不占交易场所生产者可以继续往队列里 push。也就是说获取任务和处理具体任务是并发的多消费者时处理端还能横向叠加吞吐就上去了。5. 阻塞队列为什么要配两个条件变量一个不够吗答推导不够。生产者等的是不满消费者等的是不空这是两个相反的条件只用一个变量两个方向的线程挂在同一条等待队列上唤醒信号一来被叫醒的很可能不是能干活的那一方它一看条件不成立只能重新睡回去白醒一趟还多一次抢锁的开销。两个变量把两个方向的等待队列分开消费完只唤醒生产者、生产完只唤醒消费者唤醒才打得准。6. 阻塞队列和普通队列的差别是什么答推导差别在于三种关系有没有被写进数据结构自身。普通队列空了取、满了塞要么报错要么未定义行为节奏全靠调用方自己控制阻塞队列在空时让 pop 阻塞、满时让 push 阻塞等的是相反的条件变量被唤醒后还要用 while 再判一次条件。互斥、同步从调用方的义务变成了结构自身的行为这正是生产者消费者模型最直接的落地形态。7.2 真题1. 请解释一下生产者消费者模式特别是单生产者单消费者模型的基本原理和需要解决的关键问题。答推导 · 已对照公开考点转述模型里有生产者和消费者两类执行流通过一个固定大小、初始为空的缓冲区通信生产者生成数据写入缓冲区消费者从缓冲区读取处理缓冲区是临界资源任一时刻只允许一个线程操作。要解决的关键问题有两个一是防止生产者在缓冲区满时继续写入避免覆盖尚未消费的数据二是防止消费者在缓冲区空时读取避免拿到无效数据。手段是互斥锁保护缓冲区访问条件变量做等待与通知缓冲区非满时通知生产者、非空时通知消费者。【真题·转述自 CSDN《生产者消费者模型》2. 生产者和消费者之间是互斥、同步还是两者都有答推导 · 已对照解析转述既有同步也有互斥。互斥来自生产者和消费者对缓冲池这一临界资源的访问必须互斥进行同步来自生产的先后约束产品必须在消费之前先被生产出来缓冲区满时生产者要等消费者腾位置缓冲区空时消费者要等生产者放货。两个关系叠加在同一条通道上这也是生产者消费者问题在操作层面的标准答案。【真题·转述自 牛客网《关于生产者-消费者问题描述正确的是》结语生产者消费者模型说到底就三样东西、三种关系条件变量负责把满和空两个方向的等待与唤醒接住。想清楚高效不在搬运上、而在获取与处理的并发上这个模型才算真正想明白。评论区聊聊你第一次写阻塞队列踩的坑觉得有用点个赞再走Linux 系统篇持续更新。