Linux系统篇46——线程（十一） 生产者消费者模型，三种关系和一个交易场所

📚 本文收录于「流浪」的系列专栏

🏠 博客主页：流浪 ｜ 📝 原创首发于 CSDN

前言：线程（十）把条件变量讲到了底，唤醒机制加等待队列，wait 的两段阻塞也拆完了。本篇把互斥锁和条件变量组合成完整的多线程协作模型——生产者消费者。先摆清两种角色和一个交易场所，再逐一推演三种关系为什么存在，最后回答一个最容易想当然的问题：进出交易场所明明是串行的，高效到底从哪来。

一、两种角色和一个交易场所

1.1 两种角色由线程承担

1. 角色怎么分

1. 一种角色负责生产数据，把数据放进交易场所，叫生产者
2. 一种角色负责从交易场所取出数据、处理数据，叫消费者
3. 角色描述的是执行流干的事，落到实现里，就是由线程来承担

2. 角色不绑定线程个数

1. 一个线程扮演一个角色是最常见的写法，一个生产线程配一个消费线程
2. 也可以开多个生产者线程、多个消费者线程，角色仍然只有两种
3. 角色按"干什么"划分，不按线程数量划分，线程数只影响并发度

1.2 交易场所是一块特定结构的内存

1. 它长什么样

1. 交易场所不是一句空话，它是以特定结构构成的一种内存空间
2. 最常见的结构是阻塞队列和环形缓冲区，容量有限、先进先出，本篇以阻塞队列为例
3. 打个比方就是超市：生产者把商品摆上货架，消费者从货架上取走，超市就是那个交易场所

2. 它同时是临界资源

1. 所有的写入和读取都落在这一块空间上，多方共享
2. 三种关系全部围绕它产生，它是整个模型的交汇点

整个模型就三样东西：生产者、消费者、交易场所，后面所有的关系和规则，都是从这三样东西里长出来的。

二、三种关系，逐一推演为什么会存在

2.1 生产者和生产者之间，竞争互斥

1. 竞争从哪来

1. 多个生产者往同一个交易场所放数据，而场所只有一份
2. 容量有限，谁先放谁占位置，这就是竞争关系

2. 为什么必须互斥

1. 放数据不只是"放进一个格"，还要改场所的内部结构，队尾指针、计数都会跟着动
2. 两个生产者同时改这些数据，执行交错就会把结构写坏
   • 改指针和改计数都不是一步完成，中间可以被切走
   • 切回来后写的是自己看到的旧值，对方的改动被覆盖
3. 这正是线程（八）里 ticket-- 三步被切走的非原子问题，场景换了，病根没换
4. 所以生产者之间必须互斥，一次只允许一个生产者动场所

竞争是抢位置，互斥是抢到了就独占完成，两个词说的是同一件事的两面。

2.2 消费者和消费者之间，互斥竞争

1. 为什么是互斥关系

1. 一份数据只能被一个消费者拿走，被拿两次就是重复消费
2. 取数据同样要改队列结构，并发地取，一样会把结构写坏

2. 竞争从哪来

1. 多个消费者去同一个交易场所取数据，而场所只有一份
2. 容量有限，谁先取谁占位置，这就是竞争关系

2. 这条关系什么时候显现

1. 只有一个消费者时，这条关系看不见，对手盘不存在
2. 消费者一多，它和生产者之间那条一样，是硬约束，躲不掉

单线程时看不出的问题，不代表约束不存在，只代表还没人跟它抢。

2.3 生产者和消费者之间，互斥加同步

1. 互斥的那一半

1. 生产者写、消费者读，读写的是同一块空间
2. 同时读写，读到的可能是改到一半的半成品，所以必须互斥

2. 同步的两个方向

1. 场所满了，生产者必须停手等待，消费者先消费，腾出位置再把生产者唤醒
2. 场所空了，消费者必须等待，生产者先生产，有货了再把消费者唤醒
3. 两个方向合起来，就是让"该等的等、该走的走"，访问有节奏

3. 没有同步会怎样

1. 满了还塞，没消费的数据被覆盖
2. 空了还取，取到的是无效数据
3. 互斥保证安全，同步保证顺序和节奏，缺一个模型都不成立

