一.先前代码及实现（在该篇中会用到）
2.基于linux操作系统的线程封装（可实现任意传递任意类型任意个数的参数）-CSDN博客
二.生产消费者模型
在一个多线程的进程中，通常存在如下关系生产者和消费者，其中生产者负责生产资源（产生任务）消费者负责消耗资源（获取任务），一般而言，消费者和消费者互斥，生产者和生产者互斥，消费者和生产者互斥且同步。我们将存放任务的地方（阻塞队列）称为交易场所。当阻塞队列满时，需要生产者停止发放任务，而为空时则需要消费者停止拿取任务。当阻塞队列由满变非满时，需要消费者唤醒生产者进行生产任务，反之亦然。
由于同步和互斥的存在，以及额外的阻塞队列存在，生产者将任务递达至消费者 这个步骤会产生额外的开销，但是多消费者和多生产者可以充分利用CPU的多核，这使得生产者在生产任务，消费者在处理任务可以并行处理，极大的节约了生产任务和处理任务的总时长。
三.条件变量
当阻塞队列（交易场所）由空变为非空时，可以利用条件变量唤醒消费线程，反之当阻塞队列由满变为非满时，也可以利用条件变量唤醒生产线程，条件变量可以极大的减少锁竞争所带来的开销。
有关条件变量的封装可参考基于Linux操作系统的锁的封装-CSDN博客
class condition_variable
{
public:
condition_variable(const condition_variable&)=delete;
condition_variable operator=(condition_variable)=delete;
condition_variable(mutex& mutex):_mutex(mutex) //条件变量创建
{
if(pthread_cond_init(&_cond,nullptr))
{
std::cerr<<errno<<strerror(errno)<<std::endl;
exit(1);
}
}
~condition_variable() //条件变量析构
{
if(pthread_cond_destroy(&_cond))
{
std::cerr<<errno<<strerror(errno)<<std::endl;
exit(2);
}
}
void wait() //当没有资源时则等待
{
if(pthread_cond_wait(&_cond,&_mutex.getmutex()))
{
std::cerr<<errno<<strerror(errno)<<std::endl;
exit(3);
}
}
void notify_one() //唤醒一个线程
{
if(pthread_cond_signal(&_cond))
{
std::cerr<<errno<<strerror(errno)<<std::endl;
exit(4);
}
}
void notify_all() //唤醒所有线程
{
if(pthread_cond_broadcast(&_cond))
{
std::cerr<<errno<<strerror(errno)<<std::endl;
exit(5);
}
}
private:
mutex& _mutex;
pthread_cond_t _cond;
};四.生产消费者队列封装（普通队列）
1.成员变量，构造和析构函数
template<class T>
class prodcon
{
public:
typedef T task;
prodcon(size_t max_task=normal_max_task)
:_mutex(nullptr),_num_task(0),_max_task(max_task)
,_pro_cond(_mutex),_con_cond(_mutex)
{
}
~prodcon(){}
private:
size_t _num_task; //当前任务数量
size_t _max_task; //最大任务数量
std::queue<task> _qtask; //阻塞队列 task为放置任务
condition_variable _pro_cond; //生产者条件变量，自定义封装
condition_variable _con_cond; //消费者条件变量，自定义封装
mutex _mutex; //互斥锁，自定义封装
};所有的成员变量均有对应的析构函数，因此析构函数不需要做任何处理
2.放任务和出任务
void push_task(task& in,int pri=10) //放入任务，pri是为了适配用优先级队列而引入的，无需注意
{
{
lockguard<mutex> L(_mutex); //上锁
while(isfull()) //判断是否满
{
_pro_cond.wait(); //如果满则等待
}
_qtask.push(in); //放入任务
++_num_task; //当前任务+1
}
_con_cond.notify_one(); //唤醒线程，由于在唤醒线程时队列中一定存在任务，因此
//此处可以不加锁
}
void pop_task(task& out) //取得任务
{
{
lockguard<mutex> L(_mutex); //加锁
while(isempty()) //判断是不是为空
{
_con_cond.wait();
}
out= std::move(_qtask.front());
_qtask.pop(); //取任务
--_num_task; //任务数量-1
}
_pro_cond.notify_one(); //唤醒生产者线程
}
int pop_some_tasks(task out[],int num)//取出多个任务
{
int real_num=0;
{
lockguard<mutex> L(_mutex);
while(isempty())
{
_con_cond.wait();
}
for(int i=0;i<num;++i)
{
if(isempty()) break; //如果是空代表任务完成则返回
out[i]= std::move(_qtask.front());
++real_num;
_qtask.pop();
--_num_task;
}
}
_pro_cond.notify_all();//由于一次取出多个任务，因此可以
//全局唤醒所有线程
return real_num; //返回取出了多少任务
}
private：
bool isfull() //判断是否为满
{
return !(_max_task-_num_task);
}
bool isempty() //判断是否为空
{
return !_num_task;
}3.其他成员函数
bool full()
{
lockguard<mutex> L(_mutex); //上锁
return !(_max_task-_num_task);
}
bool empty()
{
lockguard<mutex> L(_mutex); //上锁
return !_num_task;
}可以外界访问任务是否为空或者满，由于没有上锁因此要上锁，该函数不能成员函数中已上锁的时候调用，会发生死锁
五.生产消费者模型（优先级队列）
优先级队列和普通队列执行一样，唯一的不同就是优先级队列在放入任务时需要设置该任务的优先级，这也是为什么在普通队列的放入任务中，会出现 int pri=10，这是为了适配优先级队列。另外需要注意，普通队列的插入和取出均为o(1)，而优先级队列的插入和取出均为o(logn)，如果非必要，不要使用优先级队列
1.完整代码（包括普通队列和优先级队列）
#ifndef _PRODCON_HPP
#define _PRODCON_HPP
#include<iostream>
#include<vector>
#include<memory>
#include<pthread.h>
#include<cstring>
#include<functional>
#include<unistd.h>
#include<queue>
#include"lock.hpp"
static const int normal_max_task=10;
static const int max_pri=100;
static const int min_pri=0;
namespace zwr
{
template<class T>
class prodcon
{
public:
typedef T task;
prodcon(size_t max_task=normal_max_task)
:_mutex(nullptr),_num_task(0),_max_task(max_task)
,_pro_cond(_mutex),_con_cond(_mutex)
{
}
~prodcon(){}
void push_task(task& in,int pri=10)
{
{
lockguard<mutex> L(_mutex);
while(isfull())
{
_pro_cond.wait();
}
_qtask.push(in);
++_num_task;
}
_con_cond.notify_one();
}
void pop_task(task& out)
{
{
lockguard<mutex> L(_mutex);
while(isempty())
{
_con_cond.wait();
}
out= std::move(_qtask.front());
//std::cout<<"::"<<_qtask.front()<<std::endl;
_qtask.pop();
--_num_task;
}
_pro_cond.notify_one();
}
int pop_some_tasks(task out[],int num)
{
int real_num=0;
{
lockguard<mutex> L(_mutex);
while(isempty())
{
_con_cond.wait();
}
for(int i=0;i<num;++i)
{
if(isempty()) break;
out[i]= std::move(_qtask.front());
++real_num;
_qtask.pop();
--_num_task;
}
}
_pro_cond.notify_all();
return real_num;
}
bool full()
{
lockguard<mutex> L(_mutex);
return !(_max_task-_num_task);
}
bool empty()
{
lockguard<mutex> L(_mutex);
return !_num_task;
}
private:
bool isfull()
{
return !(_max_task-_num_task);
}
bool isempty()
{
return !_num_task;
}
size_t _num_task;
size_t _max_task;
std::queue<task> _qtask;
condition_variable _pro_cond;
condition_variable _con_cond;
mutex _mutex;
};
template<class T>
class priority_prodcon //优先级队列
{
public:
typedef T task;
priority_prodcon(size_t max_task=normal_max_task)
:_mutex(nullptr),_num_task(0),_max_task(max_task),_qtask(cmp())
,_pro_cond(_mutex),_con_cond(_mutex)
{
}
~priority_prodcon(){}
void push_task(task& in,int pri=10)
{
if(pri>max_pri) pri=max_pri;
if(pri<min_pri) pri=min_pri;
{lockguard<mutex> L(_mutex);
while(isfull())
{
_pro_cond.wait();
}
_qtask.push(std::make_pair(in,pri));
++_num_task;}
_con_cond.notify_one();
}
void pop_task(task& out)
{
{lockguard<mutex> L(_mutex);
while(isempty())
{
_con_cond.wait();
}
out= std::move(_qtask.top().first);
_qtask.pop();
--_num_task;}
_pro_cond.notify_one();
}
int pop_some_tasks(task out[],int num)
{
int real_num=0;
{
lockguard<mutex> L(_mutex);
while(isempty())
{
_con_cond.wait();
}
for(int i=0;i<num;++i)
{
if(isempty()) break;
out[i]= std::move(_qtask.top().first);
++real_num;
_qtask.pop();
--_num_task;
}
}
_pro_cond.notify_all();
return real_num;
}
bool full()
{
lockguard<mutex> L(_mutex);
return !(_max_task-_num_task);
}
bool empty()
{
lockguard<mutex> L(_mutex);
return !_num_task;
}
private:
bool isfull()
{
return !(_max_task-_num_task);
}
bool isempty()
{
return !_num_task;
}
struct cmp
{
cmp(){}
~cmp(){}
bool operator()(std::pair<task,int> a1,std::pair<task,int> a2)
{
if(a1.second>a2.second) return true;
else return false;
}
};
size_t _num_task;
size_t _max_task;
std::priority_queue<std::pair<task,int>,std::vector<std::pair<task,int>>,cmp> _qtask; //优先级队列
condition_variable _pro_cond;
condition_variable _con_cond;
mutex _mutex;
};
}
#endif