C++基于消息队列的多线程实现示例代码

2023-08-25 00:23:05 62

前言

实现消息队列的关键因素是考量不同线程访问消息队列的同步问题。本实现涉及到几个知识点

std::lock_guard介绍

std::lock_gurad是C++11中定义的模板类。定义如下：

templateclasslock_guard;

lock_guard对象通常用于管理某个锁(Lock)对象，因此与MutexRAII相关，方便线程对互斥量上锁，即在某个lock_guard对象的声明周期内，它所管理的锁对象会一直保持上锁状态；而lock_guard的生命周期结束之后，它所管理的锁对象会被解锁(注：类似shared_ptr等智能指针管理动态分配的内存资源)。

模板参数Mutex代表互斥量类型，例如std::mutex类型，它应该是一个基本的BasicLockable类型，标准库中定义几种基本的BasicLockable类型，分别std::mutex,std::recursive_mutex,std::timed_mutex，std::recursive_timed_mutex以及std::unique_lock

std::unique_lock介绍

lock_guard最大的缺点也是简单，没有给程序员提供足够的灵活度，因此，C++11标准中定义了另外一个与MutexRAII相关类unique_lock，该类与lock_guard类相似，也很方便线程对互斥量上锁，但它提供了更好的上锁和解锁控制。
顾名思义，unique_lock对象以独占所有权的方式（uniqueowership）管理mutex对象的上锁和解锁操作，所谓独占所有权，就是没有其他的unique_lock对象同时拥有某个mutex对象的所有权。

新创建的unique_lock对象管理Mutex对象m，并尝试调用m.lock()对Mutex对象进行上锁，如果此时另外某个unique_lock对象已经管理了该Mutex对象m，则当前线程将会被阻塞。

std::condition介绍

当std::condition_variable对象的某个wait函数被调用的时候，它使用std::unique_lock(通过std::mutex)来锁住当前线程。当前线程会一直被阻塞，直到另外一个线程在相同的std::condition_variable对象上调用了notification函数来唤醒当前线程。

std::condition_variable提供了两种wait()函数。当前线程调用wait()后将被阻塞(此时当前线程应该获得了锁（mutex），不妨设获得锁lck)，直到另外某个线程调用notify_*唤醒了当前线程。

在线程被阻塞时，该函数会自动调用lck.unlock()释放锁，使得其他被阻塞在锁竞争上的线程得以继续执行。另外，一旦当前线程获得通知(notified，通常是另外某个线程调用notify_*唤醒了当前线程)，wait()函数也是自动调用lck.lock()，使得lck的状态和wait函数被调用时相同。

在第二种情况下（即设置了Predicate），只有当pred条件为false时调用wait()才会阻塞当前线程，并且在收到其他线程的通知后只有当pred为true时才会被解除阻塞。因此第二种情况类似以下代码：

while(!pred())wait(lck);

std::function介绍

使用std::function可以将普通函数，lambda表达式和函数对象类统一起来。它们并不是相同的类型，然而通过function模板类，可以转化为相同类型的对象（function对象），从而放入一个vector或其他容器里，方便回调。

代码实现：

#pragmaonce

#ifndefMESSAGE_QUEUE_H
#defineMESSAGE_QUEUE_H

#include
#include
#include

template
classCMessageQueue
{
public:
CMessageQueue&operator=(constCMessageQueue&)=delete;
CMessageQueue(constCMessageQueue&mq)=delete;


CMessageQueue():_queue(),_mutex(),_condition(){}
virtual~CMessageQueue(){}

voidPush(Typemsg){
std::lock_guardlock(_mutex);
_queue.push(msg);
//当使用阻塞模式从消息队列中获取消息时，由condition在新消息到达时提醒等待线程
_condition.notify_one();
}
//blocked定义访问方式是同步阻塞或者非阻塞模式
boolPop(Type&msg,boolisBlocked=true){
if(isBlocked)
{
std::unique_locklock(_mutex);
while(_queue.empty())
{
_condition.wait(lock);

}
//注意这一段必须放在if语句中，因为lock的生命域仅仅在if大括号内
msg=std::move(_queue.front());
_queue.pop();
returntrue;

}
else
{
std::lock_guardlock(_mutex);
if(_queue.empty())
returnfalse;


msg=std::move(_queue.front());
_queue.pop();
returntrue;
}

}

int32_tSize(){
std::lock_guardlock(_mutex);
return_queue.size();
}

boolEmpty(){
std::lock_guardlock(_mutex);
return_queue.empty();
}
private:
std::queue_queue;//存储消息的队列
mutablestd::mutex_mutex;//同步锁
std::condition_variable_condition;//实现同步式获取消息
};

#endif//MESSAGE_QUEUE_H

线程池可以直接在构造函数中构造线程，并传入回调函数，也可以写一个Run函数显示调用。这里我们选择了第二种，对比：

在handler函数外部做循环接受消息，当消息到达后调用hanlder处理。这种实现在上层做封装，但是会在线程中频繁的切换调用函数。这种设计无法复用一些资源，如当在handler中做数据库操作时，需要频繁的连接和断开连接，可以通过定义两个虚函数Prehandler和AfterHandler来实现。
！！！构造函数中调用虚函数并不会能真正的调用子类的实现！！！
虽然可以对虚函数进行实调用，但程序员编写虚函数的本意应该是实现动态联编。在构造函数中调用虚函数，函数的入口地址是在编译时静态确定的，并未实现虚调用
写一个Run函数，将这一部分实现放在run函数中，显示调用。
《EffectiveC++》条款9：永远不要在构造函数或析构函数中调用虚函数

#ifndefTHREAD_POOL_H
#defineTHREAD_POOL_H


#include
#include
#include

#include"MessageQueue.h"

#defineMIN_THREADS1

template
classCThreadPool
{
CThreadPool&operator=(constCThreadPool&)=delete;
CThreadPool(constCThreadPool&other)=delete;

public:
CThreadPool(int32_tthreads,
std::function*pSub)>handler);
virtual~CThreadPool();

voidRun();
virtualvoidPreHandler(){}
virtualvoidAfterHandler(){}
voidSubmit(Typerecord);


private:
bool_shutdown;
int32_t_threads;
std::function*pSub)>_handler;
std::vector_workers;
CMessageQueue_tasks;

};



template
CThreadPool::CThreadPool(int32_tthreads,
std::function*pSub)>handler)
:_shutdown(false),
_threads(threads),
_handler(handler),
_workers(),
_tasks()
{

//第一种实现方案，注意这里的虚函数调用不正确
/*if(_threads
voidCThreadPool::Run()
{
if(_threads
CThreadPool::~CThreadPool()
{
for(std::thread&worker:_workers)
worker.join();
}


template
voidCThreadPool::Submit(Typerecord)
{
_tasks.Push(record);
}
#endif//!THREAD_POOL_H

总结

以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，谢谢大家对毛票票的支持。

声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。

C++基于消息队列的多线程实现示例代码

热门推荐

随机推荐