生产者和消费者之间是最复杂的一条：既要在同一块空间上互斥，又要靠满和空两个条件同步，线程（十）的条件变量就是给这一条准备的。

2.4 三种关系对照

三、为什么要有生产者消费者模型

3.1 解耦

1. 不用模型时有多耦合

1. 生产线程直接调用消费线程，就必须认识具体的消费方
2. 消费侧改个函数签名、换个参数类型，生产侧跟着一起改
3. 以用户提交任务为例：提交的一方和处理的一方直接调用，处理方一动，提交方就动，两边绑死在一起

2. 用模型之后

1. 双方只认识交易场所，互相不见面
2. 生产者只管往场所里放，消费者只管从场所里取
3. 换一个消费者、加十个消费者，生产者一行代码不用动

3.2 支持忙闲不均

1. 速度不一致是常态

1. 生产者时快时慢，消费者的处理速度也有限
2. 直接对接时，快的那一阵数据直接把处理端打爆，慢的那一阵处理端干等

2. 场所吸收两侧的速度差

1. 生产快的那一阵，多出来的数据先寄存在场所里，相当于削峰
2. 生产慢下来的空档，消费者回头把存货慢慢消化，相当于填谷
3. 短时高峰不再等于系统过载，处理能力不够就用空间来换时间

3.3 提高效率

1. 解耦让两侧可以独立开发、独立变化
2. 忙闲不均让瞬时高峰不压垮处理端
3. 但"提高效率"这个词最容易想当然，它到底高效在哪，第四章单独掰开

四、高效不在进出场所，在获取与处理并发

4.1 进出场所这一段是串行的

1. 生产和消费本身是互斥的，往场所里放、从场所里取，两步天生要排队
2. 同一时刻只能有一方在动交易场所，这是互斥关系决定的
3. 所以别指望从搬进搬出这一段上找到效率，这段只有串行

4.2 消费者拿到商品之后还有一整个处理过程

1. 拿到不等于干完

1. 消费者拿到商品，只是获取动作完成，后面还有对商品的具体处理
2. 假设拿到一个商品只要 1ms，处理这个商品却要 1s
3. 处理的这一秒里，消费者根本不占用交易场所，场所对它是空闲的

2. 这一段被大多数直觉漏掉

1. 直觉里的"生产和消费"只盯着进出场所那一瞬间
2. 真正的大头在场所之外的处理过程上，那才是耗时的主体

4.3 并发从哪冒出来

1. 消费者拿到任务、转身去处理的那一秒，生产者可以继续往队列里 push 数据
   • 填充动作发生在交易场所上，很短
   • 处理动作发生在交易场所之外，很长
2. 一边填充一边处理，获取任务和处理任务是并发的
3. 消费者不止一个时，处理端还能横向叠加，吞吐随消费者个数扩展

提高效率不体现在入交易场所和出交易场所上，而在于未来获取任务和处理具体任务是并发的。

五、交易场所落地成阻塞队列

5.1 阻塞队列和普通队列的差别

1. 普通队列空了还取、满了还塞，要么报错要么未定义行为
2. 阻塞队列把节奏写进了结构：队列空时 pop 阻塞，直到有数据进来
3. 队列满时 push 阻塞，直到有数据被取走
4. 差别的本质，就是把三种关系做成了数据结构自身的行为

5.2 一把锁加两个条件变量

1. 一把互斥锁管所有互斥

1. 生产者之间、消费者之间、生产者和消费者之间的互斥，全由这一把锁保证
2. 不管多少个线程，动场所之前先拿锁，一次只有一个在动

2. 两个条件变量分管两个方向的唤醒

1. not_full 给生产者等：队列满时生产者在这上面等待
2. not_empty 给消费者等：队列空时消费者在这上面等待
3. 为什么要两个：两边等的是相反的条件，唤醒信号混着发，被叫醒的线程一看条件还是不成立，白醒一趟
   • 只用一个变量，两个方向的线程挂在同一条等待队列上
   • 醒来后发现自己的条件仍不成立，只能重新睡回去，还多一次抢锁
