rabbitMq-----消费者管理模块

embedded/2024/10/11 2:57:40/

提示:文章写完后,目录可以自动生成,如何生成可参考右边的帮助文档

文章目录

  • 前言
  • 消费者字段
  • 消费者内存管理类
    • 新增/删除消费者
  • 总的消费者管理类


前言

只有订阅客户端订阅了一个队列消息后,他才是一个消费者,而消费者存在的意义是,当这个队列上有消息的时候我们呢需要给订阅了这个队列的客户端推送消息,也就需要获取客户端的连接,所以我们呢可以通过消费者来找到对应的连接来进行消息的推送。


消费者字段

消费者订阅的队列名称,消费者也是按照队列问单元进行管理的。
消费者标志,标志消费者唯一性
自动应答标志,客户端在订阅队列消息时可以进行设置,如果为true,则将消息推送给消费客户端后不需要等待应带,直接将消息进行删除。
还有一个回调函数。当队列有消息到来时,就需要选择一个队列消费者进行消费,那么如何进行消费呢?就是调用这个回调函数。这个函数的逻辑固定,就是通过连接发送消息。是由我们broker服务器设置的。

//当队列收到一条消息后,需要选择一个消费者进行消费,那么如何消费呢?//就是调用这个回调函数,其内部逻辑是:找到消费者对应的连接,将消息发送给客户端。using ConsumerCallBack = std::function<void(const std::string&,const BasicProperties *,const std::string &)>;struct Consumer{using ptr = std::shared_ptr<Consumer>;std::string _qname; //消费者订阅的队列名称std::string _ctag;  //消费者标识bool _auto_ack;  //自动应答标志ConsumerCallBack _cb;
}

消费者内存管理类

消费者是以队列为单位进行管理的,我们使用一个vector管理订阅该队列的消费者。
目的是方便进行rr轮转。

class QueueConsumer{private:std::string _qname; //队列名称std::mutex _mutex;  uint64_t _rr_sep;   //轮转序号std::vector<Consumer::ptr> _consumers;}

新增/删除消费者

构造一个消费者对象进行添加。删除就是遍历删除。

 //新增一个消费者 Consumer::ptr create(const std::string &qname,const std::string &ctag,bool auto_ack,const ConsumerCallBack &cb){//1.加锁std::unique_lock<std::mutex> lock(_mutex);//2.判断消费者是否存在for(auto &cons : _consumers){if(cons->_ctag == ctag){return cons;}}//3.创建消费者并添加管理Consumer::ptr csp = std::make_shared<Consumer>(qname,ctag,auto_ack,cb);_consumers.push_back(csp);return csp;
}
//删除一个消费者:取消订阅、信道关闭、连接关闭的时候删除
void remove(const std::string &ctag){//1.加锁std::unique_lock<std::mutex> lock(_mutex);//2.遍历删除for(auto it = _consumers.begin(); it != _consumers.end(); ++it){if((*it)->_ctag == ctag){_consumers.erase(it);return;}}return;
}

选择一个消费者,rr轮转的思想。
在收到一条消息后,会进行路由匹配,获取到匹配成功的队列。然后把消息推送到队列上,同时取出一个消费者进行消费。

 //选择一个消费者,rr轮转的思想Consumer::ptr choose(){//1.加锁std::unique_lock<std::mutex> lock(_mutex);//如果没有消费者,则返回空对象if(_consumers.size() == 0){return Consumer::ptr();}//2.获取当前选择的消费者下标uint64_t idx = _rr_sep % _consumers.size();_rr_sep++;//3.获取对象并返回return _consumers[idx];}

总的消费者管理类

这个类才是对外提供的类。
就是用一个哈希表把队列名称和消费者管理对象关联起来进行管理。
有一initQueueConsumer函数就是用力啊初始化队列消费者管理对象的,需要判断当前以存在的队列,来进行初始化对应的消费者管理类。

 class ConsumerManager{private:    std::mutex _mutex;std::unordered_map<std::string,QueueConsumer::ptr> _qconsumers;public:using ptr = std::shared_ptr<ConsumerManager>;ConsumerManager(){}//初始化队列消费者管理void initQueueConsumer(const std::string &qname){//1. 加锁std::unique_lock<std::mutex> lock(_mutex);//2. 重复判断auto it = _qconsumers.find(qname);if (it != _qconsumers.end()) {return ;}//3. 新增auto qconsumers = std::make_shared<QueueConsumer>(qname);_qconsumers.insert(std::make_pair(qname, qconsumers));}
}

删除队列消费者管理类对象,当一个队列删除时,就选哦删除这个队列的消费者管理类对象,

 //删除队列消费者管理void destroyQueueConsumer(const std::string &qname){std::unique_lock<std::mutex> lock(_mutex);auto it = _qconsumers.find(qname);if (it == _qconsumers.end()) {return ;}_qconsumers.erase(it);}

