快速导航×

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

template <typename T>
class BlockingQueue {
private:
    std::queue<T> data_queue_;
    std::mutex mtx_;
    std::condition_variable not_full_;
    std::condition_variable not_empty_;
    size_t max_size_;

public:
    explicit BlockingQueue(size_t max_size = 1000) : max_size_(max_size) {}

    void put(T item) {
        std::unique_lock<std::mutex> lock(mtx_);
        // 队列满时等待
        not_full_.wait(lock, [this] { return data_queue_.size() < max_size_; });
        data_queue_.push(std::move(item));
        not_empty_.notify_one(); // 通知消费者
    }

    T take() {
        std::unique_lock<std::mutex> lock(mtx_);
        // 队列空时等待
        not_empty_.wait(lock, [this] { return !data_queue_.empty(); });
        T item = std::move(data_queue_.front());
        data_queue_.pop();
        not_full_.notify_one(); // 通知生产者
        return item;
    }

    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>

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

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

int main() {
    BlockingQueue<int> queue(3); // 容量为3

    std::thread p1(producer, std::ref(queue));
    std::thread c1(consumer, std::ref(queue));

    p1.join();
    c1.join();

    return 0;
}