多线程：生产者消费者

本文展示了一个基于C的多线程生产者-消费者模型实现。核心组件ThreadSafeQueue是一个线程安全的队列模板类使用互斥锁和条件变量实现同步机制支持阻塞式push/pop操作。系统包含3个生产者线程各生产10个产品和2个消费者线程通过随机延迟模拟实际生产消费过程。当所有生产者完成后系统会设置停止标志并等待消费者处理剩余产品最后输出运行统计信息总耗时6316毫秒成功处理30个产品。该实现演示了多线程环境下安全的数据交换和线程协同工作包括生产者阻塞队列满时、消费者阻塞队列空时以及优雅终止等关键机制。代码#include iostream #include queue #include thread #include mutex #include condition_variable #include chrono #include random #include atomic #include vector /** * brief 线程安全的队列类用作生产者和消费者之间的缓冲区 * * 使用互斥锁和条件变量实现线程安全支持阻塞式的push和pop操作。 * 当队列满时生产者会被阻塞当队列空时消费者会被阻塞。 * * tparam T 队列中元素的类型 */ templatetypename T class ThreadSafeQueue { private: std::queueT queue_; /** 底层数据存储队列 */ mutable std::mutex mutex_; /** 保护队列的互斥锁 */ std::condition_variable cond_var_; /** 用于线程同步的条件变量 */ size_t max_size_; /** 队列最大容量限制 */ public: /** * brief 构造函数 * param max_size 队列的最大容量默认为10 */ explicit ThreadSafeQueue(size_t max_size 10) : max_size_(max_size) {} /** * brief 向队列中添加元素阻塞式 * * 如果队列已满当前线程会被阻塞直到有空位。 * 添加成功后会通知所有等待的消费者线程。 * * param value 要添加的元素值通过移动语义转移 */ void push(T value) { std::unique_lockstd::mutex lock(mutex_); // 等待直到队列有可用空间 cond_var_.wait(lock, [this]() { return queue_.size() max_size_; }); queue_.push(std::move(value)); std::cout [生产] 生产了: value | 队列大小: queue_.size() std::endl; // 通知所有等待的消费者有新数据可用 cond_var_.notify_all(); } /** * brief 从队列中取出元素阻塞式 * * 如果队列为空当前线程会被阻塞直到有数据可用。 * 取出成功后会通知所有等待的生产者线程。 * * param value 引用参数用于接收取出的元素值 * return bool 始终返回true保留接口扩展性 */ bool pop(T value) { std::unique_lockstd::mutex lock(mutex_); // 等待直到队列非空 cond_var_.wait(lock, [this]() { return !queue_.empty(); }); value std::move(queue_.front()); queue_.pop(); std::cout [消费] 消费了: value | 队列大小: queue_.size() std::endl; // 通知所有等待的生产者有空位可用 cond_var_.notify_all(); return true; } /** * brief 获取队列当前大小 * return size_t 队列中的元素数量 */ size_t size() const { std::lock_guardstd::mutex lock(mutex_); return queue_.size(); } /** * brief 检查队列是否为空 * return bool 如果队列为空返回true否则返回false */ bool empty() const { std::lock_guardstd::mutex lock(mutex_); return queue_.empty(); } /** * brief 唤醒所有在条件变量上等待的线程 * * 用于通知所有阻塞的生产者和消费者线程重新检查条件。 * 通常在停止标志设置后调用以便线程能够退出。 */ void notify_all() { cond_var_.notify_all(); } /** * brief 尝试弹出元素非阻塞式 * * 与pop不同此方法不会阻塞线程。如果队列为空则立即返回false。 * * param value 引用参数用于接收取出的元素值 * return bool 如果成功取出元素返回true队列为空返回false */ bool try_pop(T value) { std::lock_guardstd::mutex lock(mutex_); if (queue_.empty()) { return false; } value std::move(queue_.front()); queue_.pop(); cond_var_.notify_all(); return true; } }; /** 全局原子标志用于控制所有工作线程的优雅停止 */ std::atomicbool stop_flag(false); /** * brief 生产者线程函数 * * 按照指定的数量生产产品并放入队列中。每个产品有唯一的ID由生产者ID和序号组成。 * 生产间隔时间为100-500ms的随机延迟模拟实际生产耗时。 * * param queue 线程安全队列的引用用于存放生产的产品 * param producer_id 生产者的唯一标识ID用于生成产品编号和日志输出 * param items_to_produce 该生产者需要生产的产品总数 */ void producer(ThreadSafeQueueint queue, int producer_id, int items_to_produce) { // 初始化随机数生成器用于产生随机的生产延迟时间 std::random_device rd; std::mt19937 gen(rd()); std::uniform_int_distribution delay_dist(100, 500); // 随机延迟 100-500ms // 循环生产指定数量的产品或在收到停止信号时提前退出 for (int i 0; i items_to_produce !stop_flag; i) { int item producer_id * 1000 i; // 生成唯一的产品ID queue.push(item); // 模拟生产过程的时间消耗 std::this_thread::sleep_for(std::chrono::milliseconds(delay_dist(gen))); } std::cout [生产者 producer_id ] 完成生产共生产 items_to_produce 个产品 std::endl; } /** * brief 消费者线程函数 * * 持续从队列中消费产品直到收到停止信号且队列为空。 * 使用非阻塞的try_pop方式获取数据消费间隔时间为200-600ms的随机延迟。 * * param queue 线程安全队列的引用从中获取待消费的产品 * param consumer_id 消费者的唯一标识ID用于日志输出和统计 */ void consumer(ThreadSafeQueueint queue, int consumer_id) { // 初始化随机数生成器用于产生随机的消费延迟时间 std::random_device rd; std::mt19937 gen(rd()); std::uniform_int_distribution delay_dist(200, 600); // 随机延迟 200-600ms int consumed_count 0; // 统计当前消费者消费的产品总数 // 主消费循环持续运行直到满足退出条件 while (true) { int item; // 尝试非阻塞地获取产品 if (queue.try_pop(item)) { consumed_count; std::cout [消费者 consumer_id ] 消费了: item | 已消费总数: consumed_count std::endl; // 模拟消费过程的时间消耗 std::this_thread::sleep_for(std::chrono::milliseconds(delay_dist(gen))); } else { // 队列为空时的处理逻辑 if (stop_flag queue.size() 0) { break; } // 短暂休眠以避免忙等待减少CPU占用 std::this_thread::sleep_for(std::chrono::milliseconds(50)); } } std::cout [消费者 consumer_id ] 完成消费共消费 consumed_count 个产品 std::endl; } /** * brief 程序入口点 - 生产者消费者模式演示 * * 创建多个生产者和消费者线程演示多线程环境下的线程安全数据交换。 * 生产者负责生成数据消费者负责处理数据通过线程安全队列进行通信。 * 程序会在所有生产者完成后给消费者时间处理剩余数据然后优雅地关闭所有线程。 * * return int 程序退出码成功返回0 */ int main() { // Windows平台下设置控制台编码为UTF-8以正确显示中文 #ifdef _WIN32 system(chcp 65001 nul); #endif // 输出程序启动信息 std::cout 生产者-消费者模式演示 std::endl; std::cout 启动多线程系统... std::endl std::endl; std::cout.flush(); // 创建线程安全队列设置最大容量为10 ThreadSafeQueueint queue(10); // 定义系统配置参数 const int NUM_PRODUCERS 3; // 生产者线程数量 const int NUM_CONSUMERS 2; // 消费者线程数量 const int ITEMS_PER_PRODUCER 10; // 每个生产者生产的产品数量 // 存储所有工作线程的容器 std::vectorstd::thread threads; // 记录系统启动时间用于性能统计 auto start_time std::chrono::high_resolution_clock::now(); // 创建并启动所有生产者线程 std::cout 创建 NUM_PRODUCERS 个生产者线程... std::endl; std::cout.flush(); for (int i 0; i NUM_PRODUCERS; i) { threads.emplace_back(producer, std::ref(queue), i 1, ITEMS_PER_PRODUCER); } // 创建并启动所有消费者线程 std::cout 创建 NUM_CONSUMERS 个消费者线程... std::endl; std::cout.flush(); for (int i 0; i NUM_CONSUMERS; i) { threads.emplace_back(consumer, std::ref(queue), i 1); } // 等待所有生产者线程完成各自的生产任务 for (int i 0; i NUM_PRODUCERS; i) { if (threads[i].joinable()) { threads[i].join(); } } std::cout \n所有生产者已完成等待消费者处理剩余产品... std::endl; // 给予消费者足够时间处理队列中剩余的产品 std::this_thread::sleep_for(std::chrono::seconds(2)); // 设置全局停止标志通知消费者线程准备退出 stop_flag true; // 唤醒所有在条件变量上阻塞的消费者线程使其能够检查停止标志 queue.notify_all(); // 等待所有消费者线程完成确保它们都已优雅退出 // 消费者线程存储在threads容后半部分索引从NUM_PRODUCERS开始 for (int i NUM_PRODUCERS; i threads.size(); i) { if (threads[i].joinable()) { threads[i].join(); } } // 计算系统总运行时间 auto end_time std::chrono::high_resolution_clock::now(); auto duration std::chrono::duration_caststd::chrono::milliseconds(end_time - start_time); // 输出系统运行统计信息 std::cout \n 系统关闭 std::endl; std::cout 总耗时: duration.count() 毫秒 std::endl; std::cout 预期生产总数: NUM_PRODUCERS * ITEMS_PER_PRODUCER std::endl; std::cout 最终队列大小: queue.size() std::endl; return 0; }运行结果D:\ClionProject\CLionProjects\untitled\Multithreading.exe 生产者-消费者模式演示 启动多线程系统... 创建 3 个生产者线程... 创建 2 个消费者线程... [生产] 生产了: 1000 | 队列大小: 1 [生产] 生产了: 2000 | 队列大小: 2 [消费者 1] 消费了: 1000 | 已消费总数: 1 [消费者 2] 消费了: 2000 | 已消费总数: 1 [生产] 生产了: 3000 | 队列大小: 1 [生产] 生产了: 2001 | 队列大小: 2 [消费者 2] 消费了: 3000 | 已消费总数: 2 [消费者 1] 消费了: 2001 | 已消费总数: 2 [生产] 生产了: 3001 | 队列大小: 1 [生产] 生产了: 1001 | 队列大小: 2 [生产] 生产了: 1002 | 队列大小: 3 [生产] 生产了: 2002 | 队列大小: 4 [生产] 生产了: 2003 | 队列大小: 5 [消费者 1] 消费了: 3001 | 已消费总数: 3 [生产] 生产了: 1003 | 队列大小: 5 [消费者 2] 消费了: 1001 | 已消费总数: 3 [生产] 生产了: 3002 | 队列大小: 5 [生产] 生产了: 2004 | 队列大小: 6 [生产] 生产了: 3003 | 队列大小: 7 [生产] 生产了: 2005 | 队列大小: 8 [消费者 1] 消费了: 1002 | 已消费总数: 4 [生产] 生产了: 1004 | 队列大小: 8 [生产] 生产了: 3004 | 队列大小: 9 [消费者 2] 消费了: 2002 | 已消费总数: 4 [生产] 生产了: 2006 | 队列大小: 9 [生产] 生产了: 1005 | 队列大小: 10 [消费者 2] 消费了: 2003 | 已消费总数: 5 [生产] 生产了: 2007 | 队列大小: 10 [消费者 1] 消费了: 1003 | 已消费总数: 5 [生产] 生产了: 3005 | 队列大小: 10 [消费者 2] 消费了: 3002 | 已消费总数: 6 [生产] 生产了: 2008 | 队列大小: 10 [消费者 1] 消费了: 2004 | 已消费总数: 6 [生产] 生产了: 1006 | 队列大小: 10 [消费者 2] 消费了: 3003 | 已消费总数: 7[生产] 生产了: 2009 | 队列大小: 10 [消费者 1] 消费了: 2005 | 已消费总数: 7 [生产] 生产了: 3006 | 队列大小: 10 [生产者 2] 完成生产共生产 10 个产品 [消费者 2] 消费了: 3004 | 已消费总数: 8 [消费者 1] 消费了: 1004 | 已消费总数: 8 [生产] 生产了: 1007 | 队列大小: 9 [生产] 生产了: 3007 | 队列大小: 10 [消费者 2] 消费了: 2006 | 已消费总数: 9 [生产] 生产了: 1008 | 队列大小: 10 [消费者 [生产] 生产了: 3008 | 队列大小: 10 2] 消费了: 1005 | 已消费总数: 10 [消费者 1] 消费了: 2007 | 已消费总数: 9 [生产] 生产了: 1009 | 队列大小: 10 [消费者 1] 消费了: 3005 | 已消费总数: 10 [生产] 生产了: 3009 | 队列大小: 10 [生产者 1] 完成生产共生产 10 个产品 [消费者 2] 消费了: 2008 | 已消费总数: 11 [消费者 1] 消费了: 1006 | 已消费总数: 11 [生产者 3] 完成生产共生产 10 个产品 所有生产者已完成等待消费者处理剩余产品... [消费者 2] 消费了: 2009 | 已消费总数: 12 [消费者 1] 消费了: 3006 | 已消费总数: 12 [消费者 2] 消费了: 1007 | 已消费总数: 13 [消费者 1] 消费了: 3007 | 已消费总数: 13 [消费者 2] 消费了: 1008 | 已消费总数: 14 [消费者 1] 消费了: 3008 | 已消费总数: 14 [消费者 1] 消费了: 1009 | 已消费总数: 15 [消费者 2] 消费了: 3009 | 已消费总数: 15 [消费者 2] 完成消费共消费 15 个产品 [消费者 1] 完成消费共消费 15 个产品 系统关闭 总耗时: 6316 毫秒 预期生产总数: 30 最终队列大小: 0 进程已结束退出代码为 0