c++怎么实现一个阻塞队列_c++生产者消费者模型实现【多线程】

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

template<typename T>
class BlockingQueue {
private:
    std::queue<T> q;
    mutable std::mutex mtx;
    std::condition_variable not_empty;   // 消费者等待：队列非空
    std::condition_variable not_full;    // 生产者等待：队列未满（若限容）
    size_t max_size = 0;                 // 0 表示无容量限制

public:
    explicit BlockingQueue(size_t capacity = 0) : max_size(capacity) {}

    // 入队（阻塞直到有空间）
    void push(const T& item) {
        std::unique_lock<std::mutex> lock(mtx);
        if (max_size > 0) {
            not_full.wait(lock, [this] { return q.size() < max_size; });
        }
        q.push(item);
        not_empty.notify_one(); // 唤醒一个等待消费的线程
    }

    // 出队（阻塞直到有数据）
    T pop() {
        std::unique_lock<std::mutex> lock(mtx);
        not_empty.wait(lock, [this] { return !q.empty(); });
        T front = std::move(q.front());
        q.pop();
        if (max_size > 0) {
            not_full.notify_one(); // 可能腾出空间，唤醒一个等待生产的线程
        }
        return front;
    }

    // 尝试出队（带超时，返回 false 表示超时或为空）
    bool try_pop(T& item, int timeout_ms = 0) {
        std::unique_lock<std::mutex> lock(mtx);
        if (timeout_ms == 0) {
            not_empty.wait(lock, [this] { return !q.empty(); });
        } else {
            auto dur = std::chrono::milliseconds(timeout_ms);
            if (!not_empty.wait_for(lock, dur, [this] { return !q.empty(); })) {
                return false;
            }
        }
        item = std::move(q.front());
        q.pop();
        if (max_size > 0) not_full.notify_one();
        return true;
    }

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

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

#include <iostream>
#include <thread>
#include <vector>
#include <random>

// 生产者：生成随机数并入队
void producer(BlockingQueue<int>& bq, int id, int count) {
    std::mt19937 gen(id);
    std::uniform_int_distribution<int> dist(1, 100);
    for (int i = 0; i < count; ++i) {
        int val = dist(gen);
        bq.push(val);
        std::cout << "[P" << id << "] pushed " << val << "\n";
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
}

// 消费者：取数、打印、简单处理
void consumer(BlockingQueue<int>& bq, int id, int total) {
    for (int i = 0; i < total; ++i) {
        int val = bq.pop(); // 阻塞等待
        std::cout << "  [C" << id << "] popped " << val << "\n";
        std::this_thread::sleep_for(std::chrono::milliseconds(150));
    }
}

int main() {
    BlockingQueue<int> bq(10); // 容量为 10 的有界队列

    const int num_producers = 2;
    const int num_consumers = 3;
    const int items_per_producer = 5;

    std::vector<std::thread> threads;

    // 启动生产者
    for (int i = 0; i < num_producers; ++i) {
        threads.emplace_back(producer, std::ref(bq), i, items_per_producer);
    }

    // 启动消费者（总消费数 = 总生产数）
    int total_items = num_producers * items_per_producer;
    for (int i = 0; i < num_consumers; ++i) {
        int each = total_items / num_consumers;
        int rest = (i == 0) ? total_items % num_consumers : 0;
        threads.emplace_back(consumer, std::ref(bq), i, each + rest);
    }

    // 等待所有线程结束
    for (auto& t : threads) {
        if (t.joinable()) t.join();
    }

    std::cout << "All done.\n";
    return 0;
}