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

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


阻塞队列(BlockQueue)介绍

阻塞队列(Blocking Queue)是一种特殊类型的队列,它具有阻塞操作的特性。在并发编程中,阻塞队列可以用于实现线程间的安全通信和数据共享。

阻塞队列的 主要特点 是:

  • 队列为空时,消费者线程尝试从队列中获取(出队)元素时会被阻塞,直到有新的元素被添加到队列中为止。
  • 队列已满时,生产者线程尝试向队列中添加(入队)元素时也会被阻塞,直到有空闲容量可用。

阻塞队列通常提供入队操作、出队操作以及获取队列大小等基本方法。

基于 BlockQueue(阻塞队列) 的 生产者消费者模型,linux进阶,数据库,缓存,c++,开发语言,linux

阻塞队列的实现在下文


生产者消费者模型 介绍

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

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

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

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

基于 BlockQueue(阻塞队列) 的 生产者消费者模型,linux进阶,数据库,缓存,c++,开发语言,linux


代码实现

在代码实现上,生产者消费者模型通常涉及以下几个 角色和操作

  1. 生产者(Producer):负责生成数据并将其放入共享的缓冲区。
  2. 消费者(Consumer):从共享的缓冲区中获取数据并进行处理。
  3. 缓冲区(Buffer):用于暂存生产者生成的数据,供消费者使用。
  4. 同步机制:用于确保生产者和消费者之间的协调和同步,以避免竞态条件和数据不一致性等问题。

我们将要实现的代码中:

阻塞队列 作为缓冲区Task任务类 由生产者生产传入阻塞队列,以便消费者拿去任务消费lockGuard 与条件变量 保证 生产者消费者之间的协调,同步。


lockGuard.hpp()

在 lockGuard.hpp中我们 实现了一个 需封装了互斥锁的Mutex类和一个 实现自动加解锁的lockGuard类

Mutex类封装了pthread_mutex_t类型的互斥锁, lockGuard类是一个RAII风格的加锁方式。

通过这种方式,lockGuard对象的生命周期和锁的生命周期绑定在一起,可以确保在任何情况下都能保证锁的正确释放,避免死锁等问题

完整代码:

#pragma once                                                                           

#include <iostream>
#include <pthread.h>

using std::cout; using std::endl;

// Mutex类封装 pthread_mutex_t 互斥锁
class Mutex
{
public:
    // 构造 
    Mutex(pthread_mutex_t* mtx):_pmtx(mtx){}

    // 调用lock 进行加锁
    void lock()
    {
        cout << "进行加锁" << endl;
        pthread_mutex_lock(_pmtx);
    }

    // 调用unlock 进行解锁
    void unlock()
    {
        cout << "进行解锁" << endl;
        pthread_mutex_unlock(_pmtx);
    }
                                                                                       
    ~Mutex()
    {}

private:
    pthread_mutex_t* _pmtx; 
};


// RAII 风格的加锁方式
// 以实现自动加解锁
class lockGuard
{
public:
    // 构造
    lockGuard(pthread_mutex_t* mtx):_mtx(mtx)
    {
        _mtx.lock();
    }

    // 析构
    ~lockGuard()
    {
        _mtx.unlock();
    }

private:
    Mutex _mtx;                                                                        
};

Task.hpp(任务类)

下面的代码 是一个简化的 任务封装类用于生产者产生任务并将其放入阻塞队列,供消费者取出并执行。通过将函数与参数打包成任务,实现了任务的传递和执行。

#pragma once

#include <iostream>
#include <functional>                                                                  
    
// 表示一个函数类型。    
// func_t是一个接受两个整数参数并返回整数的函数类型    
typedef std::function<int(int, int)> func_t;    

// 任务类型: 用于生产者产生任务
class Task    
{    
public:    
    Task(){};    

    // 传入三个参数x,y,以及一个函数,task则执行func(x, y)    
    Task(int x, int y, func_t func):_x(x),_y(y),_func(func)    
    {}    
    
    // 用于执行任务。在函数体内部,会调用存储在 _func 中的函数对象,
    // 并将 _x 和 _y 作为参数传递给这个函数对象。
    // 最后 返回执行结果。
    int operator()()    
    {    
        return _func(_x,_y);    
    } 
       
public:
    // 用作函数参数
    int _x;
    int _y;
    func_t _func;
};    

BlockQueue.hpp(阻塞队列)

对 阻塞队列 进行类的实现:

BlockQueue包含以下成员变量

std::queue<T> _bq;   // 阻塞队列
int _capacity; // 容量上限
pthread_mutex_t _mtx;   // 互斥锁: 保证队列安全
pthread_cond_t _empty; // 表示bq是否为空
pthread_cond_t _full; // 表示bq是否为满 

