当前位置：首页 > news >正文

【C++】【设计模式】生产者-消费者模型

news 2025/5/14 7:02:57

生产者-消费者模型（Producer-Consumer Model）是一种经典的并发编程模式，用于解决多线程或多进程环境下的数据共享和任务协作问题。以下是对该模型的详细介绍：

一、核心概念

生产者（Producer）：负责生成数据或任务，并将其放入共享缓冲区。
消费者（Consumer）：从共享缓冲区中获取数据或任务并处理。
缓冲区（Buffer）：线程安全的队列，用于临时存储数据，解耦生产者和消费者。
同步机制：确保缓冲区在多线程环境下的正确性（如互斥锁、条件变量）。

二、实现方式

1. 基于条件变量的实现

#include <iostream>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <thread>template<typename T>
class BlockingQueue {
private:std::queue<T> queue_;mutable std::mutex mutex_;std::condition_variable cv_not_full_;std::condition_variable cv_not_empty_;size_t max_size_;public:explicit BlockingQueue(size_t max_size = 10) : max_size_(max_size) {}void push(const T& value) {std::unique_lock<std::mutex> lock(mutex_);cv_not_full_.wait(lock, [this]{ return queue_.size() < max_size_; });queue_.push(value);cv_not_empty_.notify_one();}T pop() {std::unique_lock<std::mutex> lock(mutex_);cv_not_empty_.wait(lock, [this]{ return !queue_.empty(); });T value = queue_.front();queue_.pop();cv_not_full_.notify_one();return value;}
};// 示例使用
void producer_consumer_example() {BlockingQueue<int> queue(5);  // 最大容量为5// 生产者线程auto producer = [&]() {for (int i = 0; i < 10; ++i) {queue.push(i);std::cout << "Produced: " << i << std::endl;std::this_thread::sleep_for(std::chrono::milliseconds(100));}};// 消费者线程auto consumer = [&]() {for (int i = 0; i < 10; ++i) {int value = queue.pop();std::cout << "Consumed: " << value << std::endl;std::this_thread::sleep_for(std::chrono::milliseconds(200));}};std::thread t1(producer);std::thread t2(consumer);t1.join();t2.join();
}

2. 基于`std::atomic`的无锁实现（简化版）

#include <atomic>
#include <array>template<typename T, size_t N>
class LockFreeQueue {
private:std::array<T, N> buffer_;std::atomic<size_t> head_ = {0};std::atomic<size_t> tail_ = {0};static size_t next(size_t idx) { return (idx + 1) % N; }public:bool push(const T& value) {size_t tail = tail_.load(std::memory_order_relaxed);size_t next_tail = next(tail);if (next_tail == head_.load(std::memory_order_acquire))return false;  // 队列已满buffer_[tail] = value;tail_.store(next_tail, std::memory_order_release);return true;}bool pop(T& value) {size_t head = head_.load(std::memory_order_relaxed);if (head == tail_.load(std::memory_order_acquire))return false;  // 队列为空value = buffer_[head];head_.store(next(head), std::memory_order_release);return true;}
};

三、关键组件详解

1. 缓冲区设计

有界缓冲区：设置最大容量，防止内存溢出。
无界缓冲区：理论上可无限扩展（如std::list），但需警惕内存耗尽。

2. 同步机制

条件变量：生产者等待缓冲区非满，消费者等待缓冲区非空。
原子操作：用于无锁队列，通过内存屏障保证可见性。

3. 线程协作

一对一：一个生产者和一个消费者。
多对一/一对多：多个生产者或消费者共享缓冲区。
多对多：最复杂的场景，需严格控制同步。

四、应用场景

任务队列：Web服务器将HTTP请求放入队列，工作线程处理请求。
数据处理流水线：如视频编码（采集→编码→渲染）。
事件驱动系统：生产者发布事件，消费者监听并响应。
内存池/对象池：预先创建对象放入队列，避免频繁分配内存。

五、常见问题与解决方案

问题	解决方案
死锁	统一加锁顺序，使用`std::lock`同时锁定多个互斥锁。
虚假唤醒	在条件变量等待时使用谓词（如`cv.wait(lock, []{ return !queue.empty(); })`）。
内存可见性	使用原子操作或同步原语确保修改对其他线程可见。
性能瓶颈	采用无锁算法（如CAS操作）或分区锁减少竞争。