C++怎样实现简单的生产者消费者流水线模型

作者：袖梨 2026-06-23

用std::queue+mutex+condition_variable实现线程安全流水线：生产者检查容量并wait空位，消费者检查空队列并wait数据，均用while循环防虚假唤醒，notify_one避免性能浪费；支持模板与函数对象泛化类型和处理逻辑；必要时可用deque替代queue以支持peek等操作。

用 `std::queue` + `std::mutex` + `std::condition_variable` 实现基础流水线

核心就是让生产者往队列塞数据、消费者从队列取数据，靠互斥锁保线程安全，靠条件变量避免忙等。别直接裸用 std::queue——它不是线程安全的，不加锁会崩。

关键点：队列为空时消费者必须等待；队列满时生产者最好也等（否则可能无限 push 导致 OOM）。示例里用固定容量 10 的队列模拟“流水线缓冲区”：

std::queue<int> q;std::mutex mtx;std::condition_variable cv_not_empty, cv_not_full;const size_t CAPACITY = 10;

生产者逻辑里先锁、检查容量、wait 等空位；消费者类似，但检查是否为空。注意 cv_not_full.wait() 必须传入 std::unique_lock，且条件判断要用 lambda 防止虚假唤醒。

`notify_one()` 还是 `notify_all()`？选错会卡死或性能差

生产者 push 完一个元素后，只唤醒一个等待的消费者就够了——用 notify_one()。同理，消费者 pop 后只唤醒一个生产者。用 notify_all() 会导致所有等待线程争抢锁，尤其在高并发下浪费 CPU。

立即学习“C++免费学习笔记（深入）”；

但有个坑：如果用了 notify_one() 却没有对应等待者（比如消费者已退出），这次通知就丢了。所以实际中更稳妥的做法是——只要状态变了就 notify，靠 while 循环重检条件，而不是依赖“刚好唤醒对的人”。

生产者末尾调 cv_not_empty.notify_one()
消费者末尾调 cv_not_full.notify_one()
永远用 while (q.empty()) cv_not_empty.wait(lk)，不用 if

如何让流水线支持多种数据类型和处理函数？用模板 + 函数对象封装

硬编码 int 没法复用。把队列类型、生产逻辑、消费逻辑都泛化：

template <typename T>class Pipeline {    std::queue<T> q;    std::mutex mtx;    std::condition_variable cv_not_empty, cv_not_full;    size_t capacity_;public:    Pipeline(size_t cap) : capacity_(cap) {}    void produce(T item);    T consume();};

再进一步，把“怎么生成”“怎么处理”抽成可调用对象：