C++實現一個簡單消息隊列的示例詳解

更新時間：2022年12月15日 11:02:57 作者：CodeOfCC

消息隊列在多線程的場景有時會用到，尤其是線程通信跨線程調用的時候，就可以使用消息隊列進行通信。本文將利用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<queue>
std::queue<T> _queue;

（2）互斥變量

使用一個互斥變量確保隊列的讀寫線程安全

#include<mutex>
std::mutex _mtx;

（3）條件變量

采用條件變量結合互斥變量實現消息的等待和通知。

#include<condition_variable>
std::condition_variable _cv;

3、基本流程

線程通信

二、完整代碼

采用泛型實現，消息類型可以自定義。

#include<mutex>
#include<condition_variable>
#include<queue>
/// <summary>
/// 消息隊列
/// </summary>
/// <typeparam name="T">消息類型</typeparam>
template<class T> class MessageQueue {
public:
    /// <summary>
    /// 推入消息
    /// </summary>
    /// <param name="msg">消息對象</param>
    void push(const T& msg) {
        std::unique_lock<std::mutex>lck(_mtx);
        _queue.push(msg);
        _cv.notify_one();
    }
    /// <summary>
    /// 輪詢消息
    /// </summary>
    /// <param name="msg">消息對象</param>
    /// <returns>是否接收到消息</returns>
    bool poll(T& msg) {
        std::unique_lock<std::mutex>lck(_mtx);
        if (_queue.size())
        {
            msg = _queue.front();
            _queue.pop();
            return true;
        }
        return false;
    }
    /// <summary>
    /// 等待消息
    /// </summary>
    /// <param name="msg">消息對象</param>
    void wait(T& msg) {
        std::unique_lock<std::mutex>lck(_mtx);
        while (!_queue.size()) _cv.wait(lck);
        msg = _queue.front();
        _queue.pop();
    }
    //隊列長度
    size_t size() {
        std::unique_lock<std::mutex>lck(_mtx);
        return _queue.size();
    }
private:
    //隊列
    std::queue<T> _queue;
    //互斥變量
    std::mutex _mtx;
    //條件變量
    std::condition_variable _cv;
};

三、使用示例

線程通信

等待消息

#include<thread>
//自定義消息對象
class MyMessage {
public:
    int type;
    void* param1;
    void* param2;
};
int main(int argc, char* argv[])
{
    //初始化消息隊列
    MessageQueue<MyMessage> 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");
        });
    //發(fā)送消息給線程
    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<thread>
//自定義消息對象
class MyMessage {
public:
    int type;
    void* param1;
    void* param2;
};
int main(int argc, char* argv[])
{
    //初始化消息隊列
    MessageQueue<MyMessage> 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");
        });
    //發(fā)送消息給線程
    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;
}