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

(C++) 多线程之生产者消费者问题

news 2025/6/25 20:53:34

文章目录

  • 前言
  • Code
    • Code
    • 运行效果
  • 分解讲解
    • main()
    • class ProducerConsumerProblem
    • produce()
    • consumer()
  • END

前言

生产者消费者问题_百度百科 (baidu.com)

生产者消费者问题 （英语：Producer-consumer problem），也称有限缓冲问题（英语：Bounded-buffer problem），是一个多线程同步问题的经典案例。该问题描述了两个共享固定大小缓冲区的线程——即所谓的“生产者”和“消费者”——在实际运行时会发生的问题。生产者的主要作用是生成一定量的数据放到缓冲区中，然后重复此过程。与此同时，消费者也在缓冲区消耗这些数据。该问题的关键就是要保证生产者不会在缓冲区已经装满时加入数据，消费者也不会在缓冲区为空时消耗数据。

Code

Code

#include <condition_variable>
#include <iostream>
#include <mutex>
#include <queue>
#include <string>
#include <thread>
#include <vector>constexpr int WORKERTHREAD_COUNT = 3;class ProducerConsumerProblem {
private:// 任务队列std::queue<std::string> m_queue;// mutexstd::mutex m_mutex;// condition_variablestd::condition_variable m_cv;public:/*** 生产者函数*/void produce(const int cnt) {printf("(Produce Thread:%lld) start\n", std::this_thread::get_id());// RAII 自动解锁std::lock_guard<std::mutex> lock(m_mutex);char buff[1024] = {};for (int i = 0; i < cnt; i += 1) {sprintf(buff, "Produce[%02d]", i);m_queue.push(buff);}// 唤醒所有消费者m_cv.notify_all();// 唤醒任意一个消费者// m_cv.notify_one();}/*** 消费者线程*/void consumer() {printf("(Consumer Thread:%lld) start\n", std::this_thread::get_id());constexpr int SLEEP_TIME = 5;std::string receiveBuffer;while (1) {// 比lock_guard更细致的控制力度std::unique_lock<std::mutex> lock(m_mutex);/*** @brief* 如果任务队列为空，则等待** 为什么是循环判断？* >条件变量虚假唤醒：* >  消费者线程被唤醒后* >  缓存队列中没有数据** 条件变量 wait(mutex)的作用* 1. 把互斥锁解开* 2. 阻塞，等待被唤醒* 3. 给互斥锁加锁* 等效写法* m_cv.wait(lock, [this]() { return !m_queue.empty(); });*/while (m_queue.empty()) {// 等待生产者唤醒printf("(Thread:%lld) is waiting ~~~\n",std::this_thread::get_id());m_cv.wait(lock);}// 此时任务队列有数据，且cv被唤醒// 消费者处理一个任务receiveBuffer = m_queue.front();m_queue.pop();// 当前消费者线程获得数据，手动解锁lock.unlock();// 模拟随机设定一个消费时间int solveTimeCost = (rand() % SLEEP_TIME);printf("(Thread:%lld) => %s need %d seconds ...\n",std::this_thread::get_id(), receiveBuffer.c_str(),solveTimeCost);std::this_thread::sleep_for(std::chrono::seconds(solveTimeCost));}}
};int main() {srand(time(0));ProducerConsumerProblem obj;std::vector<std::thread> thdVec;for (int i = 0; i < WORKERTHREAD_COUNT; i += 1) {// 只能move，放入vector延长对象的生存周期auto thd = std::thread(&ProducerConsumerProblem::consumer, &obj);thdVec.push_back(std::move(thd));thdVec[i].detach();}int cnt;// 通过这里的输入阻塞程序while (printf(">>Please input produce number:>\n"), std::cin >> cnt) {obj.produce(cnt);}return 0;
}

运行效果

环境

g++ (x86_64-posix-seh-rev0, Built by MinGW-W64 project) 7.3.0

效果

分别输入2 10 3

>>Please input produce number:>
(Consumer Thread:2) start
(Consumer Thread:3) start
(Consumer Thread:4) start
(Thread:2) is waiting ~~~
(Thread:3) is waiting ~~~
(Thread:4) is waiting ~~~
2
(Produce Thread:1) start
>>Please input produce number:>
(Thread:4) => Produce[00] need 1 seconds ...
(Thread:2) => Produce[01] need 1 seconds ...
(Thread:3) is waiting ~~~
(Thread:4) is waiting ~~~
(Thread:2) is waiting ~~~
10
(Produce Thread:1) start
>>Please input produce number:>
(Thread:2) => Produce[00] need 2 seconds ...
(Thread:4) => Produce[01] need 2 seconds ...
(Thread:3) => Produce[02] need 1 seconds ...
(Thread:3) => Produce[03] need 2 seconds ...
(Thread:2) => Produce[04] need 4 seconds ...
(Thread:4) => Produce[05] need 4 seconds ...
(Thread:3) => Produce[06] need 4 seconds ...
(Thread:2) => Produce[07] need 0 seconds ...
(Thread:2) => Produce[08] need 4 seconds ...
(Thread:4) => Produce[09] need 0 seconds ...
(Thread:4) is waiting ~~~
(Thread:3) is waiting ~~~
(Thread:2) is waiting ~~~
3
(Produce Thread:1) start
>>Please input produce number:>
(Thread:4) => Produce[00] need 4 seconds ...
(Thread:3) => Produce[02] need 0 seconds ...
(Thread:3) is waiting ~~~
(Thread:2) => Produce[01] need 4 seconds ...
(Thread:4) is waiting ~~~
(Thread:2) is waiting ~~~
^Z

分解讲解

main()

int main() {// 设置随机数种子srand(time(0));// 采用对象+成员函数的方式进行多线程ProducerConsumerProblem obj;std::vector<std::thread> thdVec;for (int i = 0; i < WORKERTHREAD_COUNT; i += 1) {// 只能move，放入vector延长对象的生存周期auto thd = std::thread(&ProducerConsumerProblem::consumer, &obj);thdVec.push_back(std::move(thd));thdVec[i].detach();}// 通过手动输入，生产物品int cnt;// 通过这里的输入阻塞程序while (printf(">>Please input produce number:>\n"), std::cin >> cnt) {obj.produce(cnt);}return 0;
}

这里采用对象+成员函数的形式创建线程，因为成员函数的第一个参数是一个隐式的this，因此写std::thread(&ProducerConsumerProblem::consumer, &obj);

注意，std::thread只能move，不能copy。

要将线程对象的生命周期延长，因此移动到vector中。

为了让下文的code执行，将线程进行分离detach()。

class ProducerConsumerProblem

class ProducerConsumerProblem {
private:// 任务队列std::queue<std::string> m_queue;// mutexstd::mutex m_mutex;// condition_variablestd::condition_variable m_cv;public:/*** 生产者函数*/void produce(const int cnt) ;/*** 消费者函数*/void consumer() ;
};

这里的核心数据结构就是std::condition_variable 通常配合std::unique_lock<>使用。

produce()

    void produce(const int cnt) {printf("(Produce Thread:%lld) start\n", std::this_thread::get_id());// RAII 自动解锁std::lock_guard<std::mutex> lock(m_mutex);char buff[1024] = {};for (int i = 0; i < cnt; i += 1) {sprintf(buff, "Produce[%02d]", i);m_queue.push(buff);}// 唤醒所有消费者m_cv.notify_all();// 唤醒任意一个消费者// m_cv.notify_one();}

std::lock_guard<std::mutex>可以做到最简单的构造时上锁，析构时解锁。

notify_all()唤醒所有wait中的对象。

notify_one()唤醒一个wait中的对象。

consumer()

    void consumer() {printf("(Consumer Thread:%lld) start\n", std::this_thread::get_id());constexpr int SLEEP_TIME = 5;std::string receiveBuffer;while (1) {// 比lock_guard更细致的控制力度std::unique_lock<std::mutex> lock(m_mutex);/// ！！！最重要的部分！！！while (m_queue.empty()) {// 等待生产者唤醒printf("(Thread:%lld) is waiting ~~~\n",std::this_thread::get_id());m_cv.wait(lock);}// 此时任务队列有数据，且cv被唤醒// 消费者处理一个任务receiveBuffer = m_queue.front();m_queue.pop();// 当前消费者线程获得数据，手动解锁lock.unlock();// 模拟随机设定一个消费时间int solveTimeCost = (rand() % SLEEP_TIME);printf("(Thread:%lld) => %s need %d seconds ...\n",std::this_thread::get_id(), receiveBuffer.c_str(),solveTimeCost);std::this_thread::sleep_for(std::chrono::seconds(solveTimeCost));}}

std::unique_lock<std::mutex>居于更细致的操作力度，可以手动上锁解锁。

m_cv.wait(lock);整个demo的核心！！！

  • 为什么是循环判断？
    • 条件变量虚假唤醒：消费者线程被唤醒后，缓存队列中没有数据，重新进入wait
  • 条件变量 wait(mutex)的作用
    1. 把互斥锁解开
    2. 阻塞，等待被唤醒
    3. 给互斥锁加锁

在消费者在处理任务时，可以让unique_lock主动解锁。

END

http://www.lryc.cn/news/108150.html

相关文章：

  • 【C语言学习】逃逸字符（转义字符）
  • 开发手册|Java后端开发规范重点条目整理
  • c++11 标准模板（STL）（std::basic_ofstream）（二）
  • k8s概念-pv和pvc
  • python算法指南程序员经典,python算法教程pdf百度云
  • 微服务使用步骤
  • Ubuntu 23.04 作为系统盘的体验和使用感受
  • 百分点科技跻身中国智慧应急人工智能解决方案市场前三
  • vscode如何退出/切换 github 账号
  • maven发布到中央仓库
  • C#IEnumberable＜＞
  • Flink非对齐checkpoint原理（Flink Unaligned Checkpoint）
  • Linux crontab命令：循环执行定时任务（详解）
  • Linux系统jenkins+newman+postman持续集成环境搭建
  • flutter：Future、Stream、RxDart
  • Jenkins安装、配置、自动化构建前（nodejs）后端（maven）项目
  • 【网络基础进阶之路】设计网络划分的实战详解
  • 艺术二维码 API 申请及使用
  • JVM GC ROOT分析
  • 记一道有趣的sql题
  • C高级【day2】
  • 认识Webpack插件Plugin；CleanWebpackPlugin插件；HtmlWebpackPlugin；DefinePlugin；Mode模式
  • Redis 6.0的新特性：多线程、客户端缓存与安全
  • 【雕爷学编程】MicroPython动手做（37）——驱动LCD与图文显示3
  • 自然语言处理从入门到应用——LangChain：提示（Prompts）-[提示模板：序列化提示信息]
  • 【LinearAlgebra】Chapter 12 - Linear Algebra in Probability Statistics
  • webshell详解
  • 数据结构 | 搜索和排序——搜索
  • 【python】对象
  • k8s概念-污点与容忍