基于互斥锁的生产者消费者模型

文章目录

  • 生产者消费者 定义
  • 代码实现 / 思路
    • 完整代码
      • 执行逻辑 / 思路
    • 局部具体分析
      • model.cc
      • func(消费者线程)
  • 执行结果

生产者消费者 定义

生产者消费者模型 是一种常用的 并发编程模型 ,用于解决多线程或多进程环境下的协作问题。该模型包含两类角色:生产者和消费者

生产者负责生成数据,并将数据存放到共享的缓冲区中。消费者则从缓冲区中获取数据并进行处理。生产者和消费者之间通过共享的缓冲区进行数据交互。

为了确保线程安全,生产者和消费者需要遵循一些规则

  1. 如果缓冲区已满,则生产者需要等待直到有空间可用。
  2. 如果缓冲区为空,则消费者需要等待直到有数据可用。
  3. 生产者和消费者都不能访问缓冲区的内部结构,只能通过特定的接口进行操作。

在这里插入图片描述


代码实现 / 思路

完整代码

#include <iostream>
#include <string>
#include <pthread.h>
#include <unistd.h>// 生产者消费者模型
using namespace std;#define TNUM 4 // 定义将使用的线程数
typedef void (*func_t)(const string& name, pthread_mutex_t* pmtx, pthread_cond_t* pcond);
volatile bool quit = false; // 退出信号,默认为false// 定义一个具有名称、函数和同步机制(互斥锁和条件变量)的线程数据结构
// 用于传递线程相关的信息和共享资源给不同的线程,实现线程间的通信和同步
class ThreadData
{
public:ThreadData(const string& name, func_t func, pthread_mutex_t* pmtx, pthread_cond_t* pcond): _name(name), _func(func), _pmtx(pmtx), _pcond(pcond) {}public:// 成员变量string _name; // 线程名func_t _func; // 函数指针pthread_mutex_t* _pmtx; // 互斥锁指针pthread_cond_t* _pcond; // 条件变量指针
};void func1(const string& name, pthread_mutex_t* pmtx, pthread_cond_t* pcond)
{while(!quit){// wait 需要在加锁和解锁之间pthread_mutex_lock(pmtx); // 加锁//pthread_cond_wait(pcond, pmtx); // 默认该线程在执行时,wait 代码被执行,当前线程会被立即阻塞cout << name << " running <-> 播放" << endl;pthread_mutex_unlock(pmtx); // 解锁}
}void func2(const string& name, pthread_mutex_t* pmtx, pthread_cond_t* pcond)
{while(!quit){// 加锁 等待 解锁pthread_mutex_lock(pmtx);pthread_cond_wait(pcond, pmtx);cout << name << " running <-> 下载" << endl;pthread_mutex_unlock(pmtx);}
}void func3(const string& name, pthread_mutex_t* pmtx, pthread_cond_t* pcond)
{while(!quit){// 加锁 等待 解锁pthread_mutex_lock(pmtx);pthread_cond_wait(pcond, pmtx);cout << name << " running <-> 刷新" << endl;pthread_mutex_unlock(pmtx);}
}void func4(const string& name, pthread_mutex_t* pmtx, pthread_cond_t* pcond)
{while(!quit){// 加锁 等待 解锁pthread_mutex_lock(pmtx);pthread_cond_wait(pcond, pmtx);cout << name << " running <-> 扫码用户信息" << endl;pthread_mutex_unlock(pmtx);}
}// 线程入口函数
void* Entry(void *args)
{ThreadData* td = (ThreadData*)args; // 获取线程所需的数据td->_func(td->_name, td->_pmtx, td->_pcond);delete td;return nullptr;
}int main()
{// 初始化互斥锁mtx 和 条件变量condpthread_mutex_t mtx;pthread_cond_t cond;pthread_mutex_init(&mtx, nullptr);pthread_cond_init(&cond, nullptr);// 创建 TNUM 个线程,并将每个线程相关的函数和共享的互斥锁、条件变量传递给线程的入口函数 Entry。// 每个线程都有一个不同的名称和要执行的函数(func)pthread_t tids[TNUM];func_t funcs[TNUM] = {func1, func2, func3, func4};for (int i = 0; i < TNUM; i++){string name = "Thread ";name += to_string(i+1);ThreadData *td = new ThreadData(name, funcs[i], &mtx, &cond);pthread_create(tids + i, nullptr, Entry, (void*)td); // 创建线程}// 调用 pthread_cond_signal 函数向条件变量发送信号,通知等待该条件的线程可以继续运行int cnt = 20;while(cnt){cout << "resume thread run code ...." << cnt-- << endl << endl; // 打印输出当前计数器的值,并将计数器减一pthread_cond_signal(&cond); // 恢复线程sleep(1);}// 代码设置 quit 标志为 true,// 调用 pthread_cond_broadcast 函数向所有等待该条件的线程广播信号cout << "ctrl done" << endl;quit = true;pthread_cond_broadcast(&cond); // 唤醒所有等待在条件变量 cond 上的线程// 使用 pthread_join 等待所有线程的完成,然后销毁互斥锁和条件变量for(int i = 0; i < TNUM; i++){pthread_join(tids[i], nullptr);cout << "thread: " << tids[i] << "quit" << endl;}pthread_mutex_destroy(&mtx);pthread_cond_destroy(&cond);return 0;
}
  1. 定义了4个线程函数 func1、func2、func3、func4,分别代表4个线程的执行逻辑。
  2. 定义了一个ThreadData类,用于封装线程相关的信息和共享资源
  3. 主函数中,创建了4个线程,并将每个线程的名称、函数指针、互斥锁和条件变量传递给ThreadData对象,然后通过pthread_create函数创建线程
  4. 主线程通过循环调用pthread_cond_signal函数向条件变量发送信号,唤醒一个等待该条件的线程,然后休眠1秒钟。
  5. 当计数器cnt减为0时,主线程设置quit标志为true,并通过pthread_cond_broadcast函数向所有等待该条件的线程广播信号,通知它们可以退出。
  6. 使用pthread_join函数等待所有线程的完成,然后销毁互斥锁和条件变量

其中,在整段代码中,func1、func2、func3和func4函数分别代表消费者,而主函数中通过循环调用pthread_cond_signal函数唤醒等待条件变量的线程部分代表生产者

具体来说:

  • func1函数代表一个消费者,它的执行逻辑是"播放"。
  • func2函数代表另一个消费者,它的执行逻辑是"下载"。
  • func3函数代表第三个消费者,它的执行逻辑是"刷新"。
  • func4函数代表第四个消费者,它的执行逻辑是"扫描用户信息"。

而在主函数中的循环调用pthread_cond_signal函数,将信号发送给条件变量cond,可以唤醒等待该条件的线程。这里的循环调用部分代表生产者,通过不断唤醒等待的消费者线程来模拟生产者产生了数据(信号)。

执行逻辑 / 思路

  1. 首先,主函数开始执行。在主函数中,初始化了互斥锁mtx条件变量cond

  2. 接下来,使用循环创建了4个线程,并将每个线程对应的名称、函数指针、互斥锁和条件变量传递给ThreadData对象,然后通过pthread_create函数创建线程。这样就创建了4个消费者线程。

  3. 主线程进入一个循环,循环执行20次。在每次循环中,输出当前计数器的值,并将计数器减一。然后通过pthread_cond_signal函数向条件变量发送信号唤醒一个等待该条件的线程。主线程休眠1秒钟,再进行下一次循环。这部分模拟了生产者产生数据的过程。

  4. 当计数器cnt减为0时,主线程quit标志设置为true,表示停止生产数据

  5. 主线程调用pthread_cond_broadcast函数向所有等待条件变量的线程广播信号,通知它们可以退出。这部分模拟了生产者通知消费者停止消费的过程

  6. 最后,主线程通过pthread_join函数等待所有线程的完成。每个消费者线程会不断地等在条件变量上,在接收到信号后执行相应的操作,直到收到停止信号。

  7. 当所有线程完成后,主线程销毁互斥锁和条件变量,程序结束。

总结起来,这段代码的逻辑是创建了4个消费者线程,每个线程都等待条件变量的信号,然后执行相应的操作。主线程作为生产者,通过发送信号唤醒消费者线程来模拟生产数据的过程。最后,当需要停止生产数据时,主线程发送停止信号给消费者线程,消费者线程收到信号后执行完当前操作后退出。整个过程实现了一个简单的生产者消费者模型。


局部具体分析

model.cc

正常编写代码时,为了不污染命名空间,避免命名冲突,一般不会直接进行 using namespcade std; 这里为了方便,直接进行引用。

#define TNUM 4 // 定义将使用的线程数
typedef void (*func_t)(const string& name, pthread_mutex_t* pmtx, pthread_cond_t* pcond);
volatile bool quit = false; // 退出信号,默认为false// 定义一个具有名称、函数和同步机制(互斥锁和条件变量)的线程数据结构
// 用于传递线程相关的信息和共享资源给不同的线程,实现线程间的通信和同步
class ThreadData
{
public:ThreadData(const string& name, func_t func, pthread_mutex_t* pmtx, pthread_cond_t* pcond): _name(name), _func(func), _pmtx(pmtx), _pcond(pcond) {}public:// 成员变量string _name; // 线程名func_t _func; // 函数指针pthread_mutex_t* _pmtx; // 互斥锁指针pthread_cond_t* _pcond; // 条件变量指针
};

解释:

  • func_t 是一个函数指针类型,可以指向一个接受 const string& 类型参数、 pthread_mutex_t* 类型参数和 pthread_cond_t* 类型参数的函数,返回类型为 void用于后续对接线程的功能函数
  • ThreadData 是 一个具有名称、函数和同步机制(互斥锁和条件变量)的线程数据结构。用于传递线程相关的信息和共享资源给不同的线程,实现线程间的通信和同步

func(消费者线程)

void func1(const string& name, pthread_mutex_t* pmtx, pthread_cond_t* pcond)
{while(!quit){// wait 需要在加锁和解锁之间pthread_mutex_lock(pmtx); // 加锁//pthread_cond_wait(pcond, pmtx); // 默认该线程在执行时,wait 代码被执行,当前线程会被立即阻塞cout << name << " running <-> 播放" << endl;pthread_mutex_unlock(pmtx); // 解锁}
}
  • func1 为例:
  1. 进入一个无限循环,直到全局变量quittrue才退出。
  2. 在循环内部,首先使用pthread_mutex_lock加锁,保证线程独占互斥锁
  3. 调用pthread_cond_wait等待条件变量,当前线程会被阻塞并释放互斥锁,直到其他线程调用pthread_cond_signalpthread_cond_broadcast来发送信号唤醒该线程。
  4. 线程被唤醒后,输出名称和"running <-> 播放"的信息
  5. 最后使用pthread_mutex_unlock解锁互斥锁

执行结果

在linux下,可以看出来:

当我们执行程序后,四个线程会不断地执行四种操作,并且在一个线程结束当前任务之前,其他线程会进行等待,最后输出线程退出信息。

在这里插入图片描述

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/47684.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Flask-SocketIO和Flask-Login联合开发socketio权限系统

设置 Flask, Flask-SocketIO, Flask-Login: 首先&#xff0c;确保安装了必要的库: pip install Flask Flask-SocketIO Flask-Login基础设置: from flask import Flask, render_template, redirect, url_for, request from flask_socketio import SocketIO, emit from flask_…

unity将结构体/列表与json字符串相互转化

编写Unity程序时&#xff0c;面对大量需要传输或者保存的数据时&#xff0c;为了避免编写重复的代码&#xff0c;故采用NewtonJson插件来将定义好的结构体以及列表等转为json字符串来进行保存和传输。 具体代码如下&#xff1a; using System; using System.IO; using Newtons…

Go语言基础之切片

切片 切片&#xff08;Slice&#xff09;是一个拥有相同类型元素的可变长度的序列。它是基于数组类型做的一层封装。它非常灵活&#xff0c;支持自动扩容。 切片是一个引用类型&#xff0c;它的内部结构包含地址、长度和容量。切片一般用于快速地操作一块数据集合 切片的定义…

Redis 整合中 Redisson 的使用

大家好 , 我是苏麟 , 今天带来 Redisson 使用 . 官方文档 : GitHub - redisson/redisson: Redisson - Easy Redis Java client with features of In-Memory Data Grid. Sync/Async/RxJava/Reactive API. Over 50 Redis based Java objects and services: Set, Multimap, Sorte…

Alibaba-Easyexcel 使用总结

简介 简介 EasyExcel 是一个基于 Java 的简单、省内存的读写 Excel 的开源项目&#xff0c;在尽可能节约内存的情况下支持读写百 M 的 Excel。 但注意&#xff0c;其不支持&#xff1a; 单个文件的并发写入、读取读取图片宏 常见问题 Excel 术语 Sheet&#xff0c;工作薄…

每日一学——案例难点Windows配置

在Windows上配置DNS服务器有几个步骤&#xff1a; 步骤1&#xff1a;打开网络连接设置 在任务栏上右键单击网络图标&#xff0c;并选择“打开网络和Internet设置”。 在新窗口中&#xff0c;选择“更改适配器选项”。 在打开的窗口中&#xff0c;找到正在使用的网络适配器&a…

RNN+LSTM正弦sin信号预测 完整代码数据视频教程

视频讲解:RNN+LSTM正弦sin信号预测_哔哩哔哩_bilibili 效果演示: 数据展示: 完整代码: import torch import torch.nn as nn import torch.optim as optim import numpy as np import matplotlib.pyplot as plt import pandas as pd from sklearn.preprocessing import…

Nginx代理功能与负载均衡详解

序言 Nginx的代理功能与负载均衡功能是最常被用到的&#xff0c;关于nginx的基本语法常识与配置已在上篇文章中有说明&#xff0c;这篇就开门见山&#xff0c;先描述一些关于代理功能的配置&#xff0c;再说明负载均衡详细。 Nginx代理服务的配置说明 1、上一篇中我们在http…

angular实现全局组件

之前我们实现全局组件的第一种方式。我们是在定义了组件的时候通过在declares:[component],然后exports出该组件。最后在页面中每次导入该组件&#xff0c;而这次我们将采用另一种方式来实现 1 新建公用组件&#xff1a; navbreadcrumbnavbreadcrumb.component.htmlnavbreadc…

python的安装

1.进入python官网下载安装包&#xff1a;https://www.python.org/ 2.安装 注意&#xff1a;勾选Add Python 3.8 to PATH选项&#xff0c;并且建议选择Install Now选项&#xff0c;东西多。 解释&#xff1a;这里的path是指系统路径&#xff0c;即将python.exe文件添加到系 统…

非阻塞重试与 Spring Kafka 的集成测试

如何为启用重试和死信发布的消费者的 Spring Kafka 实现编写集成测试。 Kafka 非阻塞重试 Kafka 中的非阻塞重试是通过为主主题配置重试主题来完成的。如果需要&#xff0c;还可以配置其他死信主题。如果所有重试均已用尽&#xff0c;事件将转发至 DLT。公共领域提供了大量资…

go 指针

我们知道go中除了map 切片等类型都是进行值传递的&#xff0c;也就是copy一份&#xff0c;不会修改原有数据 普通指针 package mainimport "fmt"func main() {var a int 1var ip *intip &afmt.Printf("ip: %v\n", ip)fmt.Printf("ip: %v\n&qu…

数据库SQL语句使用

-- 查询所有数据库 show databases; -- 创建数据库&#xff0c;数据库名为mydatabase create database mydatabase; -- 如果没有名为 mydatabase的数据库则创建&#xff0c;有就不创建 create database if not exists mydatabase; -- 如果没有名为 mydatabase的数据库则创建…

ffmpeg 子进程从内存读取文件、提取图片到内存

除了网络、文件io&#xff0c;由python或java或go或c等语言开启的ffmpeg子进程还支持pipe&#xff0c;可以从stdin读入数据&#xff0c;输出转化后的图像到stdout。无需编译 ffmpeg&#xff0c;直接调用 ffmpeg.exe不香么&#xff01; “从内存读”可用于边下载边转码&#xf…

使用lodash的throttle函数会触发两次

当使用lodash的throttle函数时会触发两次&#xff0c;分别在最开始和最后。 严格来说不算是bug&#xff0c;因为官方文档写的很清楚。throttle函数其实有三个参数&#xff1a; _.throttle(func, [wait0], [options]) func: 要节流的函数 wait: 等待时间 options: 选项 op…

【Java面试题】线程创建的三种方式及区别?

三种线程创建方式 继承Thread类&#xff0c;子类重写run&#xff08;&#xff09;方法&#xff0c;调用子类的strat&#xff08;&#xff09;启动线程。实现Runnable接口&#xff0c;实现run&#xff08;&#xff09;方法&#xff0c;调用对象start&#xff08;&#xff09;启…

【回味“经典”】DFS基础训练(N皇后,装载问题)

这篇文章是一年前写的 走进“深度搜索基础训练“&#xff0c;踏入c算法殿堂&#xff08;一&#xff09;和 走进“深度搜索基础训练“&#xff0c;踏入c算法殿堂&#xff08;二&#xff09;的重编版。 希望以此&#xff0c;唤起对那位故人的回忆。 小航走入赛场&#xff0c;比赛…

常见的网络设备有哪些?分别有什么作用?

个人主页&#xff1a;insist--个人主页​​​​​​ 本文专栏&#xff1a;网络基础——带你走进网络世界 本专栏会持续更新网络基础知识&#xff0c;希望大家多多支持&#xff0c;让我们一起探索这个神奇而广阔的网络世界。 目录 一、网络设备的概述 二、常见的网络设备 1、…

eslintignore无效解决办法

项目的根目录下新建.eslintignore&#xff0c;但是无论怎么配置&#xff0c;该文件总是无法生效。本想解决不生效的问题&#xff0c;但是一直无法解决&#xff0c;于是换了一种解决问题的思路。 方法一&#xff1a; 在需要进行忽略的文件顶部加上 /* eslint-disable */这样e…