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

该文实现了基于C++标准库的线程安全阻塞队列模板类BlockingQueue，支持容量限制、阻塞入队/出队及超时尝试出队；配套生产者生成随机数、消费者取数处理，并通过多线程协同演示完整生产者-消费者模型。

（以下为纯 C++ 多线程阻塞队列 + 生产者消费者模型实现，基于标准库、、和，无第三方依赖，可直接编译运行）

核心思路：用互斥锁保护共享队列，用条件变量让空队列时消费者等待、满队列时生产者等待。C++ 标准库的 std::condition_variable 天然支持“等待某个条件成立”，配合 wait() 和 notify_one()/notify_all() 即可实现线程安全的阻塞行为。

1. 线程安全的阻塞队列模板类

支持任意类型 T，可设容量上限（可选），内部自动处理 wait/notify 逻辑：

#include 
#include 
#include 
#include 

template
class BlockingQueue {
private:
    std::queue 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 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 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 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 lock(mtx);
        return q.size();
    }

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

2. 生产者与消费者函数（分离职责）

每个生产者/消费者都是独立函数，通过引用使用同一个队列，适合传给 std::thread：

#include 
#include 
#include 
#include 

// 生产者：生成随机数并入队
void producer(BlockingQueue& bq, int id, int count) {
    std::mt19937 gen(id);
    std::uniform_int_distribution 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& 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));
    }
}

3. 主函数：启动多线程并协同运行

注意：需确保生产总数 ≥ 消费总数，否则消费者会永久阻塞（实际项目中建议加退出信号或使用 `try_pop` + 超时 + 中断机制）：

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

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

    std::vector 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;
}

4. 关键细节提醒

始终用 std::unique_lock 配合 wait：它允许在等待期间临时释放锁，并在唤醒后自动重新加锁；普通 lock_guard 不支持
条件检查必须用 lambda（谓词重载）：避免虚假唤醒（spurious wakeup），不要只用 wait(lock) 后手动判断
notify_one() 通常够用：除非多个线程等待同一条件且需全部响应（如广播终止信号），否则用 notify_one() 更高效
移动语义优化：pop 中用 std::move(q.front()) 避免不必要的拷贝（尤其对大对象）
异常安全：所有 RAII 锁（unique_lock / lock_guard）保证异常时自动解锁