C++ 实现线程安全的任务队列
flyfish 2015⑶⑹
1、3个接口函数说明
1 add 新增任务
2 get_nonblocking 非阻塞获得任务或空任务
3 get_blocking 阻塞获得任务
头文件
#pragma once
#include <deque>
#include <boost/thread/mutex.hpp>
#include <boost/thread/locks.hpp>
#include <boost/thread/condition_variable.hpp>
//任务 网络发送任务使用的结构，通常有1个发送缓冲区和1个实际要发送的长度
class task
{
public:
unsigned char data[2048];
unsigned int len;//实际发送长度
task::task();
task::~task();
};
class task_queue
{
private:
std::deque<task> tasks;
boost::mutex tasks_mutex;
boost::condition_variable cv;
public:
task_queue::task_queue();
task_queue::~task_queue();
void add(const task& task);
std::tuple<bool,task> get_nonblock();
task get_block();
};
实现文件
#include "task_queue.h"
//task
task::task()
{
for (int i=0;i<2048;i++)
data[i] = 0;
}
task::~task()
{
}
// task queue
task_queue::task_queue()
{
}
task_queue::~task_queue()
{
}
void task_queue::add(const task& task)
{
boost::unique_lock<boost::mutex> lock(tasks_mutex);//不允许其他线程履行
tasks.push_back(task);
lock.unlock();
cv.notify_one();//通知其他线程继续
}
std::tuple<bool,task> task_queue::get_nonblock()
{
boost::lock_guard<boost::mutex> lock(tasks_mutex);
std::tuple<bool,task> ret;
if (!tasks.empty())
{
ret=std::make_tuple(true,tasks.front());
tasks.pop_front();
}
else
{
task tmp;
ret=std::make_tuple(false,tmp);
}
return ret;
}
task task_queue::get_block()
{
boost::unique_lock<boost::mutex> lock(tasks_mutex);
while (tasks.empty())
{
cv.wait(lock);
}
task ret=tasks.front();
tasks.pop_front();
return ret;
}
2 解释
1 notify_one用于唤醒1个等待该条件(condition)产生的线程
2 可使用boost::mutex::scoped_lock 替换boost::unique_lock<boost::mutex>，可以免遗漏unlock
例如
boost::mutex mutex_;
try
{
mutex_.lock();
//do something
mutex_.unlock();
}
catch(...)
{
mutex_.unlock();
return 0;
}
使用boost::mutex::scoped_lock就能够避免catch遗漏unlock
3 boost::unique_lock仅仅比boost::lock_guard附加1些功能
4 如果任务在获得以后不删除，就能够使用多读1写方式，就要实现读写锁
读操作产生时： 写线程停止操作，允许多个线程同时读操作
写操作产生时： 只允许同1时刻只有1个线程写操作，其他不管读写线程全部停止。
代码类似
typedef boost::shared_lock<boost::shared_mutex> r_lock;
typedef boost::unique_lock<boost::shared_mutex> w_lock;
boost::shared_mutex mutex_;
void read()
{
r_lock lock(mutex_);
//do something
}
void write()
{
w_lock lock(mutex_);
//do something
}
以上程序在VC2010 Boost库下编译通过