生产者消费者问题 C++实现

知识准备

thread

介绍

• 定位于头文件的class thread
• 表示单个执行线程, 没有两个thread对象会表示同一个线程
  • 不可复制构造
  • 不可复制赋值

成员类

• id

成员函数

• get_id

  • 返回线程的id

• hardware_concurrency

  • 返回实现支持的并发线程数

• join

  • 等待线程完成其执行

sleep_for

介绍

• 定义域头文件

• 声明

  • template< class Rep, class Period >
    void sleep_for( const std::chrono::duration<Rep, Period>& sleep_duration );
    

  • sleep_duration : 要睡眠的时长

mutex

介绍

• 定义于头文件

• mutex 提供排他性非递归所有权语义：

  • 调用方线程从它成功调用 lock 或 try_lock 开始，到它调用 unlock 为止占有 mutex 。
  • 线程占有 mutex 时，所有其他线程若试图要求 mutex 的所有权，则将阻塞（对于 lock 的调用）或收到 false 返回值（对于 try_lock ）.
  • 调用方线程在调用 lock 或 try_lock 前必须不占有 mutex 。

成员函数

• lock

  • 锁定互斥, 若互斥不可用则堵塞

• unlock

  • 解锁互斥

unique_lock

介绍

• 定义于

• 声明

  template< class Mutex >
  class unique_lock;
  

成员函数

• lock
  • 锁定关联互斥
• unlock
  • 解锁关联互斥
• mutex
  • 返回指向关联互斥的指针

codition_variable

介绍

• 定义于头文件 <condition_variable>
• 声明class condition_variable;
• 有意修改变量的线程必须:
  • 1. 获得std:: mutex(通过std::unique_lock)
    2. 在保有锁时进行修改
    3. 执行 notify_one 或 notify_all
• 任何有意在 std::condition_variable 上等待的线程必须
  • 1. 获得std::unique_lock <std::mutex>
    2. 执行 wait 、 wait_for 或 wait_until ，等待操作自动释放互斥，并悬挂线程的执行
    3. 线程被唤醒，且自动重获得互斥
• **std::condition_variable只可与 std::unique_lock<std::mutex>一同使用；

成员函数

• notify_one

  • 通知一个等待线程

• notify_all

  • 通知所有等待线程

• wait

  • 阻塞当前进程, 直至被唤醒

• wait_for

  • 阻塞当前线程，直到条件变量被唤醒，或到指定时限时长后
  • wait_until

    • 阻塞当前线程，直到条件变量被唤醒，或直到抵达指定时间点
      

代码示例

// operator_system.cpp: 定义控制台应用程序的入口点。
//
 
#include "stdafx.h"
#include<iostream>
#include <mutex>
#include <condition_variable>
#include <windows.h>
#include <thread>
 
using namespace std;
 
static const int buffer_size = 10; // 缓存大小
static const int item_total = 100; //总共要生产 item_total个item
 
// 缓存结构体, 使用循环队列当做缓存
struct Buffer 
{
	int buffer[buffer_size];
	size_t read_position; // 当前读位置
	size_t write_position; // 当前写位置
	mutex mtx; // 读写互斥
	//条件变量
	condition_variable not_full; 
	condition_variable not_empty;
}buffer_res;
 
typedef struct Buffer Buffer;
 
void porduce_item(Buffer *b, int item)
{
	unique_lock<mutex> lock(b->mtx);//设置互斥锁
 
	while(((b->write_position + 1) % buffer_size) == b->read_position) {
		//当前缓存已经满了
		cout << "buffer is full now, producer is wating....." << endl;
		(b->not_full).wait(lock); // 等待缓存非full
	}
	// 向缓存中添加item
	(b->buffer)[b->write_position] = item;
	(b->write_position)++;
 
	// 若到达最后一个, 写位置置位0
	if (b->write_position == buffer_size)
		b->write_position = 0;
 
	(b->not_empty).notify_all();
	lock.unlock();
}
 
