C++實現一個簡單訊息佇列的範例詳解

2022-12-17 14:01:16

前言

訊息佇列在多執行緒的場景有時會用到，尤其是執行緒通訊跨執行緒呼叫的時候，就可以使用訊息佇列進行通訊。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:%dn", msg.type);
            if (msg.type == 1001)
                break;
        }
        printf("thread exitedn");
        });
    //傳送訊息給執行緒
    MyMessage msg;
    printf("send number message to thread.1001 exitn");
    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:%dn", msg.type);
                if (msg.type == 1001)
                    break;
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
        printf("thread exitedn");
        });
    //傳送訊息給執行緒
    MyMessage msg;
    printf("send number message to thread.1001 exitn");
    while (1)
    {
        scanf("%d", &msg.type);
        mq.push(msg);
        if (msg.type == 1001)
            break;
    }
    t1.join();
    return 0;
}