生产者消费者c++ 讲解和代码示例

生产者-消费者问题的C++讲解和代码示例

一、问题描述

生产者-消费者问题是经典的多线程同步问题，涉及两个类型的线程：

生产者线程：负责生成数据并放入共享缓冲区。
消费者线程：负责从共享缓冲区取出数据进行处理。

关键挑战在于：

同步：确保生产者和消费者在访问共享缓冲区时不发生冲突。
互斥：防止多个线程同时修改缓冲区导致数据不一致。
避免死锁：设计合理的等待和通知机制，防止线程无限期地等待。

二、解决方案

在C++中，可以使用以下同步机制：

std::mutex：互斥锁，用于保护共享数据的访问。
std::condition_variable：条件变量，用于线程间的等待和通知。
std::unique_lock<std::mutex>：配合条件变量使用的锁。

三、代码示例

下面是一个使用C++11线程库实现的生产者-消费者模型。

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <chrono>std::mutex mtx; // 互斥锁
std::condition_variable cv; // 条件变量
std::queue<int> buffer; // 共享缓冲区
const unsigned int MAX_BUFFER_SIZE = 10; // 缓冲区最大容量void producer(int id) {int data = 0;while (true) {// 模拟生产数据的时间std::this_thread::sleep_for(std::chrono::milliseconds(100));std::unique_lock<std::mutex> lock(mtx);// 等待缓冲区未满cv.wait(lock, []() { return buffer.size() < MAX_BUFFER_SIZE; });// 生产数据并放入缓冲区buffer.push(data);std::cout << "生产者 " << id << " 生产了数据 " << data << std::endl;data++;// 通知消费者cv.notify_all();}
}void consumer(int id) {while (true) {std::unique_lock<std::mutex> lock(mtx);// 等待缓冲区不为空cv.wait(lock, []() { return !buffer.empty(); });// 从缓冲区取出数据int data = buffer.front();buffer.pop();std::cout << "消费者 " << id << " 消费了数据 " << data << std::endl;// 通知生产者cv.notify_all();// 模拟处理数据的时间lock.unlock(); // 解锁以允许生产者继续生产std::this_thread::sleep_for(std::chrono::milliseconds(150));}
}int main() {std::thread producers[2], consumers[2];// 启动生产者线程for (int i = 0; i < 2; ++i) {producers[i] = std::thread(producer, i);}// 启动消费者线程for (int i = 0; i < 2; ++i) {consumers[i] = std::thread(consumer, i);}// 等待线程完成（此示例中线程是无限循环，可根据需要修改）for (int i = 0; i < 2; ++i) {producers[i].join();consumers[i].join();}return 0;
}

四、代码解析

1. 全局变量

互斥锁 mtx：保护对共享缓冲区的访问，防止数据竞争。
条件变量 cv：用于线程间的等待和通知机制。
共享缓冲区 buffer：存放生产者生成的数据，供消费者消费。
MAX_BUFFER_SIZE：限制缓冲区的最大容量，防止过度填充。

2. 生产者函数

void producer(int id) {int data = 0;while (true) {// 模拟生产时间std::this_thread::sleep_for(std::chrono::milliseconds(100));std::unique_lock<std::mutex> lock(mtx);// 等待缓冲区未满cv.wait(lock, []() { return buffer.size() < MAX_BUFFER_SIZE; });// 放入数据buffer.push(data);std::cout << "生产者 " << id << " 生产了数据 " << data << std::endl;data++;// 通知可能等待的消费者cv.notify_all();}
}

使用unique_lock获取互斥锁，确保对缓冲区的独占访问。
使用cv.wait等待缓冲区有空间（未满）。
生产数据并放入缓冲区。
使用cv.notify_all通知等待的消费者线程。

3. 消费者函数

void consumer(int id) {while (true) {std::unique_lock<std::mutex> lock(mtx);// 等待缓冲区不为空cv.wait(lock, []() { return !buffer.empty(); });// 取出数据int data = buffer.front();buffer.pop();std::cout << "消费者 " << id << " 消费了数据 " << data << std::endl;// 通知可能等待的生产者cv.notify_all();// 模拟处理数据的时间lock.unlock(); // 释放锁std::this_thread::sleep_for(std::chrono::milliseconds(150));}
}

获取互斥锁，确保对缓冲区的独占访问。
使用cv.wait等待缓冲区有数据（不为空）。
消费数据并从缓冲区移除。
使用cv.notify_all通知等待的生产者线程。
在处理数据时释放锁，允许其他线程访问缓冲区。

4. 主函数

int main() {std::thread producers[2], consumers[2];// 启动生产者和消费者线程for (int i = 0; i < 2; ++i) {producers[i] = std::thread(producer, i);consumers[i] = std::thread(consumer, i);}// 等待线程完成（无限循环，实际应用中可添加终止条件）for (int i = 0; i < 2; ++i) {producers[i].join();consumers[i].join();}return 0;
}