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

这篇具有很好参考价值的文章主要介绍了基于互斥锁的生产者消费者模型。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

生产者消费者 定义

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

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

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

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

基于互斥锁的生产者消费者模型,linux进阶,数据结构&&算法,c++,开发语言,linux


代码实现 / 思路

完整代码

#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 和 条件变量cond
    pthread_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下,可以看出来:

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

基于互斥锁的生产者消费者模型,linux进阶,数据结构&amp;&amp;算法,c++,开发语言,linux文章来源地址https://www.toymoban.com/news/detail-658925.html

到了这里,关于基于互斥锁的生产者消费者模型的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 基于 BlockQueue(阻塞队列) 的 生产者消费者模型

    阻塞队列(Blocking Queue) 是一种特殊类型的队列,它具有阻塞操作的特性。在并发编程中,阻塞队列可以用于实现线程间的安全通信和数据共享。 阻塞队列的 主要特点 是: 当 队列为空时 ,消费者线程尝试从队列中获取(出队)元素时会被阻塞,直到有新的元素被添加到队

    2024年02月12日
    浏览(48)
  • 【linux】线程同步+基于BlockingQueue的生产者消费者模型

    喜欢的点赞,收藏,关注一下把! 在线程互斥写了一份抢票的代码,我们发现虽然加锁解决了抢到负数票的问题,但是一直都是一个线程在抢票,它错了吗,它没错但是不合理。那我们应该如何安全合理的抢票呢? 讲个小故事。 假设学校有一个VIP学霸自习室,这个自习室有

    2024年02月03日
    浏览(98)
  • 【Linux】基于环形队列的生产者消费者模型的实现

    文章目录 前言 一、基于环形队列的生产者消费者模型的实现 上一篇文章我们讲了信号量的几个接口和基于环形队列的生产者消费者模型,下面我们就快速来实现。 首先我们创建三个文件,分别是makefile,RingQueue.hpp,以及main.cc。我们先简单搭建一下环形队列的框架: 首先我们

    2024年02月11日
    浏览(42)
  • 【Java系列】多线程案例学习——基于阻塞队列实现生产者消费者模型

    个人主页:兜里有颗棉花糖 欢迎 点赞👍 收藏✨ 留言✉ 加关注💓本文由 兜里有颗棉花糖 原创 收录于专栏【Java系列专栏】【JaveEE学习专栏】 本专栏旨在分享学习JavaEE的一点学习心得,欢迎大家在评论区交流讨论💌 什么是阻塞式队列(有两点): 第一点:当队列满的时候

    2024年02月04日
    浏览(52)
  • 【Linux】POSIX信号量 | 基于环形队列的生产者消费者模型

    ​🌠 作者:@阿亮joy. 🎆 专栏: 《学会Linux》 🎇 座右铭:每个优秀的人都有一段沉默的时光,那段时光是付出了很多努力却得不到结果的日子,我们把它叫做扎根 POSIX 信号量和 SystemV 信号量作用相同,都是用于同步操作,达到无冲突的访问共享资源目的。 但 POSIX 可以用于

    2023年04月08日
    浏览(62)
  • 【Linux学习】多线程——同步 | 条件变量 | 基于阻塞队列的生产者消费者模型

    🐱作者:一只大喵咪1201 🐱专栏:《Linux学习》 🔥格言: 你只管努力,剩下的交给时间! 以生活中消费者生产者为例: 生活中,我们大部分人都扮演着消费者的角色,会经常在超市买东西,比如买方便面,而超市的方便面是由供应商生成的。所以我们就是消费者,供应商

    2024年02月05日
    浏览(50)
  • 【Linux】生产者消费者模型:基于阻塞队列和环形队列 | 单例模式线程池

    死锁是指在一组进程中的各个进程均占有不会释放的资源,但因互相申请被其他进程所站用不会释放的资源而处于的一种永久等待状态。 当多线程并发执行并都需要访问临界资源时,因为每个线程都是不同的执行流,这就有可能 导致数据不一致问题 ,为了避免此问题的发生

    2024年01月24日
    浏览(47)
  • 【Linux学习】多线程——信号量 | 基于环形队列的生产者消费者模型 | 自旋锁 | 读写锁

    🐱作者:一只大喵咪1201 🐱专栏:《Linux学习》 🔥格言: 你只管努力,剩下的交给时间! 之前在学习进程间通信的时候,本喵简单的介绍过一下信号量,今天在这里进行详细的介绍。 这是之前写的基于阻塞队列的生产者消费者模型中向阻塞队列中push任务的代码。 上面代码

    2024年02月07日
    浏览(54)
  • 生产者-消费者模型

    目录 1、生产者-消费者模型是什么 2、Java中的实现 3、应用于消息队列 3.1 引入依赖 3.2 rabbitmq网站新建队列queue 3.3 模块中配置application.yml 3.4 生产者实现类 3.5 单元测试,发送msg到rabbitmq的队列(my_simple_queue) 3.6 消费者实现类 3.7 从rabbitmq队列(my_simple_queue)消费数据 3.8 队列的配

    2024年02月06日
    浏览(39)
  • Python多线程Thread——生产者消费者模型 python队列与多线程——生产者消费者模型

    下面面向对象的角度看线程 那么你可以试试看能不能用面向对象的方法实现生产者消费者模型吧。

    2024年02月09日
    浏览(52)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包