C++实现一个简单消息队列的示例详解

消息队列在多线程的场景有时会用到，尤其是线程通信跨线程调用的时候，就可以使用消息队列进行通信。本文将利用C++实现一个简单的消息队列，感兴趣的可以了解一下

前言

消息队列在多线程的场景有时会用到，尤其是线程通信跨线程调用的时候，就可以使用消息队列进行通信。C++实现一个能用的消息队列还是比较简单的，只需要一个队列一个互斥变量和一个条件变量，这些在标准库中都有提供。基于曾经写过的项目，总结出来最简单的消息队列的实现将在下文中介绍。

一、如何实现

1、接口定义

一个基本的消息队列只需要3个接口：

（1）推送消息

推送消息即是将消息写入队列，这个通常采用异步实现，推送之后立刻返回。如果要实现Windows的SendMessage则会比较复杂，最好的方式是放到外部实现，消息队列只提供异步推送消息。

void push(const T& msg);

（2）等待消息

等待队列的消息，这个方法是同步的，只有接收到消息才会返回。

//等待消息 void wait(T& msg);

（3）轮询消息

轮询消息和等待消息一样也是接收消息，只是无论是否接收到消息轮询消息会立刻返回。轮询消息也是有一定的使用场景，尤其是接收消息线程需要一定的调度逻辑时就需要轮询消息避免线程堵塞。

bool poll(T& msg);

2、用到的对象

（1）队列

我们使用一个队列来存放消息

#include std::queue _queue;

（2）互斥变量

使用一个互斥变量确保队列的读写线程安全

#include std::mutex _mtx;

（3）条件变量

采用条件变量结合互斥变量实现消息的等待和通知。

#include std::condition_variable _cv;

3、基本流程

线程通信

二、完整代码

采用泛型实现，消息类型可以自定义。

#include #include #include ///  /// 消息队列 /// 
 /// 消息类型 template class MessageQueue { public: ///  /// 推入消息 /// 
 /// 消息对象 void push(const T& msg) { std::unique_locklck(_mtx); _queue.push(msg); _cv.notify_one(); } ///  /// 轮询消息 /// 
 /// 消息对象 /// 是否接收到消息 bool poll(T& msg) { std::unique_locklck(_mtx); if (_queue.size()) { msg = _queue.front(); _queue.pop(); return true; } return false; } ///  /// 等待消息 /// 
 /// 消息对象 void wait(T& msg) { std::unique_locklck(_mtx); while (!_queue.size()) _cv.wait(lck); msg = _queue.front(); _queue.pop(); } //队列长度 size_t size() { std::unique_locklck(_mtx); return _queue.size(); } private: //队列 std::queue _queue; //互斥变量 std::mutex _mtx; //条件变量 std::condition_variable _cv; };

三、使用示例

线程通信

等待消息

#include //自定义消息对象 class MyMessage { public: int type; void* param1; void* param2; }; int main(int argc, char* argv[]) { //初始化消息队列 MessageQueue mq; //启动线程 std::thread t1([&]() { MyMessage msg; while (1) { //等待队列的消息 mq.wait(msg); printf("receive message type:%d\n", msg.type); if (msg.type == 1001) break; } printf("thread exited\n"); }); //发送消息给线程 MyMessage msg; printf("send number message to thread.1001 exit\n"); while (1) { scanf("%d", &msg.type); mq.push(msg); if (msg.type == 1001) break; } t1.join(); return 0; }

轮询消息

#include //自定义消息对象 class MyMessage { public: int type; void* param1; void* param2; }; int main(int argc, char* argv[]) { //初始化消息队列 MessageQueue mq; //启动线程 std::thread t1([&]() { MyMessage msg; while (1) { //轮询队列的消息 if (mq.poll(msg)) { printf("receive message type:%d\n", msg.type); if (msg.type == 1001) break; } std::this_thread::sleep_for(std::chrono::milliseconds(10)); } printf("thread exited\n"); }); //发送消息给线程 MyMessage msg; printf("send number message to thread.1001 exit\n"); while (1) { scanf("%d", &msg.type); mq.push(msg); if (msg.type == 1001) break; } t1.join(); return 0; }