4. 消费完唤醒 not_full 那头的生产者，生产完唤醒 not_empty 那头的消费者

3. 判条件必须用 while

1. 唤醒不等于条件成立，伪唤醒和被抢先消费都会让条件再次变假
2. 线程（十）讲的 while 再判一次，在这里原样适用，用 if 单次判断迟早出事

5.3 从单生产单消费到多生产多消费

1. 单生产单消费时，互斥关系依然存在，只是对手盘各只有一个人
2. 换成多生产多消费，代码结构一行不用改
3. 生产者之间的互斥被那把锁顺带保证，消费者之间同理
4. 变的只是线程数量，不变的是模型的骨架

六、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){Block<Task>*bq=(Block<Task>*)mes;while(true){sleep(1);Task t=bq->Pop();std::cout<<"我是客户端，我拿到了一份数据"<<"x + y ="<<t.Result()<<std::endl;}returnnullptr;}void*productor(void*mes){Block<Task>*bq=(Block<Task>*)mes;intx=1;inty=1;while(true){sleep(1);std::cout<<"我是服务端，我生产了一份数据"<<"x + y = ?"<<std::endl;Taskt(x,y);t.Execute();bq->Push(t);x++;y++;}returnnullptr;}intmain(){Block<Task>*bq=newBlock<Task>();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::queue<T>_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_wait=0;_p_wait=0;_cap=clientnum;pthread_mutex_init(&_block,nullptr);pthread_cond_init(&_empty_cond,nullptr);pthread_cond_init(&_full_cond,nullptr);}~Block(){_cap=0;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_wait>0){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_wait>0){pthread_cond_signal(&_full_cond);std::cout<<"唤醒服务端"<<std::endl;}pthread_mutex_unlock(&_block);returndata;}

完整demo

#include<iostream>#include<vector>#include<pthread.h>#include<mutex>#include<queue>intclientnum=5;namespaceBlockQueue{template<classT>classBlock{private:boolIsEmpty(){return_q.empty();}boolIsfull(){return_q.size()>=_cap;}public:Block(){_c_wait=0;_p_wait=0;_cap=clientnum;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_wait>0){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_wait>0){pthread_cond_signal(&_full_cond);std::cout<<"唤醒服务端"<<std::endl;}pthread_mutex_unlock(&_block);returndata;}~Block(){_cap=0;pthread_mutex_destroy(&_block);pthread_cond_destroy(&_empty_cond);pthread_cond_destroy(&_full_cond);}private:std::queue<T>_q;int_cap;pthread_mutex_t _block;pthread_cond_t _empty_cond;pthread_cond_t _full_cond;int_c_wait;int_p_wait;};}

六、全篇总结

1. 模型的三样东西

1. 两种角色：生产者和消费者，由线程承担
2. 一个交易场所：以特定结构构成的内存空间，同时是临界资源

2. 三种关系

1. 生产者之间：竞争关系，落实为互斥
2. 消费者之间：互斥
3. 生产者和消费者之间：互斥加同步，满和空各管一个方向

3. 为什么要有这个模型

1. 解耦：双方只认识交易场所，互不绑定
2. 忙闲不均：场所作缓冲，削峰填谷
3. 提高效率：效率的真正来源要看下一章的结论

4. 高效的真正来源

1. 进出场所是串行的，效率不在这段上
2. 拿任务 1ms、处理任务 1s，处理期间场所空着
3. 生产者此时继续 push，获取任务和处理任务是并发的

5. 落地形态

1. 阻塞队列：空了取阻塞、满了塞阻塞
2. 一把互斥锁加两个条件变量，判条件用 while
3. 单生产单消费换多生产多消费，骨架不变

七、文末面试题

7.1 推导题

1. 生产者和生产者之间为什么是竞争、互斥关系？

答（推导）：交易场所只有一份、容量有限，多个生产者都往里放数据，放的位置要靠抢，这就是竞争关系；而放数据要改场所的内部结构，队尾指针和计数这些操作不是原子的，两个生产者同时改，交错执行就会把结构写坏，和 ticket-- 三步被切走是同一个病根。所以必须互斥，一次只允许一个生产者动场所。