int consume_item(Buffer *b)
{
	int data;
	unique_lock <mutex> lock(b->mtx);
	while (b->write_position == b->read_position)
	{	// 当前buffer 为空
		cout << "buffer is empty , consumer is waiting....." << endl;
		(b->not_empty).wait(lock);
	}
 
	data = (b->buffer)[b->read_position];
	(b->read_position)++;
 
	if (b->read_position >= buffer_size)
		b->read_position = 0;
 
	(b->not_full).notify_all();
	lock.unlock();
 
	return data;
}
 
//生产者任务
void producer() {
	for (int i = 1; i<= item_total;i++) {
		cout << "prodece the " << i << "^th item ..." << endl;
		porduce_item(&buffer_res, i);
	}
}
 
//消费者任务
void consumer()
{
	static int cnt = 0;
	while(1) {
		Sleep(1);
		int item = consume_item(&buffer_res);
		cout << "consume the " << item << "^th item" << endl;
		if (++cnt == item_total)
			break;
	}
}
 
//初始化 buffer
void init_buffer(Buffer *b)
{
	b->write_position = 0;
	b->read_position = 0;
}
 
int main()
{
	init_buffer(&buffer_res);
	thread prodece(producer);
	thread consume(consumer);
	prodece.join();
	consume.join();
	getchar();
}
 

查看其他精彩文章
深入分析HashMap
AOP核心原理和SpringAOP
10分钟入门SpringAOP

生产者消费者问题C++实现

2020/10/11 11:23

问题

设计C/C++程序(可以嵌入汇编语言)，以忙等待方式实现信号量及其P、V操作。利用你实现的信号量，实现生产者-消费者问题。给出实现方法、主要源代码和测试结果。

代码

# include<iostream>
# include<thread>
# include<vector>
# include<mutex>
# include<condition_variable>
# include<queue>
#include <ctime>
#include <windows.h>
using namespace std;

# define PRODUCT_SIZE 5//生产者数量
# define CUSTOMER_SIZE 20//消费者数量
#define MAX_SIZE 10//最大产品数量
mutex mut;//互斥锁
condition_variable con;//条件变量
queue<int> que;//队列，模拟缓冲区
void Producter()
{

	while (true)
	{
	
		Sleep(10);
		srand(int (time(NULL)));
		std::unique_lock <std::mutex> lck(mut);
		/*当 std::condition_variable对象的某个wait 函数被调用的时候，
		它使用 std::unique_lock(通过 std::mutex) 来锁住当前线程。
		当前线程会一直被阻塞，直到另外一个线程在相同的 std::condition_variable 对象上调用*/
		while (que.size() > MAX_SIZE)//当队列已满时等待，不再生产
		{
			con.wait(lck);//P操作
		}
		int data = rand();//随机产生数字代表商品的生产
		que.push(data);//将数据推入队列代表商品加入缓冲区
		cout << this_thread::get_id() << "生产了产品：" << data << endl;
		Sleep(500);
		con.notify_all();//唤醒所有等待的进程，即V操作

	}
}
void Customer()
{
	while (true)
	{
		std::unique_lock <std::mutex> lck(mut);
		while (que.empty())//当队列为空时等待，不再消费
		{
			//用condition_variable对象实现对缓冲区的互斥操作
			con.wait(lck);//P操作
		}
		cout << this_thread::get_id() << "消费了产品：" << que.front() << endl;
		Sleep(500);
		que.pop();
		con.notify_all();//唤醒所有等待的进程，即V操作
	}
}
int main()
{
	vector<thread> threadPoll;
	//创建生产者和消费者
	for (int i = 0; i < PRODUCT_SIZE; ++i)
	{
		threadPoll.push_back(thread(Producter));
	}
	for (int i = 0; i < PRODUCT_SIZE + CUSTOMER_SIZE; ++i)
	{
		threadPoll.push_back(thread(Customer));
	}
	for (int i = 0; i < PRODUCT_SIZE + CUSTOMER_SIZE; ++i)
	{
		threadPoll[i].join();//原始线程等到新线程执行完毕后再销毁
	}
	return 0;
}

