本文展示了一个基于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 队列中元素的类型 */ template<typename T> class ThreadSafeQueue { private: std::queue<T> 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_lock<std::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_lock<std::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_guard<std::mutex> lock(mutex_); return queue_.size(); } /** * @brief 检查队列是否为空 * @return bool 如果队列为空返回true，否则返回false */ bool empty() const { std::lock_guard<std::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_guard<std::mutex> lock(mutex_); if (queue_.empty()) { return false; } value = std::move(queue_.front()); queue_.pop(); cond_var_.notify_all(); return true; } }; /** 全局原子标志，用于控制所有工作线程的优雅停止 */ std::atomic<bool> stop_flag(false); /** * @brief 生产者线程函数 * * 按照指定的数量生产产品并放入队列中。每个产品有唯一的ID（由生产者ID和序号组成）。 * 生产间隔时间为100-500ms的随机延迟，模拟实际生产耗时。 * * @param queue 线程安全队列的引用，用于存放生产的产品 * @param producer_id 生产者的唯一标识ID，用于生成产品编号和日志输出 * @param items_to_produce 该生产者需要生产的产品总数 */ void producer(ThreadSafeQueue<int>& 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(ThreadSafeQueue<int>& 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 ThreadSafeQueue<int> queue(10); // 定义系统配置参数 const int NUM_PRODUCERS = 3; // 生产者线程数量 const int NUM_CONSUMERS = 2; // 消费者线程数量 const int ITEMS_PER_PRODUCER = 10; // 每个生产者生产的产品数量 // 存储所有工作线程的容器 std::vector<std::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_cast<std::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