以及除构造函数/析构函数外的以下 BlockQueue包含以下成员函数

bool isQueueEmpty() // 判断队列是否为空
bool isQueueFull() // 判断队列是否为满
void push(const T &in) // 生产者用于制造任务
void pop(const T &in) // 消费者用于消耗任务

完整代码:

#pragma once

#include <iostream>
#include <queue>
#include <mutex>
#include <pthread.h>

#include "lockGuard.hpp"

const int gDefaultCap = 5; // 作为阻塞队列的默认容量

// 阻塞队列
template <class T>
class BlockQueue
{
private:
    // 判断队列是否为空
    bool isQueueEmpty()
    {
        return _bq.size() == 0;
    }

    // 判满
    bool isQueueFull()
    {
        return _bq.size() == _capacity; // 当size == _capacity 证明队列已满
    }

public:
    // 构造
    BlockQueue(int capacity = gDefaultCap) : _capacity(capacity)
    {
        // 初始化互斥锁 && 条件变量
        pthread_mutex_init(&_mtx, nullptr);
        pthread_cond_init(&_empty, nullptr);
        pthread_cond_init(&_full, nullptr);
    }

    // 析构
    ~BlockQueue()
    {
        // 销毁 互斥锁 && 条件变量
        pthread_mutex_destroy(&_mtx);
        pthread_cond_destroy(&_full);
        pthread_cond_destroy(&_empty);
    }

    // 生产者进程
    void push(const T &in)
    {
        // 创建一个lockGuard 变量
        lockGuard lockguard(&_mtx);

        while(isQueueFull())
        {   
            // 如果此时阻塞队列为满,进程等待,直到有空位时改变_full
            pthread_cond_wait(&_full, &_mtx);
        }

        // 此时阻塞队列有空位,正常插入元素,并
        _bq.push(in);
        pthread_cond_signal(&_empty); // 发送信号,表示队列不再为空

        pthread_mutex_unlock(&_mtx);
    }

    // 消费者进程
    void pop(T *out)
    {
        lockGuard lockguard(&_mtx);
        // pthread_mutex_lock(&mtx_);
        while (isQueueEmpty())  // 如果队列为空,等待生产者制造任务
            pthread_cond_wait(&_empty, &_mtx);

        // 此时队列内有任务,
        *out = _bq.front(); // 拿_bq的头部元素,并执行pop(拿任务+销毁)
        _bq.pop();

        pthread_cond_signal(&_full);

        pthread_mutex_unlock(&_mtx);
    }

private:
    std::queue<T> _bq;   // 阻塞队列
    int _capacity; // 容量上限
    pthread_mutex_t _mtx;   // 互斥锁: 保证队列安全
    pthread_cond_t _empty; // 表示bq是否为空
    pthread_cond_t _full; // 表示bq是否为满 
};

conProd.cc(生产者消费者模型 主进程)

该文件中包含以下函数:

  • myAdd 函数:一个简单的加法函数,即实际执行任务所执行的函数

  • consumer 函数消费者线程的执行函数。该函数从阻塞队列中获取任务,并执行任务的函数。

  • productor 函数生产者线程的执行函数。该函数随机生成两个整数参数,创建一个任务对象,并将任务对象插入到阻塞队列中。

  • main 函数主函数,用于创建并启动多个消费者线程和生产者线程。通过调用 pthread_create 创建线程,并通过 pthread_join 等待线程结束。

完整代码:

#include "blockQueue.hpp"                                                           
#include "Task.hpp"    
#include <pthread.h>
#include <unistd.h> 
      
// 加法函数,用于生产者进程产生任务
int myAdd(int x, int y)    
{    
    return x + y;    
}    

// 消费者进程
void *consumer(void* args)    
{  
  // 将获得的agrs 参数 强制转化为BlockQueue<Task>* 类型 并赋值给变量bqueue
  BlockQueue<Task>* bqueue = (BlockQueue<Task>*)args;
  
  while(true)
  {
    // 获取任务
    Task t;
    bqueue->pop(&t); // 执行任务 + 销毁
    // 打印任务信息,因为我们使用的仅仅是一个加法函数,所以直接打印"+"
    cout << pthread_self() << " consumer: " << t._x << " + " << t._y << " = " << t() << endl;
  }

  return nullptr;
}

// 生产者进程
void* productor(void* args)
{
  BlockQueue<Task>* bqueue = (BlockQueue<Task>*)args;

  while(true)
  {
    // 制造任务
    // 生产者将任务传到缓冲区,消费者再将其消耗
    // 任务不一定有生产者制造,也可能通过外部获得

    // 随机产生x, y两个参数,执行Task
    int x = rand() % 10 + 1;
    usleep(rand() % 1000);  
    int y = rand() % 5 + 1;

    Task t(x, y, myAdd);
    // 发送任务
    bqueue->push(t);
    // 输出消息
    cout << pthread_self() << " productor: " << t._x << " + " << t._y << " = ?" << endl;
    sleep(1);
  }
  return nullptr;
}