运行截图

1.利用条件变量condition_variable实现PV操作，当condition_variable对象的某个wait 函数被调用的时候，它使用 unique_lock(通过 mutex) 来锁住当前线程。当前线程会一直被阻塞，直到另外一个线程在相同的condition_variable 对象上调用
2.用产生的随机数字代表商品的生产，加入队列表示加入缓冲区
3.线程销毁采用join方式，等待新线程执行完毕再销毁原始线程

C++实现生产者和消费者模型

1、实现细节

• 具体的实现逻辑是构建一个queue来存储生产的数据，queue不满时可以生产，不空时可以消费。
• 对于这个队列，采用阻塞队列的实现思路。
• 先实现构造函数，初始化一个unique_lock供condition_variable使用。
  • 如何在类里面使用unique_lock等需要初始化，并且初始化会加锁的对象。这要研究下。我的理解是构造列表初始化，然后函数体里unlock。
  • 对于条件变量，申请两个，分别控制consumer和producer。
• 然后就是入和出队列的细节。
  • 首先加锁。
  • 循环判断一下目前的队列情况，对于各自的特殊情况（队满和队空）进行处理。
  • 唤醒一个线程来处理特殊情况。
  • 等待处理完毕。
  • 处理入和出队列操作。
  • 最后释放锁。

2、单生产者-单消费者模型

• 单生产者-单消费者模型中只有一个生产者和一个消费者，
• 生产者不停地往产品库中放入产品，
• 消费者则从产品库中取走产品，
• 产品库容积有限制，只能容纳一定数目的产品，
• 如果生产者生产产品的速度过快，则需要等待消费者取走产品之后，产品库不为空才能继续往产品库中放置新的产品，
• 相反，如果消费者取走产品的速度过快，则可能面临产品库中没有产品可使用的情况，此时需要等待生产者放入一个产品后，消费者才能继续工作。

C++11实现单生产者单消费者模型的代码如下：

#include <unistd.h>

#include <cstdlib>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <thread>

static const int bufSize = 10; // Item buffer size.
static const int ProNum = 20;   // How many items we plan to produce.

struct resource {
	int buf[bufSize]; // 产品缓冲区, 配合 read_pos 和 write_pos 模型环形队列.
	size_t read_pos; // 消费者读取产品位置.
	size_t write_pos; // 生产者写入产品位置.
	std::mutex mtx; // 互斥量,保护产品缓冲区
	std::condition_variable not_full; // 条件变量, 指示产品缓冲区不为满.
	std::condition_variable not_empty; // 条件变量, 指示产品缓冲区不为空.
} instance; // 产品库全局变量, 生产者和消费者操作该变量.

typedef struct resource resource;


void Producer(resource *ir, int item)
{
	std::unique_lock<std::mutex> lock(ir->mtx);
	while (((ir->write_pos + 1) % bufSize)
		== ir->read_pos) { // item buffer is full, just wait here.
		std::cout << "Producer is waiting for an empty slot...\n";
		(ir->not_full).wait(lock); // 生产者等待"产品库缓冲区不为满"这一条件发生.
	}

	(ir->buf)[ir->write_pos] = item; // 写入产品.
	(ir->write_pos)++; // 写入位置后移.

	if (ir->write_pos == bufSize) // 写入位置若是在队列最后则重新设置为初始位置.
		ir->write_pos = 0;

	(ir->not_empty).notify_all(); // 通知消费者产品库不为空.
}