2. 消费者和消费者之间的互斥，什么时候才会显现？

答（推导）：只有一个消费者时这条关系看不见，因为不存在对手盘；消费者一多，同一份数据只能被一个消费者拿走，否则就是重复消费，而且取数据同样要改队列结构，并发地取一样写坏。所以这条关系是模型自带的硬约束，不是线程多起来才新加的规则，只是单线程时没机会暴露。

3. 生产者和消费者之间为什么既要互斥又要同步？

答（推导）：互斥这一半来自读写同一块空间，同时读写会读到改到一半的半成品；同步这一半来自满和空两个条件，场所满时生产者继续塞会覆盖没消费的数据，场所空时消费者继续取会拿到无效数据，所以满时生产者等消费者腾位置、空时消费者等生产者放货。互斥保证安全，同步保证节奏，缺一个模型都不成立。

4. 生产和消费明明是互斥的，高效到底从哪来？

答（推导）：高效不在入交易场所和出交易场所上，这两段受互斥约束，天生串行。真正的大头在场所之外：消费者拿到任务只是获取动作完成，假设拿到只要 1ms、处理却要 1s，处理这一秒里消费者根本不占交易场所，生产者可以继续往队列里 push。也就是说获取任务和处理具体任务是并发的，多消费者时处理端还能横向叠加，吞吐就上去了。

5. 阻塞队列为什么要配两个条件变量，一个不够吗？

答（推导）：不够。生产者等的是"不满"，消费者等的是"不空"，这是两个相反的条件；只用一个变量，两个方向的线程挂在同一条等待队列上，唤醒信号一来，被叫醒的很可能不是能干活的那一方，它一看条件不成立只能重新睡回去，白醒一趟还多一次抢锁的开销。两个变量把两个方向的等待队列分开，消费完只唤醒生产者、生产完只唤醒消费者，唤醒才打得准。

6. 阻塞队列和普通队列的差别是什么？

答（推导）：差别在于三种关系有没有被写进数据结构自身。普通队列空了取、满了塞，要么报错要么未定义行为，节奏全靠调用方自己控制；阻塞队列在空时让 pop 阻塞、满时让 push 阻塞，等的是相反的条件变量，被唤醒后还要用 while 再判一次条件。互斥、同步从调用方的义务变成了结构自身的行为，这正是生产者消费者模型最直接的落地形态。

7.2 真题

1. 请解释一下生产者消费者模式，特别是单生产者单消费者模型的基本原理和需要解决的关键问题。

答（推导 · 已对照公开考点，转述）：模型里有生产者和消费者两类执行流，通过一个固定大小、初始为空的缓冲区通信，生产者生成数据写入缓冲区，消费者从缓冲区读取处理，缓冲区是临界资源，任一时刻只允许一个线程操作。要解决的关键问题有两个，一是防止生产者在缓冲区满时继续写入，避免覆盖尚未消费的数据；二是防止消费者在缓冲区空时读取，避免拿到无效数据。手段是互斥锁保护缓冲区访问，条件变量做等待与通知，缓冲区非满时通知生产者、非空时通知消费者。
【真题·转述自 CSDN《生产者消费者模型》

2. 生产者和消费者之间是互斥、同步，还是两者都有

答（推导 · 已对照解析，转述）：既有同步也有互斥。互斥来自生产者和消费者对缓冲池这一临界资源的访问必须互斥进行；同步来自生产的先后约束，产品必须在消费之前先被生产出来，缓冲区满时生产者要等消费者腾位置，缓冲区空时消费者要等生产者放货。两个关系叠加在同一条通道上，这也是生产者消费者问题在操作层面的标准答案。
【真题·转述自 牛客网《关于生产者-消费者问题描述正确的是》

💬结语：生产者消费者模型说到底就三样东西、三种关系，条件变量负责把"满"和"空"两个方向的等待与唤醒接住。想清楚高效不在搬运上、而在获取与处理的并发上，这个模型才算真正想明白。评论区聊聊你第一次写阻塞队列踩的坑，觉得有用点个赞再走，Linux 系统篇持续更新。