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

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

什么是阻塞队列

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

核心实现要点

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

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

无界阻塞队列示例(常用场景)

下面是一个简洁、可直接运行的无界阻塞队列实现(支持 move 语义,线程安全):

#include 
#include 
#include 
#include 

template class BlockingQueue { private: std::queue queue; mutable std::mutex mtx; std::condition_variable notempty; std::condition_variable notfull; // 可选,无界时仅作占位

public: void push(T item) { std::uniquelock lock(mtx); queue_.push(std::move(item)); notempty.notify_one(); // 唤醒等待消费的线程 }

T pop() {
    std::unique_lockzuojiankuohaophpcnstd::mutexyoujiankuohaophpcn 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_lockzuojiankuohaophpcnstd::mutexyoujiankuohaophpcn 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_lockzuojiankuohaophpcnstd::mutexyoujiankuohaophpcn lock(mtx_);
    return queue_.empty();
}

};

生产者-消费者完整实例

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

#include 
#include 
#include 
#include 

BlockingQueue bq;

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)); } }

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; } } }

int main() { std::vector producers, consumers;

// 启动 2 个生产者,各发 3 个数
for (int i = 0; i zuojiankuohaophpcn 2; ++i) {
    producers.emplace_back(producer, i, 3);
}

// 启动 3 个消费者
for (int i = 0; i zuojiankuohaophpcn 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。输出会交错显示生产和消费过程,体现线程间自然同步。

注意事项与优化建议

实际工程中还需考虑:

  • 若需有界队列,构造时传入最大容量,在 push() 中检查并用 not_full_.wait() 阻塞
  • 增加 stop() 接口配合 std::atomic_bool 实现优雅退出(避免消费者无限等待)
  • 对频繁操作场景,可用 std::deque 替代 std::queue 获得更好缓存局部性
  • 避免在锁内做耗时操作(如 I/O、复杂计算),保持临界区尽量短