int Consumer(resource *ir)
{
	int data;
	std::unique_lock<std::mutex> lock(ir->mtx);
	// item buffer is empty, just wait here.
	while (ir->write_pos == ir->read_pos) {
		std::cout << "Consumer is waiting for items...\n";
		(ir->not_empty).wait(lock); // 消费者等待"产品库缓冲区不为空"这一条件发生.
	}

	data = (ir->buf)[ir->read_pos]; // 读取某一产品
	(ir->read_pos)++; // 读取位置后移

	if (ir->read_pos >= bufSize) // 读取位置若移到最后，则重新置位.
		ir->read_pos = 0;

	(ir->not_full).notify_all(); // 通知消费者产品库不为满.

	return data; // 返回产品.
}


void ProducerTask() // 生产者任务
{
	for (int i = 1; i <= ProNum; ++i) {
		// sleep(1);
		std::cout << "Produce the " << i << "^th item..." << std::endl;
		Producer(&instance, i); // 循环生产 ProNum 个产品.
	}
}

void ConsumerTask() // 消费者任务
{
	static int cnt = 0;
	while (1) {
		sleep(1);
		int item = Consumer(&instance); // 消费一个产品.
		std::cout << "Consume the " << item << "^th item" << std::endl;
		if (++cnt == ProNum) break; // 如果产品消费个数为 ProNum, 则退出.
	}
}

void Initresource(resource *ir)
{
	ir->write_pos = 0; // 初始化产品写入位置.
	ir->read_pos = 0; // 初始化产品读取位置.
}

int main()
{
	Initresource(&instance);
	std::thread producer(ProducerTask); // 创建生产者线程.
	std::thread consumer(ConsumerTask); // 创建消费之线程.
	producer.join();
	consumer.join();
}

3、单生产者-多消费者模型

与单生产者和单消费者模型不同的是，单生产者-多消费者模型中可以允许多个消费者同时从产品库中取走产品。所以除了保护产品库在多个读写线程下互斥之外，还需要维护消费者取走产品的计数器，代码如下:

#include <unistd.h>

#include <cstdlib>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <thread>

static const int bufSize = 8; // Item buffer size.
static const int ProNum = 30;   // How many items we plan to produce.

struct resource {
	int buf[bufSize]; // 产品缓冲区, 配合 read_pos 和 write_pos 模型环形队列.
	size_t read_pos; // 消费者读取产品位置.
	size_t write_pos; // 生产者写入产品位置.
	size_t item_counter;
	std::mutex mtx; // 互斥量,保护产品缓冲区
	std::mutex item_counter_mtx;
	std::condition_variable not_full; // 条件变量, 指示产品缓冲区不为满.
	std::condition_variable not_empty; // 条件变量, 指示产品缓冲区不为空.
} instance; // 产品库全局变量, 生产者和消费者操作该变量.

typedef struct resource resource;


void Producer(resource *ir, int item)
{
	std::unique_lock<std::mutex> lock(ir->mtx);
	while (((ir->write_pos + 1) % bufSize)
		== ir->read_pos) { // item buffer is full, just wait here.
		std::cout << "Producer is waiting for an empty slot...\n";
		(ir->not_full).wait(lock); // 生产者等待"产品库缓冲区不为满"这一条件发生.
	}

	(ir->buf)[ir->write_pos] = item; // 写入产品.
	(ir->write_pos)++; // 写入位置后移.

	if (ir->write_pos == bufSize) // 写入位置若是在队列最后则重新设置为初始位置.
		ir->write_pos = 0;

	(ir->not_empty).notify_all(); // 通知消费者产品库不为空.
	lock.unlock(); // 解锁.
}

