多线程安全循环消息队列，互斥锁 + 条件变量，通用多生产多消费-c++

#include <iostream>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <chrono>

// 通用消息类型
struct Message {
    int msg_id;
    std::string content;
};

class MsgLoopQueue {
private:
    std::queue<Message> m_queue;
    std::mutex m_mtx;
    std::condition_variable m_cv;
    bool m_exit = false;

public:
    // 发送消息
    void send(const Message& msg) {
        std::lock_guard<std::mutex> lock(m_mtx);
        m_queue.push(msg);
        m_cv.notify_one(); // 唤醒等待线程
    }

    // 阻塞获取消息，返回false表示退出
    bool recv(Message& out_msg) {
        std::unique_lock<std::mutex> lock(m_mtx);
        // 队空且未退出则等待
        m_cv.wait(lock, [this]() {
            return !m_queue.empty() || m_exit;
        });
        if (m_exit) return false;

        out_msg = m_queue.front();
        m_queue.pop();
        return true;
    }

    // 停止消息循环
    void stop() {
        std::lock_guard<std::mutex> lock(m_mtx);
        m_exit = true;
        m_cv.notify_all();
    }
};

// 消息循环线程函数
void msg_loop(MsgLoopQueue& queue) {
    Message msg;
    std::cout << "消息循环线程启动\n";
    while (queue.recv(msg)) {
        // 业务处理消息
        std::cout << "处理消息 ID:" << msg.msg_id
                  << " 内容:" << msg.content << "\n";
    }
    std::cout << "消息循环退出\n";
}

int main() {
    MsgLoopQueue queue;
    // 启动消息循环线程
    std::thread loop_thread(msg_loop, std::ref(queue));

    // 模拟主线程发送消息
    for (int i = 1; i <= 5; ++i) {
        queue.send({i, "测试消息" + std::to_string(i)});
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
    }

    // 等待处理完后停止
    std::this_thread::sleep_for(std::chrono::seconds(1));
    queue.stop();
    loop_thread.join();
    return 0;
}