创建/删除指定队列的消费者管理类对象中的消费者
根据队列的名称,选择指定的消费者管理类对象进行一个操作即可,

Consumer::ptr create(const std::string &qname,const std::string &ctag,bool auto_ack,const ConsumerCallBack &cb){QueueConsumer::ptr qcp;{std::unique_lock<std::mutex> lock(_mutex);auto it = _qconsumers.find(qname);if(it == _qconsumers.end()){ELOG("没有找到队列 %s 的消费者管理句柄!",qname.c_str());return Consumer::ptr();}qcp = it->second;}//这个create不需要我们这里的mutex保护,他有自己的mutexreturn qcp->create(qname,ctag,auto_ack,cb);
}
void remove(const std::string &ctag,const std::string &qname){QueueConsumer::ptr qcp;{std::unique_lock<std::mutex> lock(_mutex);auto it = _qconsumers.find(qname);if(it == _qconsumers.end()){ELOG("没有找到队列 %s 的消费者管理句柄!",qname.c_str());return ;}qcp = it->second;}qcp->remove(ctag);
}

选择指定队列的消费者管理类对象中的消费者


Consumer::ptr choose(const std::string &qname){QueueConsumer::ptr qcp;{std::unique_lock<std::mutex> lock(_mutex);auto it = _qconsumers.find(qname);if(it == _qconsumers.end()){ELOG("没有找到队列 %s 的消费者管理句柄!",qname.c_str());return Consumer::ptr();}qcp = it->second;}return qcp->choose();
}

http://www.ppmy.cn/embedded/125668.html

相关文章

打破常规,BD仓储物流的效能提升!

当前&#xff0c;随着国家战略的推进&#xff0c;JS与民用领域的融合不断加深&#xff0c;物流业也步入了军民融合的新时代。在智能仓储物流方面&#xff0c;JS物流的智能化进展受到了BD系统的高度关注和重视。 一、建设JS仓储物流RFID基础设施 JS物流领域引入RFID技术的基础工…

238. 除自身以外数组的乘积

文章目录 1.题目2.思路3.代码 1.题目 238. 除自身以外数组的乘积 给你一个整数数组 nums&#xff0c;返回 数组 answer &#xff0c;其中 answer[i] 等于 nums 中除 nums[i] 之外其余各元素的乘积 。 题目数据 保证 数组 nums之中任意元素的全部前缀元素和后缀的乘积都在 32…

【Unity】版本不一致且未升级资产,导致 Unity Sprite 2D 动画播放错误

自己的 Unity版本是 2022.3.45f1。目前折腾的这插件 2D Action RPG Engine: Mythril2D &#xff0c;推荐使用的 Unity 版本是 2021.3.18。 倒腾了这个 unity animation 动画半天&#xff0c;发现这个 animation sprite resolver 在导入动画帧的时候&#xff0c;一直都导入的是…

netty详细说明ByteBuf的使用

一、ByteBuf的使用 ByteBuf是Netty提供的用于处理网络数据的字节缓冲区&#xff0c;具有以下特点和使用注意事项&#xff1a; 基本概念和特点 功能强大&#xff1a;Netty的ByteBuffer替代品&#xff0c;解决了JDK API的局限性&#xff0c;为网络应用程序开发者提供了更好的AP…

请求响应-08.响应-案例

一.案例 获取员工数据&#xff0c;返回统一响应结果&#xff0c;在页面上渲染展示 二.展示最终效果 三.步骤 步骤一&#xff1a; <dependency><groupId>org.dom4j</groupId><artifactId>dom4j</artifactId><version>2.1.3<…

从FastBEV来学习如何做PTQ以及量化

0. 简介 对于深度学习而言&#xff0c;通过模型加速来嵌入进C是非常有意义的&#xff0c;因为本身训练出来的pt文件其实效率比较低下&#xff0c;在讲完BEVDET后&#xff0c;这里我们将以CUDA-FastBEV作为例子&#xff0c;来向读者展示如何去跑CUDA版本的Fast-BEV&#xff0c;…

WSL2环境下Ubuntu的Docker安装与配置

检查是否存在安装残留&#xff0c;移除可能会造成冲突的组件。 for pkg in docker.io docker-doc docker-compose docker-compose-v2 podman-docker containerd runc; do sudo apt-get remove $pkg; done从apt Docker仓库中安装官方GPG key&#xff1a; sudo apt-get update …

锐明Crocus系统 RepairRecord.do SQL注入漏洞

0x01 产品描述&#xff1a; 明锐技术是一家专注于AI和视频技术的商用车智能物联&#xff08;AIoT&#xff09;解决方案提供商&#xff0c;Crocus系统是其核心产品之一。该系统旨在利用人工智能、高清视频、大数据和自动驾驶技术&#xff0c;提高企业或车队的运营效率&#xff0…