c++消息队列

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

class MessageQueue {
public:
    void enqueue(int message) {
        std::lock_guard<std::mutex> lock(mtx);
        queue.push(message);
        cond.notify_one(); // 通知等待的线程有新消息
    }

    int dequeue() {
        std::unique_lock<std::mutex> lock(mtx);
        cond.wait(lock, [this] { return !queue.empty(); }); // 等待直到队列不为空
        int message = queue.front();
        queue.pop();
        return message;
    }

private:
    std::queue<int> queue;
    std::mutex mtx;
    std::condition_variable cond;
};

void producer(MessageQueue& mq, int num_messages) {
    for (int i = 0; i < num_messages; ++i) {
        std::cout << "Producing message " << i << std::endl;
        mq.enqueue(i);
        std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 模拟生产时间
    }
}

void consumer(MessageQueue& mq, int num_messages) {
    for (int i = 0; i < num_messages; ++i) {
        int message = mq.dequeue();
        std::cout << "Consuming message " << message << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds(150)); // 模拟消费时间
    }
}

int main() {
    MessageQueue mq;
    const int num_messages = 10;

    std::thread t1(producer, std::ref(mq), num_messages);
    std::thread t2(consumer, std::ref(mq), num_messages);

    t1.join();
    t2.join();

    return 0;
}

解释说明

1. MessageQueue 类:

   • enqueue(int message)：将消息加入队列，并通知等待的线程。
   • dequeue()：从队列中取出消息，如果队列为空则等待。

2. 线程同步机制:

   • 使用 std::mutex 和 std::lock_guard 来确保对共享资源（队列）的互斥访问。
   • 使用 std::condition_variable 来实现线程间的通信，当有新消息时通知等待的消费者线程。

3. 生产者和消费者线程:

   • producer 函数模拟生产消息并将其加入队列。
   • consumer 函数从队列中取出消息并处理。

4. 主函数:

   • 创建一个 MessageQueue 实例。
   • 启动两个线程分别执行生产和消费操作。
   • 使用 join 确保主线程等待子线程完成。