int Consumer(resource *ir)
{
	int data;
	std::unique_lock<std::mutex> lock(ir->mtx);
	// item buffer is empty, just wait here.
	while (ir->write_pos == ir->read_pos) {
		std::cout << "Consumer is waiting for items...\n";
		(ir->not_empty).wait(lock); // 消费者等待"产品库缓冲区不为空"这一条件发生.
	}

	data = (ir->buf)[ir->read_pos]; // 读取某一产品
	(ir->read_pos)++; // 读取位置后移

	if (ir->read_pos >= bufSize) // 读取位置若移到最后，则重新置位.
		ir->read_pos = 0;

	(ir->not_full).notify_all(); // 通知消费者产品库不为满.
	lock.unlock(); // 解锁.

	return data; // 返回产品.
}


void ProducerTask() // 生产者任务
{
	for (int i = 1; i <= ProNum; ++i) {
		// sleep(1);
		std::cout << "Producer thread " << std::this_thread::get_id()
			<< " producing the " << i << "^th item..." << std::endl;
		Producer(&instance, i); // 循环生产 ProNum 个产品.
	}
	std::cout << "Producer thread " << std::this_thread::get_id()
		<< " is exiting..." << std::endl;
}

void ConsumerTask() // 消费者任务
{
	bool ready_to_exit = false;
	while (1) {
		sleep(1);
		std::unique_lock<std::mutex> lock(instance.item_counter_mtx);
		if (instance.item_counter < ProNum) {
			int item = Consumer(&instance);
			++(instance.item_counter);
			std::cout << "Consumer thread " << std::this_thread::get_id()
				<< " is consuming the " << item << "^th item" << std::endl;
		}
		else
			ready_to_exit = true;
		if (ready_to_exit == true)
			break;
	}
	std::cout << "Consumer thread " << std::this_thread::get_id()
		<< " is exiting..." << std::endl;
}

void Initresource(resource *ir)
{
	ir->write_pos = 0; // 初始化产品写入位置.
	ir->read_pos = 0; // 初始化产品读取位置.
	ir->item_counter = 0;
}

int main()
{
	Initresource(&instance);
	std::thread producer(ProducerTask);
	std::thread consumer1(ConsumerTask);
	std::thread consumer2(ConsumerTask);
	std::thread consumer3(ConsumerTask);
	std::thread consumer4(ConsumerTask);

	producer.join();
	consumer1.join();
	consumer2.join();
	consumer3.join();
	consumer4.join();
}

4、多生产者-单消费者模型

与单生产者和单消费者模型不同的是，多生产者-单消费者模型中可以允许多个生产者同时向产品库中放入产品。所以除了保护产品库在多个读写线程下互斥之外，还需要维护生产者放入产品的计数器，代码如下:

#include <unistd.h>

#include <cstdlib>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <thread>

static const int bufSize = 8; // Item buffer size.
static const int ProNum = 20;   // How many items we plan to produce.

struct resource {
	int buf[bufSize]; // 产品缓冲区, 配合 read_pos 和 write_pos 模型环形队列.
	size_t read_pos; // 消费者读取产品位置.
	size_t write_pos; // 生产者写入产品位置.
	size_t item_counter;
	std::mutex mtx; // 互斥量,保护产品缓冲区
	std::mutex item_counter_mtx;
	std::condition_variable not_full; // 条件变量, 指示产品缓冲区不为满.
	std::condition_variable not_empty; // 条件变量, 指示产品缓冲区不为空.
} instance; // 产品库全局变量, 生产者和消费者操作该变量.

typedef struct resource resource;


void Producer(resource *ir, int item)
{
	std::unique_lock<std::mutex> lock(ir->mtx);
	while (((ir->write_pos + 1) % bufSize)
		== ir->read_pos) { // item buffer is full, just wait here.
		std::cout << "Producer is waiting for an empty slot...\n";
		(ir->not_full).wait(lock); // 生产者等待"产品库缓冲区不为满"这一条件发生.
	}

	(ir->buf)[ir->write_pos] = item; // 写入产品.
	(ir->write_pos)++; // 写入位置后移.

	if (ir->write_pos == bufSize) // 写入位置若是在队列最后则重新设置为初始位置.
		ir->write_pos = 0;

	(ir->not_empty).notify_all(); // 通知消费者产品库不为空.
}

