C++實現(xiàn)一個簡單消息隊列的示例詳解

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

消息隊列在多線程的場景有時會用到，尤其是線程通信跨線程調(diào)用的時候，就可以使用消息隊列進行通信。本文將利用C++實現(xiàn)一個簡單的消息隊列，感興趣的可以了解一下

前言

消息隊列在多線程的場景有時會用到，尤其是線程通信跨線程調(diào)用的時候，就可以使用消息隊列進行通信。C++實現(xiàn)一個能用的消息隊列還是比較簡單的，只需要一個隊列一個互斥變量和一個條件變量，這些在標準庫中都有提供。基于曾經(jīng)寫過的項目，總結(jié)出來最簡單的消息隊列的實現(xiàn)將在下文中介紹。

一、如何實現(xiàn)

1、接口定義

一個基本的消息隊列只需要3個接口：

（1）推送消息

推送消息即是將消息寫入隊列，這個通常采用異步實現(xiàn)，推送之后立刻返回。如果要實現(xiàn)Windows的SendMessage則會比較復雜，最好的方式是放到外部實現(xiàn)，消息隊列只提供異步推送消息。

void push(const T& msg);

（2）等待消息

等待隊列的消息，這個方法是同步的，只有接收到消息才會返回。

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

（3）輪詢消息

輪詢消息和等待消息一樣也是接收消息，只是無論是否接收到消息輪詢消息會立刻返回。輪詢消息也是有一定的使用場景，尤其是接收消息線程需要一定的調(diào)度邏輯時就需要輪詢消息避免線程堵塞。

bool poll(T& msg);

2、用到的對象

（1）隊列

我們使用一個隊列來存放消息

#include<queue>
std::queue<T> _queue;

（2）互斥變量

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

#include<mutex>
std::mutex _mtx;

（3）條件變量

采用條件變量結(jié)合互斥變量實現(xiàn)消息的等待和通知。

#include<condition_variable>
std::condition_variable _cv;

3、基本流程

線程通信

二、完整代碼

采用泛型實現(xiàn)，消息類型可以自定義。

#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;
}