C++线程安全队列实现详解

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

template<typename T>
class ThreadSafeQueue {
private:
    std::queue<T> data_queue;
    mutable std::mutex mtx;
    std::condition_variable cv;

public:
    ThreadSafeQueue() = default;

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

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

    std::shared_ptr<T> pop() {
        std::unique_lock<std::mutex> lock(mtx);
        cv.wait(lock, [this]{ return !data_queue.empty(); });
        auto result = std::make_shared<T>(std::move(data_queue.front()));
        data_queue.pop();
        return result;
    }

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

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

#include <iostream>
#include <thread>

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

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

int main() {
    ThreadSafeQueue<int> queue;
    std::thread p(producer, std::ref(queue));
    std::thread c(consumer, std::ref(queue));

    p.join();
    c.join();

    return 0;
}