基础生产者消费者模型
网上一般教程是使用std::queue,定义消费者 Consumer ,定义Producter类,在main函数里面加锁保证线程安全。
本片文章,实现一个线程安全的队列 threadsafe_queue,只在队列内部加锁。如此可适配,多生产者多消费者的场景
线程安全的队列 threadsafe_queue
#pragma once
#include<mutex>
#include <condition_variable>
#include<queue>
//最大产品数量
#define MAX_SIZE 20
#include <iostream>
template<typename T>
class threadsafe_queue
{
private:
mutable std::mutex mut;
std::queue<T> data_queue;
std::condition_variable data_cond;
public:
threadsafe_queue()
{}
void wait_and_pop(T& value) // 2
{
std::unique_lock<std::mutex> lk(mut);
data_cond.wait(lk, [this]{return !data_queue.empty(); });
value = std::move(data_queue.front());
data_queue.pop();
std::cout << "Consumer pop : " << value << std::endl;
data_cond.notify_all();
}
std::shared_ptr<T> wait_and_pop() // 3
{
std::unique_lock<std::mutex> lk(mut);
data_cond.wait(lk, [this]{return !data_queue.empty(); }); // 4
std::shared_ptr<T> res(
std::make_shared<T>(std::move(data_queue.front())));
data_queue.pop();
data_cond.notify_all();
return res;
}
bool try_pop(T& value)
{
std::lock_guard<std::mutex> lk(mut);
if (data_queue.empty())
return false;
value = std::move(data_queue.front());
data_queue.pop();
data_cond.notify_all();
return true;
}
std::shared_ptr<T> try_pop()
{
std::lock_guard<std::mutex> lk(mut);
if (data_queue.empty())
return std::shared_ptr<T>(); // 5
std::shared_ptr<T> res(
std::make_shared<T>(std::move(data_queue.front())));
data_queue.pop();
data_cond.notify_all();
return res;
}
void push(T new_value)
{
std::unique_lock<std::mutex> lk(mut);
data_cond.wait(lk, [this] {return data_queue.size() < MAX_SIZE; });
std::cout << "Producter push : " << new_value << std::endl;
data_queue.push(std::move(new_value));
data_cond.notify_all(); // 1
}
bool empty() const
{
std::lock_guard<std::mutex> lk(mut);
return data_queue.empty();
}
};
消费者
Consumer 头文件
#pragma once
#include"threadsafe_queue.h"
/*
基础版 消费者生产者模型
此文件为 消费者
*/
class Consumer
{
public:
Consumer(threadsafe_queue<int>& queue);
~Consumer();
void start();
private:
threadsafe_queue<int>& m_pQueue;
};
cpp文件
#include "Consumer.h"
#include <iostream>
#include<windows.h>
Consumer::Consumer(threadsafe_queue<int> &queue)
: m_pQueue(queue)
{
}
Consumer::~Consumer()
{
}
void Consumer::start()
{
while (true)
{
int value;
m_pQueue.wait_and_pop(value);
Sleep(10);
}
}
生产者
#pragma once
#include"threadsafe_queue.h"
class Producter
{
public:
Producter(threadsafe_queue<int>& queue,int i);
~Producter();
void start();
private:
threadsafe_queue<int>& m_pQueue;
int index;
};
#include "Producter.h"
#include<stdlib.h>
#include <iostream>
#include<windows.h>
Producter::Producter(threadsafe_queue<int>& queue, int i)
: m_pQueue(queue),index(i)
{
}
Producter::~Producter()
{
}
void Producter::start()
{
int i = 0;//为了测试是哪个线程,加入了哪些数据用的
while (true)
{
//int data = rand();
m_pQueue.push(i*2+ index);// index =0 为偶数生产者,只生产偶数,index =1则是生产奇数
Sleep(100);
i++;
}
}
运行结果如下:
控制消费者有序消费
优先队列做缓存
头文件
#pragma once
#include<mutex>
#include <condition_variable>
#include <iostream>
#include <algorithm>
#include <vector>
#include <queue>
//最大产品数量
#define MAX_SIZE 20
/*
*/
template<typename T, class _Container = std::vector<T>, class _Pr = std::less<typename _Container::value_type>>
class threadsafe_priority_queue
{
public:
threadsafe_priority_queue();
~threadsafe_priority_queue();
void push(T new_value);
bool empty() const;
void wait_and_pop(T& value);
std::shared_ptr<T> wait_and_pop();
bool try_pop(T& value);
std::shared_ptr<T> try_pop();
void wait_and_pop(T& value, bool (*func)(const T&, int), int curIndex);
std::shared_ptr<T> wait_and_pop(bool (*func)(const T&, int), int curIndex);
bool try_pop(T& value, bool (*func)(const T&, int), int curIndex);
std::shared_ptr<T> try_pop( bool (*func)(const T&, int), int curIndex);
private:
mutable std::mutex mut;
std::priority_queue<T, _Container, _Pr> data_queue;
std::condition_variable data_cond;
};
实现
#include "threadsafe_priority_queue.h"
#include <iomanip>
template<typename T, class _Container, class _Pr >
threadsafe_priority_queue<T, _Container, _Pr>::threadsafe_priority_queue()
{}
template<typename T, class _Container, class _Pr >
threadsafe_priority_queue<T, _Container, _Pr>::~threadsafe_priority_queue()
{
}
template<typename T, class _Container, class _Pr >
void threadsafe_priority_queue<T, _Container, _Pr>::wait_and_pop(T& value) // 2
{
std::unique_lock<std::mutex> lk(mut);
data_cond.wait(lk, [this] {return !data_queue.empty(); });
value = std::move(data_queue.front());
data_queue.pop();
std::cout << "Consumer pop : " << value << std::endl;
data_cond.notify_all();
}
template<typename T, class _Container, class _Pr >
std::shared_ptr<T> threadsafe_priority_queue<T, _Container, _Pr>::wait_and_pop() // 3
{
std::unique_lock<std::mutex> lk(mut);
data_cond.wait(lk, [this] {return !data_queue.empty(); }); // 4
std::shared_ptr<T> res(
std::make_shared<T>(std::move(data_queue.front())));
data_queue.pop();
data_cond.notify_all();
return res;
}
template<typename T, class _Container, class _Pr >
bool threadsafe_priority_queue<T, _Container, _Pr>::try_pop(T& value)
{
std::lock_guard<std::mutex> lk(mut);
if (data_queue.empty())
return false;
value = std::move(data_queue.front());
data_queue.pop();
data_cond.notify_all();
return true;
}
template<typename T, class _Container, class _Pr >
std::shared_ptr<T> threadsafe_priority_queue<T, _Container, _Pr>::try_pop()
{
std::lock_guard<std::mutex> lk(mut);
if (data_queue.empty())
return std::shared_ptr<T>(); // 5
std::shared_ptr<T> res(
std::make_shared<T>(std::move(data_queue.front())));
data_queue.pop();
data_cond.notify_all();
return res;
}
template<typename T, class _Container, class _Pr >
void threadsafe_priority_queue<T, _Container, _Pr>::push(T new_value)
{
std::unique_lock<std::mutex> lk(mut);
data_cond.wait(lk, [this] {return data_queue.size() < MAX_SIZE; });
std::cout.flags(std::ios::left); //左对齐
std::cout << "Producter push : " << new_value << std::endl;
data_queue.push(std::move(new_value));
data_cond.notify_all(); // 1
}
template<typename T, class _Container, class _Pr >
bool threadsafe_priority_queue<T, _Container, _Pr>::empty() const
{
std::lock_guard<std::mutex> lk(mut);
return data_queue.empty();
}
template<typename T, class _Container, class _Pr >
void threadsafe_priority_queue<T, _Container, _Pr>::wait_and_pop(T& value, bool (*func)(const T&, int), int curIndex )
{
std::unique_lock<std::mutex> lk(mut);
data_cond.wait(lk, [this, func, curIndex] {
if (!data_queue.empty() && func(data_queue.top(), curIndex))
{
return true;
}
if (!data_queue.empty())
std::cout << "top value: " << data_queue.top() << " ,current index : " << curIndex << std::endl;
return false;
});
value = std::move(data_queue.top());
data_queue.pop();
//右对齐
std::cout << std::setw(100) << setiosflags(std::ios::right)<< "Consumer pop : " << value << std::endl;
data_cond.notify_all();
}
template<typename T, class _Container, class _Pr >
std::shared_ptr<T> threadsafe_priority_queue<T, _Container, _Pr>::wait_and_pop(bool (*func)(const T&, int), int curIndex)
{
std::unique_lock<std::mutex> lk(mut);
data_cond.wait(lk, [this, func, curIndex] {
if (!data_queue.empty() && func(data_queue.front(), curIndex))
{
return true;
}
return false;
}); // 4
std::shared_ptr<T> res(
std::make_shared<T>(std::move(data_queue.front())));
data_queue.pop();
data_cond.notify_all();
return res;
}
template<typename T, class _Container, class _Pr >
bool threadsafe_priority_queue<T, _Container, _Pr>::try_pop(T& value, bool (*func)(const T&, int), int curIndex)
{
std::lock_guard<std::mutex> lk(mut);
if (data_queue.empty() || func(data_queue.front(), curIndex))
{
return std::shared_ptr<T>();
}
std::shared_ptr<T> res(
std::make_shared<T>(std::move(data_queue.front())));
data_queue.pop();
data_cond.notify_all();
return res;
}
template<typename T, class _Container, class _Pr >
std::shared_ptr<T> threadsafe_priority_queue<T, _Container, _Pr>::try_pop(bool (*func)(const T&, int), int curIndex)
{
std::lock_guard<std::mutex> lk(mut);
if (data_queue.empty() || func(data_queue.front(), curIndex))
{
return std::shared_ptr<T>();
}
std::shared_ptr<T> res(
std::make_shared<T>(std::move(data_queue.front())));
data_queue.pop();
data_cond.notify_all();
return res;
}
生产者
#include "threadsafe_priority_queue.h"
class Producter2
{
public:
Producter2(threadsafe_priority_queue<int, std::vector<int>, std::greater<int>>& queue, int i);
~Producter2();
void start();
private:
threadsafe_priority_queue<int, std::vector<int>, std::greater<int>>& m_pQueue;
int index;
};
Producter2::Producter2(threadsafe_priority_queue<int, std::vector<int>, std::greater<int>>& queue, int i)
: m_pQueue(queue), index(i)
{
}
Producter2::~Producter2()
{
}
void Producter2::start()
{
int i = 0;
while (i < 200)
{
//int data = rand();
m_pQueue.push(i * 3 + index);
Sleep(30);
i++;
}
}
消费者
#include "threadsafe_priority_queue.h"
class Consumer2
{
public:
Consumer2(threadsafe_priority_queue<int, std::vector<int>, std::greater<int>>& queue);
~Consumer2();
void start();
private:
threadsafe_priority_queue<int, std::vector<int>, std::greater<int>>& m_pQueue;
int m_curIndex=-1;
};
Consumer2::Consumer2(threadsafe_priority_queue<int, std::vector<int>, std::greater<int>>& queue)
: m_pQueue(queue)
{
}
Consumer2::~Consumer2()
{
}
bool compare( const int & nextIndex,int curindex)
{
return nextIndex == (curindex + 1);
}
void Consumer2::start()
{
int i = 0;
while (i < 200)
{
int value;
m_pQueue.wait_and_pop(value, compare, m_curIndex);
m_curIndex++;
Sleep(10);
i++;
}
}
main函数文章来源:https://www.toymoban.com/news/detail-637754.html
#include <iostream>
#include "Producter.h"
#include "Consumer.h"
# include<thread>
#include"threadsafe_queue.h"
#include"threadsafe_queue.cpp"
#include<windows.h>
#include "threadsafe_priority_queue.h"
#include "threadsafe_priority_queue.cpp"
int main()
{
threadsafe_priority_queue<int, std::vector<int>, std::greater<int>> queue;
Consumer2 aonsumer(queue);
Producter2 producter(queue,0);
Producter2 producter2(queue, 1);
Producter2 producter3(queue, 2);
std::thread tProducter3(&Producter2::start, &producter3);
std::thread tProducter2(&Producter2::start, &producter2);
std::thread tProducter1(&Producter2::start, &producter);
std::thread tConsumer(&Consumer2::start, &aonsumer);
tConsumer.join();
tProducter3.join();
tProducter2.join();
tProducter1.join();
return 0;
}
如何停止消费者?
采用毒丸策略
多线程生产者、消费者模式中,如何停止消费者文章来源地址https://www.toymoban.com/news/detail-637754.html
到了这里,关于生产者消费者模型 C++ 版的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!