c++如何实现一个阻塞队列 c++生产者消费者模型【实例】-C++-PHP中文网

c++如何实现一个阻塞队列 c++生产者消费者模型【实例】

冰火之心

发布： 2025-12-22 16:21:28

原创

563人浏览过

c++kquote>阻塞队列是线程安全的队列，空时消费者pop自动等待，满时生产者push自动等待，天然适配生产者-消费者模型；核心需用std::deque、std::mutex和std::condition_variable实现。

c++如何实现一个阻塞队列 c++生产者消费者模型【实例】

什么是阻塞队列

阻塞队列是一种线程安全的队列，当队列为空时，消费者线程调用 pop() 会自动等待，直到有新元素入队；当队列为满时（如有容量限制），生产者线程调用 push() 也会等待，直到有空位。这种“自动等待+唤醒”机制，天然适配生产者-消费者模型。

核心实现要点

要用 C++ 实现一个可靠的阻塞队列，需结合以下三要素：

std::queue 或 std::deque 作为底层容器（推荐 deque，支持高效首尾操作）
std::mutex 保护共享数据，防止多线程并发访问冲突
std::condition_variable 实现线程挂起与唤醒：一个用于“非空”通知（消费者等数据），一个用于“非满”通知（生产者等空间）——若无容量限制，可只用一个条件变量

无界阻塞队列示例（常用场景）

下面是一个简洁、可直接运行的无界阻塞队列实现（支持 move 语义，线程安全）：

#include <queue>
#include <mutex>
#include <condition_variable>
#include <chrono>
<p>template <typename T>
class BlockingQueue {
private:
std::queue<T> queue<em>;
mutable std::mutex mtx</em>;
std::condition_variable not<em>empty</em>;
std::condition_variable not<em>full</em>; // 可选，无界时仅作占位</p><p>public:
void push(T item) {
std::unique<em>lock<std::mutex> lock(mtx</em>);
queue_.push(std::move(item));
not<em>empty</em>.notify_one(); // 唤醒等待消费的线程
}</p><pre class='brush:php;toolbar:false;'>T pop() {
    std::unique_lock<std::mutex> lock(mtx_);
    not_empty_.wait(lock, [this] { return !queue_.empty(); });
    T item = std::move(queue_.front());
    queue_.pop();
    return item;
}

// 带超时的 pop（避免永久阻塞）
bool pop(T& item, std::chrono::milliseconds timeout) {
    std::unique_lock<std::mutex> lock(mtx_);
    if (not_empty_.wait_for(lock, timeout, [this] { return !queue_.empty(); })) {
        item = std::move(queue_.front());
        queue_.pop();
        return true;
    }
    return false;
}

bool empty() const {
    std::unique_lock<std::mutex> lock(mtx_);
    return queue_.empty();
}

登录后复制

};

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

生产者-消费者完整实例

用上面的 BlockingQueue 启动多个生产者和消费者线程，模拟真实协作：

百度AI开放平台

百度提供的综合性AI技术服务平台，汇集了多种AI能力和解决方案

105

查看详情

#include <iostream>
#include <thread>
#include <vector>
#include <atomic>
<p>BlockingQueue<int> bq;</p><p>void producer(int id, int count) {
for (int i = 0; i < count; ++i) {
int val = id * 100 + i;
bq.push(val);
std::cout << "[P" << id << "] pushed " << val << "\n";
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
}</p><p>void consumer(int id) {
while (true) {
int val;
if (bq.pop(val, std::chrono::milliseconds(500))) {
std::cout << "[C" << id << "] consumed " << val << "\n";
} else {
std::cout << "[C" << id << "] timeout, exiting...\n";
break;
}
}
}</p><p>int main() {
std::vector<std::thread> producers, consumers;</p><pre class='brush:php;toolbar:false;'>// 启动 2 个生产者，各发 3 个数
for (int i = 0; i < 2; ++i) {
    producers.emplace_back(producer, i, 3);
}

// 启动 3 个消费者
for (int i = 0; i < 3; ++i) {
    consumers.emplace_back(consumer, i);
}

for (auto& t : producers) t.join();
for (auto& t : consumers) t.join();

return 0;

登录后复制

}

编译运行时加 -std=c++17 -pthread。输出会交错显示生产和消费过程，体现线程间自然同步。