int main()
{
  // getpid():获取当前进程的进程ID(PID),用于区分不同的进程。
  // 0x11451 用于增加种子的随机性
  srand((uint64_t)time(nullptr) ^ getpid() ^ 0x11451);

  BlockQueue<Task>* bqueue = new BlockQueue<Task>();

  pthread_t con[2], pro[2]; // 声明两个消费者 / 生产者,增加并行性
  // 可以将 &con[1] 换为 con+1
  pthread_create(&con[0], nullptr, consumer, bqueue);
  pthread_create(&con[1], nullptr, consumer, bqueue);
  pthread_create(&pro[0], nullptr, productor, bqueue);
  pthread_create(&pro[1], nullptr, productor, bqueue);

  // 执行完毕,等待进程销毁
  pthread_join(con[0], nullptr);
  pthread_join(con[1], nullptr);
  pthread_join(pro[0], nullptr);
  pthread_join(pro[1], nullptr);

  delete bqueue; // 销毁队列
  
  return 0;
}

执行结果

基于 BlockQueue(阻塞队列) 的 生产者消费者模型,linux进阶,数据库,缓存,c++,开发语言,linux

根据上面的执行结果,可以看出,程序先连续生产(即加锁信息的打印),阻塞队列满了后开始消费,后面重复 生产消费(即加锁解锁)的步骤。文章来源地址https://www.toymoban.com/news/detail-663484.html

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

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

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

相关文章

  • 多线程(初阶七:阻塞队列和生产者消费者模型)

    目录 一、阻塞队列的简单介绍 二、生产者消费者模型 1、举个栗子: 2、引入生产者消费者模型的意义: (1)解耦合 (2)削峰填谷 三、模拟实现阻塞队列 1、阻塞队列的简单介绍 2、实现阻塞队列 (1)实现普通队列 (2)加上线程安全 (3)加上阻塞功能 3、运用阻塞队列

    2024年02月05日
    浏览(41)
  • 多线程学习之生产者和消费者与阻塞队列的关系

    生产者消费者问题,实际上主要是包含了两类线程: 生产者线程用于生产数据 消费者线程用于消费数据 生产者和消费者之间通常会采用一个共享的数据区域,这样就可以将生产者和消费者进行解耦, 两者都不需要互相关注对方的 Object类的等待和唤醒方法 方法名 说明 void

    2024年02月11日
    浏览(43)
  • 【Linux】生产者消费者模型(阻塞队列与环形队列)和POSIX信号量

    我们这里举一个例子,来解释生产者消费者模型,我们学生–消费者,供应商–生产者,超市–交易场所,我们买东西只需要关系售货架子上是否有商品即可,没有了商品,超市从供应商进行供货。供应商和供应商不能同时向一个货架进行供货,所以生产者之间是互斥的关系

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

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

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

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

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

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

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

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

    2024年02月09日
    浏览(56)
  • 优雅封装RabbitMQ实现动态队列、动态生产者,动态消费者绑定

    前言 SpringBoot 集成 RabbitMQ 公司老大觉得使用注解太繁琐了,而且不能动态生成队列所以让我研究是否可以动态绑定,所以就有了这个事情。打工人就是命苦没办法,硬着头皮直接就上了,接下来进入主题吧。 需求思路分析 根据老大的需求,大致分为使用配置文件进行配置,

    2024年02月16日
    浏览(42)
  • Linux之信号量 | 消费者生产者模型的循环队列

    目录 一、信号量 1、概念 2、信号量操作函数 二、基于环形队列的生产者消费者模型 1、模型分析 2、代码实现 1、单生产单消费的生产者消费者模型 2、多生产多消费的生产者消费者模型 引入:前面我们讲到了,对临界资源进行访问时,为了保证数据的一致性,我们需要对临

    2024年04月17日
    浏览(40)
  • 消息队列Pulsar入门(一) 生产者/消费者/Topic详解,附源码演示

    部署问题 连接Pulsar 创建方式 简单方法创建 loadConf自定义配置创建 Pulsar官网 发送模式 同步发送 异步发送 访问方式/发送方式 Share模式(默认情况) 请注意: Exclusive WaitForExclusive 创建方式 简单方法创建 监听器方法创建 loadConf自定义配置创建 多主题订阅 传入List数组的多主题订阅

    2024年02月08日
    浏览(44)

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

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

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

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

二维码1

领取红包

二维码2

领红包