int Consumer(resource *ir)
{
	int data;
	std::unique_lock<std::mutex> lock(ir->mtx);
	// item buffer is empty, just wait here.
	while (ir->write_pos == ir->read_pos) {
		std::cout << "Consumer is waiting for items...\n";
		(ir->not_empty).wait(lock); // 消费者等待"产品库缓冲区不为空"这一条件发生.
	}

	data = (ir->buf)[ir->read_pos]; // 读取某一产品
	(ir->read_pos)++; // 读取位置后移

	if (ir->read_pos >= bufSize) // 读取位置若移到最后，则重新置位.
		ir->read_pos = 0;

	(ir->not_full).notify_all(); // 通知消费者产品库不为满.

	return data; // 返回产品.
}


void ProducerTask() // 生产者任务
{
	bool ready_to_exit = false;
	while (1) {
		sleep(1);
		std::unique_lock<std::mutex> lock(instance.item_counter_mtx);
		if (instance.item_counter < ProNum) {
			++(instance.item_counter);
			Producer(&instance, instance.item_counter);
			std::cout << "Producer thread " << std::this_thread::get_id()
				<< " is producing the " << instance.item_counter
				<< "^th item" << std::endl;
		}
		else 
			ready_to_exit = true;
		if (ready_to_exit == true) 
			break;
	}
	std::cout << "Producer thread " << std::this_thread::get_id()
		<< " is exiting..." << std::endl;
}

void ConsumerTask() // 消费者任务
{
	static int cnt = 0;
	while (1) {
		sleep(1);
		cnt++;
		if (cnt <= ProNum)
		{
			int item = Consumer(&instance); // 消费一个产品.
			std::cout << "Consumer thread " << std::this_thread::get_id()
				<< " is consuming the " << item << "^th item" << std::endl;
		}
		
		else
			break; // 如果产品消费个数为 ProNum, 则退出.
	}
	std::cout << "Consumer thread " << std::this_thread::get_id()
		<< " is exiting..." << std::endl;
}

void Initresource(resource *ir)
{
	ir->write_pos = 0; // 初始化产品写入位置.
	ir->read_pos = 0; // 初始化产品读取位置.
	ir->item_counter = 0;
}

int main()
{
	Initresource(&instance);
	std::thread producer1(ProducerTask);
	std::thread producer2(ProducerTask);
	std::thread producer3(ProducerTask);
	std::thread producer4(ProducerTask);
	std::thread consumer(ConsumerTask);

	producer1.join();
	producer2.join();
	producer3.join();
	producer4.join();
	consumer.join();
}

5、多生产者-多消费者模型

该模型可以说是前面两种模型的综合，程序需要维护两个计数器，分别是生产者已生产产品的数目和消费者已取走产品的数目。另外也需要保护产品库在多个生产者和多个消费者互斥地访问。

#include <unistd.h>

#include <cstdlib>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <thread>

static const int bufSize = 8; // Item buffer size.
static const int ProNum = 20;   // How many items we plan to produce.

struct resource {
	int buf[bufSize]; // 产品缓冲区, 配合 read_pos 和 write_pos 模型环形队列.
	size_t read_pos; // 消费者读取产品位置.
	size_t write_pos; // 生产者写入产品位置.
	size_t pro_item_counter;
	size_t con_item_counter;
	std::mutex mtx; // 互斥量,保护产品缓冲区
	std::mutex pro_mtx;
	std::mutex con_mtx;
	std::condition_variable not_full; // 条件变量, 指示产品缓冲区不为满.
	std::condition_variable not_empty; // 条件变量, 指示产品缓冲区不为空.
} instance; // 产品库全局变量, 生产者和消费者操作该变量.

typedef struct resource resource;


