C++多线程:生产者消费者模式

文章目录

一、模式简介
二、头文件、全局变量
2.1 仓库类的设计
- 2.1.1 关于仓库类的分析
- 2.1.2 仓库类的设计代码
2.2 工厂类的设计
- 2.2.1 关于工厂类的分析
- 2.2.2 工厂类的设计代码
- - a 将产品item放到仓库repo
  - b 将产品item从仓库repo取出
  - c 生产者操作
  - d 消费者操作
- 2.2.3 主函数代码
三、运行效果和说明

一、模式简介

假设你有一个工厂Factory，配有一个仓库Repository，仓库大小为20，要生产200个产品，你有两个工人producer，三个受众群体consumer，应当如何描述这200个产品从生产到消费的全过程？
这一种生活中的问题在多线程程序设计上称为生产者消费者模式，形象的表示了多线程中数据获取、数据存储、数据处理的过程,关键点在于用互斥锁、条件变量等解决数据存取、同时存、同时取之间的冲突。

二、头文件、全局变量

引入头文件，按照问题描述，可以得到计划产品个数和仓库大小两个全局变量。

#include<iostream>
#include<thread>
#include<mutex>
#include<condition_variable>
#include<queue>
using namespace std;const int kProduceAimSize = 200;	//计划产品个数
const int kRepositorySize = 20;		//仓库大小

2.1 仓库类的设计

2.1.1 关于仓库类的分析

按照上上面提出的问题，我们大致能分析得到以下线索：
工厂描述了问题总体，是一个用于解决问题的类，核心在于用于数据交互的仓库类。
对于仓库的数据管理，应当设计三个互斥量，依次为：存取冲突互斥量、同时存的互斥量、同时取的互斥量。两个条件变量，依次表示为：仓库状态为空、仓库状态为满。

2.1.2 仓库类的设计代码

template <typename _T>
class Repository {		//仓库
public:deque<_T> items_buff;mutex mmutex;					//生产者和消费者互斥量，解决从仓库中存取的冲突mutex mmutex_produce;			//生产者之间的互斥量，解决同时生产同一个产品的冲突mutex mmutex_consume;			//消费者之间的互斥量，解决同时消费同一个产品的冲突mutex mmutex_print;				//保证由cout打印的提示信息不会中断condition_variable repo_full;	//描述仓库为满的条件变量condition_variable repo_empty;	//描述仓库为空的条件变量size_t cnt_produce;				//生产者目前产生的产品总数size_t cnt_consume;				//消费者目前消费的产品总数size_t current_size;			//仓库中产品个数Repository() {					//初始化cnt_produce=0;cnt_consume=0;current_size = 0;}
};

2.2 工厂类的设计

2.2.1 关于工厂类的分析

按照问题描述，工厂由工人和消费者组成，流程上需要完成四个基本任务：
1、生产产品
2、将产品放入仓库
3、从仓库中取出产品
4、消费产品
同时注意工厂类包含一个仓库类实例。

因此工厂类的大致架构如下：
在这里插入图片描述

2.2.2 工厂类的设计代码

a 将产品item放到仓库repo

void PutInto(Repository<_T>& repo, _T item) {unique_lock< mutex> lk(repo.mmutex);//仓库是一个存取数据的地方，因而对仓库进行操作需要上锁。(解决从仓库中存取的冲突)while (repo.current_size == kRepositorySize) {	//如果仓库为满,无法向仓库中继续放unique_lock<mutex> pt(repo.mmutex_print);cout << "仓库已满，无法放入更多产品，需先消费。" << endl;pt.unlock();repo.repo_full.wait(lk);	//将取互斥量抛出，等待其他线程通知}repo.items_buff.push_back(item);repo.current_size++;repo.repo_empty.notify_all();	//有产品了，唤醒仓库为空情况下的线程
}

b 将产品item从仓库repo取出

_T GetFrom(Repository<_T>& repo) {unique_lock<mutex> lk(repo.mmutex);	//仓库是一个存取数据的地方，因而对仓库进行操作需要上锁。(解决从仓库中存取的冲突)while (repo.current_size == 0) {	//如果仓库为空，等待unique_lock<mutex> pt(repo.mmutex_print);cout << "无货源，等待..." << endl;pt.unlock();repo.repo_empty.wait(lk);	//将存取互斥量抛出，等待其他线程通知}_T data = repo.items_buff.front();repo.items_buff.pop_front();repo.current_size--;repo.repo_full.notify_all();	//从仓库取出了产品，唤醒仓库为满情况下的线程return data;
}

c 生产者操作

void ProduceTask() {bool ready_to_exit = false; //线程结束条件while (true) {unique_lock<mutex> lk(repo.mmutex_produce);	//不能生产同一个产品if (repo.cnt_produce < kProduceAimSize) {//需要生产(没达到目标)repo.cnt_produce++;//生产产品假设0.05sthis_thread::sleep_for(0.05s);	_T item = repo.cnt_produce;	//生产产品具体过程的化简unique_lock<mutex> pt(repo.mmutex_print);cout << "生产者id:" << this_thread::get_id() << " as[" << item<<"]" << endl;pt.unlock();PutInto(repo, item);}else {ready_to_exit = true;}if (ready_to_exit == true)break;}
}

d 消费者操作

void ConsumeTask() {bool ready_to_exit = false;	//线程结束条件while (true){unique_lock<mutex> lk(repo.mmutex_consume);	//不能同时消费一个产品if (repo.cnt_consume < kProduceAimSize) {//需要消费(没达到目标)_T item = GetFrom(repo);	//获取需要消费的产品repo.cnt_consume++;//消费产品代码，假设消费0.06sthis_thread::sleep_for(0.06s);	//消费过程的化简unique_lock<mutex> pt(repo.mmutex_print);cout << "消费者id:" << this_thread::get_id() << " as[" << item<<"]" << endl;pt.unlock();}else {ready_to_exit = true;}if (ready_to_exit == true)break;}
}

2.2.3 主函数代码

int main() {cout << "主线程id:" << this_thread::get_id() << endl;//一个工厂类Factory<int> MyFactory;//两个生产者thread producer1(&Factory<int>::ProduceTask, ref(MyFactory));thread producer2(&Factory<int>::ProduceTask, ref(MyFactory));//三个消费者thread consumer1(&Factory<int>::ConsumeTask, ref(MyFactory));thread consumer2(&Factory<int>::ConsumeTask, ref(MyFactory));thread consumer3(&Factory<int>::ConsumeTask, ref(MyFactory));producer1.join();	producer2.join();consumer1.join();consumer2.join();consumer3.join();return 0;
}