C++实现生产者和消费者模型
- C++实现生产者和消费者模型
- 1、实现细节
- 2、单生产者-单消费者模型
- 3、单生产者-多消费者模型
- 4、多生产者-单消费者模型
- 5、多生产者-多消费者模型
- 参考
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;
static const int ProNum = 20;
struct resource {
int buf[bufSize];
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) {
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);
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) {
std::cout << "Produce the " << i << "^th item..." << std::endl;
Producer(&instance, i);
}
}
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;
}
}
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();
}
![在这里插入图片描述](https://img-blog.csdnimg.cn/20200909151720631.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pNVzE0MDc=,size_16,color_FFFFFF,t_70#pic_center)
3、单生产者-多消费者模型
与单生产者和单消费者模型不同的是,单生产者-多消费者模型中可以允许多个消费者同时从产品库中取走产品。所以除了保护产品库在多个读写线程下互斥之外,还需要维护消费者取走产品的计数器,代码如下:
#include <unistd.h>
#include <cstdlib>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <thread>
static const int bufSize = 8;
static const int ProNum = 30;
struct resource {
int buf[bufSize];
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) {
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);
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) {
std::cout << "Producer thread " << std::this_thread::get_id()
<< " producing the " << i << "^th item..." << std::endl;
Producer(&instance, i);
}
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();
}
![在这里插入图片描述](https://img-blog.csdnimg.cn/20200909163516563.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pNVzE0MDc=,size_16,color_FFFFFF,t_70#pic_center)
4、多生产者-单消费者模型
与单生产者和单消费者模型不同的是,多生产者-单消费者模型中可以允许多个生产者同时向产品库中放入产品。所以除了保护产品库在多个读写线程下互斥之外,还需要维护生产者放入产品的计数器,代码如下:
#include <unistd.h>
#include <cstdlib>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <thread>
static const int bufSize = 8;
static const int ProNum = 20;
struct resource {
int buf[bufSize];
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) {
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);
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;
}
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();
}
![在这里插入图片描述](https://img-blog.csdnimg.cn/20200909165222294.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pNVzE0MDc=,size_16,color_FFFFFF,t_70#pic_center)
5、多生产者-多消费者模型
该模型可以说是前面两种模型的综合,程序需要维护两个计数器,分别是生产者已生产产品的数目和消费者已取走产品的数目。另外也需要保护产品库在多个生产者和多个消费者互斥地访问。
#include <unistd.h>
#include <cstdlib>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <thread>
static const int bufSize = 8;
static const int ProNum = 20;
struct resource {
int buf[bufSize];
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) {
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);
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;
}
![在这里插入图片描述](https://img-blog.csdnimg.cn/20200909170907805.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pNVzE0MDc=,size_16,color_FFFFFF,t_70#pic_center)
参考
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
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)