void Producer(resource *ir, int item)
{
	std::unique_lock<std::mutex> lock(ir->mtx);
	while (((ir->write_pos + 1) % bufSize)
		== ir->read_pos) { // item buffer is full, just wait here.
		std::cout << "Producer is waiting for an empty slot...\n";
		(ir->not_full).wait(lock); // 生产者等待"产品库缓冲区不为满"这一条件发生.
	}

	(ir->buf)[ir->write_pos] = item; // 写入产品.
	(ir->write_pos)++; // 写入位置后移.

	if (ir->write_pos == bufSize) // 写入位置若是在队列最后则重新设置为初始位置.
		ir->write_pos = 0;

	(ir->not_empty).notify_all(); // 通知消费者产品库不为空.
}

int Consumer(resource *ir)
{
	int data;
	std::unique_lock<std::mutex> lock(ir->mtx);
	// item buffer is empty, just wait here.
	while (ir->write_pos == ir->read_pos) {
		std::cout << "Consumer is waiting for items...\n";
		(ir->not_empty).wait(lock); // 消费者等待"产品库缓冲区不为空"这一条件发生.
	}

	data = (ir->buf)[ir->read_pos]; // 读取某一产品
	(ir->read_pos)++; // 读取位置后移

	if (ir->read_pos >= bufSize) // 读取位置若移到最后，则重新置位.
		ir->read_pos = 0;

	(ir->not_full).notify_all(); // 通知消费者产品库不为满.

	return data; // 返回产品.
}


void ProducerTask() // 生产者任务
{
	bool ready_to_exit = false;
	while (1) {
		sleep(1);
		std::unique_lock<std::mutex> lock(instance.pro_mtx);
		if (instance.pro_item_counter < ProNum) {
			++(instance.pro_item_counter);
			Producer(&instance, instance.pro_item_counter);
			std::cout << "Producer thread " << std::this_thread::get_id()
				<< " is producing the " << instance.pro_item_counter
				<< "^th item" << std::endl;
		}
		else 
			ready_to_exit = true;
		lock.unlock();
		if (ready_to_exit == true) 
			break;
	}
	std::cout << "Producer thread " << std::this_thread::get_id()
		<< " is exiting..." << std::endl;
}

void ConsumerTask() // 消费者任务
{
	bool ready_to_exit = false;
	while (1) {
		sleep(1);
		std::unique_lock<std::mutex> lock(instance.con_mtx);
		if (instance.con_item_counter < ProNum) {
			int item = Consumer(&instance);
			++(instance.con_item_counter);
			std::cout << "Consumer thread " << std::this_thread::get_id()
				<< " is consuming the " << item << "^th item" << std::endl;
		}
		else
			ready_to_exit = true;
		lock.unlock();
		if (ready_to_exit == true)
			break;
	}
	std::cout << "Consumer thread " << std::this_thread::get_id()
		<< " is exiting..." << std::endl;
}

void Initresource(resource *ir)
{
	ir->write_pos = 0; // 初始化产品写入位置.
	ir->read_pos = 0; // 初始化产品读取位置.
	ir->pro_item_counter = 0;
	ir->con_item_counter = 0;
}

int main()
{
	Initresource(&instance);
	std::thread producer1(ProducerTask);
	std::thread producer2(ProducerTask);
	std::thread producer3(ProducerTask);
	std::thread producer4(ProducerTask);

	std::thread consumer1(ConsumerTask);
	std::thread consumer2(ConsumerTask);
	std::thread consumer3(ConsumerTask);
	std::thread consumer4(ConsumerTask);

	producer1.join();
	producer2.join();
	producer3.join();
	producer4.join();

	consumer1.join();
	consumer2.join();
	consumer3.join();
	consumer4.join();
	return 0;
}

参考

1、https://www.cnblogs.com/haippy/p/3252092.html
2、https://blog.csdn.net/qq_41681241/article/details/86708303
3、https://blog.csdn.net/h_wulingfei/article/details/104897449