C++实现简单消息队列：核心原理、代码与使用详解

一、如何实现简单的消息队列

一个简单的消息队列通常只需提供三个核心操作：推送消息、等待消息（阻塞模式）以及轮询消息（非阻塞模式）。其内部结构依赖 C++ 标准库提供的原子操作原语来保证数据安全性。

第1步：定义接口行为

为了保证接口的通用性与易用性，消息队列应包含以下关键方法：

• 推入消息（Push）：将数据写入队列并唤醒等待者。此操作应为异步过程，调用后即刻返回，无需等待消费端处理。
• 轮询消息（Poll）：检查是否有可用数据。无论是否读取成功均立刻返回，适用于需要保持主线程循环不卡顿的场景。
• 等待消息（Wait）：获取下一条消息。如果队列为空，当前执行流将被挂起，直到有新消息加入并触发唤醒信号。

第2步：配置基础对象与资源

底层实现主要涉及三种标准类型的数据结构：

1. 数据存储：<queue>：利用 std::queue<T> 维持 FIFO（先进先出）的顺序，存放待传递的对象。
2. 并发锁：<mutex>：通过 std::mutex 互斥量保护队列读写操作，防止多线程竞争导致的数据不一致或越界访问。
3. 条件变量：<condition_variable>：作为线程间的信号灯。生产者在推送数据后调用通知函数，让消费者从休眠状态苏醒；消费者则在队列为空时进入等待池。

线程之间通过这种机制交换信息的过程如下所示：

二、完整泛型实现代码

采用 C++ 模板技术可以使得该消息队列适配任意自定义类型。以下是完整的头文件风格实现代码：

#include <mutex>
#include <condition_variable>
#include <queue>

/// <summary>
/// 泛型消息队列类
/// </summary>
template <class T>
class MessageQueue {
public:
    /// <summary>
    /// 将消息推入队列尾部
    /// </summary>
    void push(const T& msg) {
        std::unique_lock<std::mutex> lck(_mtx);
        _queue.push(msg);
        _cv.notify_one(); // 唤醒一个处于 wait 状态的消费者
    }

    /// <summary>
    /// 尝试从队列头部取出消息（非阻塞）
    /// </summary>
    bool poll(T& msg) {
        std::unique_lock<std::mutex> lck(_mtx);
        if (!_queue.empty()) {
            msg = _queue.front();
            _queue.pop();
            return true;
        }
        return false;
    }

    /// <summary>
    /// 阻塞式获取消息，直到有数据到达
    /// </summary>
    void wait(T& msg) {
        std::unique_lock<std::mutex> lck(_mtx);
        // 循环内检测条件以防范虚假唤醒
        while (_queue.empty()) {
            _cv.wait(lck);
        }
        msg = _queue.front();
        _queue.pop();
    }

private:
    std::queue<T> _queue;
    std::mutex _mtx;
    std::condition_variable _cv;
};

三、实际应用场景演示

下面展示了如何在 `main` 函数及子线程中集成上述模板类，涵盖常用的两种消费逻辑。

场景A：线程间等待通信（阻塞模型）

当业务逻辑允许当前线程暂停直至获得指令时使用。代码如下：

#include <thread>
#include <iostream>
using namespace std;

// 假设已定义上文中的 MessageQueue 模板类

struct MyMessage {
    int type;
    void* param1 = nullptr;
    void* param2 = nullptr;
};

int main() {
    MessageQueue<MyMessage> mq;

    // 工作线程：持续阻塞等待新任务
    thread t1([&]() {
        MyMessage msg;
        while (true) {
            mq.wait(msg); // 此处会暂停直到 push 被调用
            cout << "Receive message type:" << msg.type << endl;
            if (msg.type == 1001)
                break;      // 收到退出信号
        }
        cout << "Thread exited." << endl;
    });

    // 主控线程：模拟外部输入推送事件
    MyMessage msg; msg.type = 0;
    cout << "Send '1001' to stop..." << endl;
    while (true) {
        cin >> msg.type;
        mq.push(msg);
        if (msg.type == 1001) break;
    }
    t1.join();
    return 0;
}

场景B：高频轮询调度（非阻塞模型）

对于实时游戏引擎或高频交易等不能轻易丢下线程的执行环境，建议搭配短暂的休眠进行轮询：

// ... 省略头文件和类定义 ...

int main() {
    MessageQueue<MyMessage> mq;

    thread t1([&]() {
        MyMessage msg;
        while (true) {
            // 如果 poll 返回 false 表示队列为空，此时不应阻塞
            if (mq.poll(msg)) {
                cout << "Polling received type:" << msg.type << endl;
                if (msg.type == 1001) break;
            } else {
                // 避免空转消耗 CPU 算力
                this_thread::sleep_for(chrono::milliseconds(10));
            }
        }
    });

    // 同样由外部线程推送消息...
    MyMessage msg; 
    // ... (推送逻辑同上)
}

以上就是使用 C++ 搭建简易异步消息队列的详细内容，更多关于 C++ 并发编程与队列优化的资料请关注本站其它相关文章！