C++怎么实现一个线程安全的队列_C++多线程安全队列实现思路与示例

冰火之心 2025-10-31 00:00:00 次阅读

线程安全队列通过互斥锁和条件变量实现，确保多线程下数据同步；push插入元素并通知等待线程，wait_and_pop阻塞等待非空，try_pop提供非阻塞尝试，empty和size返回队列状态，适用于生产者-消费者模型。

在C++多线程编程中，线程安全的队列是常见的需求，比如生产者-消费者模型。要实现一个线程安全的队列，关键在于保护共享数据不被多个线程同时访问或修改。通常的做法是结合标准库中的 std::queue、std::mutex 和 std::condition_variable 来实现线程同步。

基本思路

实现线程安全队列的核心目标是：

多个线程可以安全地入队（push）和出队（pop）元素
当队列为空时，消费者线程应阻塞等待，直到有新元素加入
使用互斥锁保护队列的内部状态
使用条件变量通知等待线程数据已就绪

线程安全队列实现示例

#include 
#include 
#include 
#include 
#include 
#include 

template
class ThreadSafeQueue {
private:
    std::queue data_queue;
    mutable std::mutex mtx;
    std::condition_variable cv;

public:
    ThreadSafeQueue() = default;

    void push(T value) {
        std::lock_guard lock(mtx);
        data_queue.push(std::move(value));
        cv.notify_one(); // 唤醒一个等待的线程
    }

    bool try_pop(T& value) {
        std::lock_guard lock(mtx);
        if (data_queue.empty()) {
            return false;
        }
        value = std::move(data_queue.front());
        data_queue.pop();
        return true;
    }

    void wait_and_pop(T& value) {
        std::unique_lock lock(mtx);
        cv.wait(lock, [this] { return !data_queue.empty(); });
        value = std::move(data_queue.front());
        data_queue.pop();
    }

    bool empty() const {
        std::lock_guard lock(mtx);
        return data_queue.empty();
    }

    size_t size() const {
        std::lock_guard lock(mtx);
        return data_queue.size();
    }
};

使用示例：生产者-消费者模型

void producer(ThreadSafeQueue& queue) {
    for (int i = 0; i < 5; ++i) {
        queue.push(i);
        std::cout << "Produced: " << i << "\n";
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
}

void consumer(ThreadSafeQueue& queue) {
    for (int i = 0; i < 5; ++i) {
        int value;
        queue.wait_and_pop(value);
        std::cout << "Consumed: " << value << "\n";
    }
}

int main() {
    ThreadSafeQueue queue;
    std::vector threads;

    threads.emplace_back(producer, std::ref(queue));
    threads.emplace_back(consumer, std::ref(queue));

    for (auto& t : threads) {
        t.join();
    }

    return 0;
}