devbox
从前慢现在也慢
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

仿RabbitMq实现消息队列正式篇（虚拟机篇）

作者：从前慢现在也慢 | 2024-08-15 11:34:38

赞

踩

仿RabbitMq实现消息队列正式篇（虚拟机篇）
@TOC

目录

虚拟机模块

要管理的数据

要管理的操作

消息管理模块

要管理的数据

消息信息

消息主体

消息的管理

管理方法

管理数据

管理操作

队列消息管理

交换机数据管理

要管理的数据

要管理的操作

代码展示

队列数据管理

要管理的数据

要管理的操作

代码展示

绑定数据管理模块

管理的数据

管理的操作

代码展示

虚拟机代码展示

虚拟机模块

要管理的数据

  • 交换机数据管理句柄
  • 队列数据管理句柄
  • 绑定信息数据管理模块
  • 消息数据管理模块

要管理的操作

  • 声明 / 删除交换机：在删除交换机的时候，要删除相关的绑定信息
  • 声明 / 删除队列 ：在删除队列的时候，要删除相关的绑定信息以及消息数据
  • 队列的绑定 / 解绑 ：绑定的时候，必须交换机和队列是存在的
  • 获取指定队列的消息
  • 对指定队列的指定消息进行确认
  • 获取交换机相关的所有绑定信息：一条消息要发布给指定交换机的时候，交换机获取所有绑定信息，来确认消息要发布到哪个队列

基于这个，那我们先完成虚拟机模块管理的相关数据

消息管理模块

  1. syntax = "proto3";  
  2. package mymq;
  3.  
  4. enum ExchangeType  # 交换类型
  5. {
  6.     UNKNOWTYPE = 0;
  7.     DIRECT = 1;
  8.     FANOUT = 2;
  9.     TOPIC = 3; 
  10. };
  11.  
  12. enum DeliveryMode # 是否持久化
  13. {
  14.     UNKNOWMODE = 0;
  15.     UNDURABLE = 1;
  16.     DURABLE = 2;
  17. };
  18.  
  19. message BasicProperties #消息的属性
  20. {
  21.     string id = 1;                         # 消息ID, UUID
  22.     DeliveryMode delivery_mode = 2;        # 持久化模块
  23.     string rounting_key = 3;               # routing_key绑定信息
  24. };
  25.  
  26. message Message
  27. {
  28.     message Payload
  29.     {
  30.         BasicProperties properties = 1;     # 消息属性
  31.         string body = 2;                    # 有效载荷数据
  32.         string valid = 3;                   # 消息是否有效
  33.     };
  34.     Payload payload = 1;                    # 真正持久化的只有这一个字段
  35.     uint32 offset = 2;                      # 消息的位置
  36.     uint32 length = 3;                      # 消息的长度
  37. };

要管理的数据

消息信息

  1. ID：消息的唯一标识
  2. 持久化标识：表示是否对消息进行持久化（还取决于队列的持久化标志）
  3. routing_key：决定了当前消息要发布的队列（消息发布到交换机后，根据绑定队列binding_key决定是否发布到指定队列）
消息主体
  1. 存储偏移量：消息以队列为单元存储在文件中，这个偏移量，是当前消息相对于文件起始位置的偏移量
  2. 消息长度：从偏移量位置取出指定长度的消息（解决了粘包问题）
  3. 是否有效标志：标识当前消息是否已经删除（删除一条消息，并不是用后面的消息覆盖掉前面的数据，而是重置有效标志位，当一个文件中的有效消息占据总消息比例不到50%，且数据量超过2000，则进行垃圾回收，重新整理文件数据存储）

        当系统重启时，也需要重新加载有效消息即可。

消息的管理

管理方法

以队列为单元进行管理（因为消息的所有操作都是以队列为单元）

管理数据
  1.  消息链表
  2. 待确定消息hash
  3. 持久化消息hash
  4. 持久化的有效消息数据
  5. 持久化的总的消息数据
管理操作
  • 向队列新增消息
  • 获取队首消息
  • 对消息进行确认
  • 恢复队列历史消息
  • 垃圾回收
  • 删除队列相关消息文件
队列消息管理
  • 初始化队列消息结构
  • 移除队列消息结果
  • 向队列新增消息
  • 对队列消息进行确认
  • 恢复队列历史消息
  1. #ifndef __M_MSG_H__
  2. #define __M_MSG_H__
  3.  
  4. #include "../mqcommon/logger.hpp"
  5. #include "../mqcommon/helper.hpp"
  6. #include "../mqcommon/msg.pb.h"
  7. #include <iostream>
  8. #include <mutex>
  9. #include <memory>
  10. #include <unordered_map>
  11. #include <list>
  12.  
  13. namespace mymq
  14. {
  15.  
  16.     // 消息的持久化管理
  17.     //  a. 管理数据
  18.     //  i. 队列消息⽂件存储的路径
  19.     //  ii. 队列消息的存储⽂件名
  20.     //  iii. 队列消息的临时交换⽂件名
  21.     // b. 管理操作
  22.     // i. ⽇志消息存储在⽂件中（4B⻓度+(属性+内容+有效位)序列化消息，连续存储即可）
  23.     // ii. 提供队列消息⽂件创建/删除功能
  24.     // iii. 提供队列消息的新增持久化/删除持久化
  25.     // iv. 提供持久化内容的垃圾回收（其实就是重新加载出所有有效消息返回，并重新⽣成新的消息
  26.     // 存储⽂件）
  27. #define DATAFILE_SUBFIX ".mqd"
  28. #define TMPFILE_SUBFIX ".mqb.tmp"
  29.     using MessagePtr = std::shared_ptr<mymq::Message>;
  30.  
  31.     // 消息持久化
  32.     class MessageMapper
  33.     {
  34.     public:
  35.         MessageMapper(std::string &basedir, const std::string &qname)
  36.             : _qname(qname)
  37.         {
  38.             if (basedir.back() != '/')
  39.             {
  40.                 basedir.push_back('/');
  41.             }
  42.             _datafile = basedir + qname + DATAFILE_SUBFIX;
  43.             _tmpfile = basedir + qname + TMPFILE_SUBFIX;
  44.             if (FileHelper(basedir).exists() == false)
  45.             {
  46.                 FileHelper::createDirectory(basedir);
  47.             }
  48.             createMsgFile();
  49.         }
  50.  
  51.         bool createMsgFile()
  52.         {
  53.             if (FileHelper(_datafile).exists() == true)
  54.             {
  55.                 return true;
  56.             }
  57.             bool ret = FileHelper::createFile(_datafile);
  58.             if (ret == false)
  59.             {
  60.                 ELOG("创建队列数据文件：%s 失败", _datafile.c_str());
  61.                 return false;
  62.             }
  63.             return true;
  64.         }
  65.  
  66.         void removeMsgFile()
  67.         {
  68.             FileHelper::removeFile(_datafile);
  69.             FileHelper::removeFile(_tmpfile);
  70.         }
  71.  
  72.         bool insert(MessagePtr &msg)
  73.         {
  74.             return insert(_datafile, msg);
  75.         }
  76.  
  77.         bool remove(MessagePtr &msg)
  78.         {
  79.             // 1. 将msg中的有效标志位修改为‘0’
  80.             msg->mutable_payload()->set_valid("0");
  81.             // 2，对msg进行序列化
  82.             std::string body = msg->SerializeAsString();
  83.             if (body.size() != msg->length())
  84.             {
  85.                 ELOG("不能修改文件中的数据信息，因为生成的数据与原数据长度不一样");
  86.                 return false;
  87.             }
  88.             // 3. 将序列化后的消息，写入到数据在文件中的指定位置（覆盖原有的数据）
  89.             FileHelper helper(_datafile);
  90.             bool ret = helper.write(body.c_str(), msg->offset(), body.size());
  91.             if (ret == false)
  92.             {
  93.                 DLOG("向队列数据文件写入失败！");
  94.                 return false;
  95.             }
  96.             return true;
  97.         }
  98.  
  99.         std::list<MessagePtr> gc()
  100.         {
  101.             std::list<MessagePtr> result;
  102.             bool ret = load(result);
  103.             if (ret == false)
  104.             {
  105.                 ELOG("加载有效数据失败\n");
  106.                 return result;
  107.             }
  108.             // 2. 将有效数据，进行序列化存储到临时文件中
  109.             FileHelper::createFile(_tmpfile);
  110.             for (auto e : result)
  111.             {
  112.                 DLOG("向临时文件中添加数据 : %s", e->payload().body().c_str());
  113.                 ret = insert(_tmpfile, e);
  114.                 if (ret == false)
  115.                 {
  116.                     ELOG("向临时文件中添加数据失败");
  117.                     return result;
  118.                 }
  119.             }
  120.             // 3. 删除源文件
  121.             ret = FileHelper::removeFile(_datafile);
  122.             if (ret == false)
  123.             {
  124.                 ELOG("删除原文件失败");
  125.                 return result;
  126.             }
  127.             // 4. 修改临时文件的名字，改成原文件的名称
  128.             ret = FileHelper(_tmpfile).rename(_datafile.c_str());
  129.             if (ret == false)
  130.             {
  131.                 ELOG("修改文件名称失败");
  132.                 return result;
  133.             }
  134.             // 5. 返回新的有效数据
  135.             return result;
  136.         }
  137.  
  138.     private:
  139.         bool load(std::list<MessagePtr> &result)
  140.         {
  141.             // 1. 加载出文件中的所有的有效数据：存储格式：4字节长度 | 数据 | 4字节长度 | 数据 ......
  142.             FileHelper data_file_helper(_datafile);
  143.             size_t offset = 0;   // 用来定位
  144.             size_t msg_size = 0; // 用来 4 字节长度的数据长度
  145.             size_t file_size = data_file_helper.size();
  146.             while (offset < file_size)
  147.             {
  148.                 bool ret = data_file_helper.read((char*)&msg_size, offset, sizeof(size_t));
  149.                 if (ret == false)
  150.                 {
  151.                     DLOG("读取消息长度失败！");
  152.                 }
  153.                 offset += sizeof(size_t);
  154.                 std::string msg_body(msg_size, '\0');
  155.                 ret = data_file_helper.read(&msg_body[0], offset, msg_size);
  156.                 // 为什么使用 &msg_body[0]
  157.                 // &msg_body[0] 返回的是指向 std::string 内部字符数组首元素的指针，即指向字符串内容的指针。
  158.                 // 适用于需要直接操作字符串内容的场景，例如文件读写操作
  159.  
  160.                 // &msg_body：
  161.                 // &msg_body 返回的是 std::string 对象本身的指针，即指向 std::string 对象的指针。
  162.                 // 不适用于直接操作字符串内容的场景，而是用于操作整个 std::string 对象本身。
  163.                 if (ret == false)
  164.                 {
  165.                     DLOG("读取消息数据失败");
  166.                     return false;
  167.                 }
  168.                 offset += msg_size;
  169.                 MessagePtr msgp = std::make_shared<Message>();
  170.                 msgp->mutable_payload()->ParseFromString(msg_body);
  171.  
  172.                 // 如果是无效消息，则直接处理下一个
  173.                 if (msgp->payload().valid() == "0")
  174.                 {
  175.                     ELOG("加载到无效消息：%s", msgp->payload().body().c_str());
  176.                     continue;
  177.                 }
  178.                 // 有效消息则保存下来
  179.                 result.push_back(msgp);
  180.             }
  181.             return true;
  182.         }
  183.  
  184.         bool insert(const std::string &filename, MessagePtr &msg)
  185.         {
  186.             // 新增数据都是添加在文件末尾的
  187.             //  1. 进行消息的序列化，获取到格式化后的消息
  188.             // 因为使用的是 sqlite 数据库，存储的是二进制信息，所以需要进行序列化
  189.             std::string body = msg->payload().SerializeAsString();
  190.             // 2. 获取文件长度
  191.             FileHelper helper(filename);
  192.             size_t fsize = helper.size();
  193.             size_t msg_size = body.size();
  194.             // 写入逻辑：1. 先写入 4 字节数据长度，2. 再写入指定长度数据
  195.             bool ret = helper.write((char *)&msg_size, fsize, sizeof(size_t));
  196.             if (ret == false)
  197.             {
  198.                 DLOG("向队列数据文件写入数据长度失败！");
  199.                 return false;
  200.             }
  201.             // 将数据写入文件的指定位置
  202.             ret = helper.write(body.c_str(), fsize + sizeof(size_t), msg_size);
  203.             if (ret == false)
  204.             {
  205.                 DLOG("向队列数据文件写入数据失败！");
  206.                 return false;
  207.             }
  208.             // 更新msg中的实际存储信息
  209.             msg->set_offset(fsize + sizeof(size_t));
  210.             msg->set_length(msg_size);
  211.             return true;
  212.         }
  213.  
  214.     private:
  215.         std::string _qname;
  216.         std::string _datafile;
  217.         std::string _tmpfile;
  218.     };
  219.  
  220.     // 管理数据
  221.     class QueueMessage
  222.     {
  223.     public:
  224.         using ptr = std::shared_ptr<QueueMessage>;
  225.         // 初始化
  226.         QueueMessage(std::string &basedir, const std::string &qname)
  227.             : _mapper(basedir, qname),
  228.               _qname(qname),
  229.               _valid_count(0),
  230.               _total_count(0)
  231.         {}
  232.  
  233.         bool recovery()
  234.         {
  235.             // 恢复历史消息
  236.             std::unique_lock<std::mutex> lock(_mutex);
  237.             _msgs = _mapper.gc();
  238.             for (auto &msg : _msgs)
  239.             {
  240.                 _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
  241.             }
  242.             _valid_count = _total_count = _msgs.size();
  243.             return true;
  244.         }
  245.  
  246.         bool insert(const BasicProperties *bp, const std::string &body, bool queue_is_durable)
  247.         {
  248.             // 构造消息对象
  249.             MessagePtr msg = std::make_shared<Message>();
  250.             msg->mutable_payload()->set_body(body);
  251.             if (bp != nullptr)
  252.             {
  253.                 DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
  254.                 msg->mutable_payload()->mutable_properties()->set_id(bp->id());
  255.                 msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
  256.                 msg->mutable_payload()->mutable_properties()->set_rounting_key(bp->rounting_key());
  257.             }
  258.             else
  259.             {
  260.                 DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
  261.                 msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
  262.                 msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
  263.                 msg->mutable_payload()->mutable_properties()->set_rounting_key("");
  264.             }
  265.             std::unique_lock<std::mutex> lock(_mutex);
  266.             // 2. 判断消息是否需要持久化
  267.             if (msg->payload().properties().delivery_mode() == DURABLE)
  268.             {
  269.                 msg->mutable_payload()->set_valid("1");
  270.                 // 3. 进行持久化存储
  271.                 bool ret = _mapper.insert(msg);
  272.                 if (ret == false)
  273.                 {
  274.                     ELOG("持久化存储信息：%s失败了！", body.c_str());
  275.                 }
  276.  
  277.                 _valid_count += 1;
  278.                 _total_count += 1;
  279.                 _durable_msgs.insert(make_pair(msg->payload().properties().id(), msg));
  280.             }
  281.             // 4. 内容的管理
  282.             _msgs.push_back(msg);
  283.             return true;
  284.         }
  285.  
  286.         MessagePtr front()
  287.         {
  288.             std::unique_lock<std::mutex> lock(_mutex);
  289.             if (_msgs.size() == 0)
  290.             {
  291.                 return MessagePtr();
  292.             }
  293.             // 获取一条队首消息：从_msgs中获取数据
  294.             MessagePtr msg = _msgs.front();
  295.             _msgs.pop_front();
  296.             // 将该消息对象，向待确认的hash表中添加一份，等到收到消息确认后进行删除
  297.             _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
  298.             return msg;
  299.         }
  300.  
  301.         // 每次删除消息后，判断是否需要垃圾回收
  302.         bool remove(const std::string &msg_id)
  303.         {
  304.             std::unique_lock<std::mutex> lock(_mutex);
  305.             // 1. 从待确认队列中找到信息
  306.             auto it = _waitack_msgs.find(msg_id);
  307.             if (it == _waitack_msgs.end())
  308.             {
  309.                 DLOG("没有找到要删除的消息：%s!", msg_id.c_str());
  310.                 return true;
  311.             }
  312.             // 2. 根据消息的持久化模式，决定是否删除持久化消息
  313.             if (it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
  314.             {
  315.                 // 删除持久化消息
  316.                 _mapper.remove(it->second);
  317.                 _durable_msgs.erase(msg_id);
  318.                 _valid_count -= 1;
  319.                 gc();
  320.             }
  321.             // 4. 删除内存中的信息
  322.             _waitack_msgs.erase(msg_id);
  323.             return true;
  324.         }
  325.  
  326.         size_t getable_count()
  327.         {
  328.             std::unique_lock<std::mutex> lock(_mutex);
  329.             return _msgs.size();
  330.         }
  331.  
  332.         size_t total_count()
  333.         {
  334.             std::unique_lock<std::mutex> lock(_mutex);
  335.             return _total_count;
  336.         }
  337.  
  338.         size_t durable_count()
  339.         {
  340.             std::unique_lock<std::mutex> lock(_mutex);
  341.             return _durable_msgs.size();
  342.         }
  343.         size_t waitack_count()
  344.         {
  345.             std::unique_lock<std::mutex> lock(_mutex);
  346.             return _waitack_msgs.size();
  347.         }
  348.         void clear()
  349.         {
  350.             std::unique_lock<std::mutex> lock(_mutex);
  351.             _mapper.removeMsgFile();
  352.             _msgs.clear();
  353.             _durable_msgs.clear();
  354.             _waitack_msgs.clear();
  355.             _valid_count = 0;
  356.             _total_count = 0;
  357.         }
  358.  
  359.     private:
  360.         bool GCCheck() {
  361.                 //持久化的消息总量大于2000， 且其中有效比例低于50%则需要持久化
  362.                 if (_total_count > 2000 && _valid_count * 10  / _total_count < 5) {
  363.                     return true;
  364.                 }
  365.                 return false;
  366.         }
  367.  
  368.         void gc()
  369.         {
  370.             // 1. 进行垃圾回收，获取到垃圾回收后，有效的消息信息链表
  371.             if(GCCheck() == false)
  372.             {
  373.                 return; 
  374.             }
  375.             std::list<MessagePtr> msgs = _mapper.gc();
  376.             for(auto& msg : msgs)
  377.             {
  378.                 auto it = _durable_msgs.find(msg->payload().properties().id());
  379.                 if(it == _durable_msgs.end())
  380.                 {
  381.                     DLOG("垃圾回收之后，有一条持久化消息，在内存中没有进行管理");
  382.                     _msgs.push_back(msg); // 做法：重新添加到推送链表的末尾
  383.                     _durable_msgs.insert(make_pair(msg->payload().properties().id(), msg));
  384.                     continue;
  385.                 }
  386.                 // 2. 更新每一条消息的实际存储位置
  387.                 it->second->set_offset(msg->offset());
  388.                 it->second->set_length(msg->length());
  389.             }
  390.             _valid_count = _total_count = msgs.size();
  391.         }
  392.  
  393.     private:
  394.         std::mutex _mutex;
  395.         std::string _qname;
  396.         size_t _valid_count;
  397.         size_t _total_count;
  398.         MessageMapper _mapper;
  399.         std::list<MessagePtr> _msgs;
  400.         std::unordered_map<std::string, MessagePtr> _durable_msgs; // 持久化消息hash
  401.         std::unordered_map<std::string, MessagePtr> _waitack_msgs; // 待确认消息hash
  402.     };
  403.  
  404.     // 实现一个对外的总体消息管理类
  405.     // 消息的总体对外管理
  406.     // i. 初始化新建队列的消息管理结构，并创建消息存储⽂件
  407.     // ii. 删除队列的消息管理结构，以及消息存储⽂件
  408.     // iii. 向指定队列新增消息
  409.     // iv. 获取指定队列队⾸消息
  410.     // v. 确认指定队列待确认消息（删除）
  411.     // vi. 判断指定队列消息是否为空
  412.  
  413.     class MessageManager
  414.     {
  415.     public:
  416.         using ptr = std::shared_ptr<MessageManager>;
  417.         MessageManager(const std::string basedir)
  418.             :_basedir(basedir)
  419.         {}
  420.  
  421.         void clear()
  422.         {
  423.             std::unique_lock<std::mutex> lock(_mutex);
  424.             for(auto& it : _queue_msgs)
  425.             {
  426.                 it.second->clear();
  427.             }
  428.         }
  429.  
  430.         void initQueueMessage(const std::string& qname)
  431.         {
  432.             QueueMessage::ptr qmp;
  433.             {
  434.                 std::unique_lock<std::mutex> lock(_mutex);
  435.                 auto it = _queue_msgs.find(qname);
  436.                 if(it != _queue_msgs.end())
  437.                 {
  438.                     return ;
  439.                 }
  440.                 qmp = std::make_shared<QueueMessage>(_basedir, qname);
  441.                 _queue_msgs.insert(std::make_pair(qname, qmp));
  442.             }
  443.             qmp->recovery();
  444.         }
  445.  
  446.         void destroyQueueMessage(const std::string& qname)
  447.         {
  448.             QueueMessage::ptr qmp;
  449.             {
  450.                 std::unique_lock<std::mutex> lock(_mutex);
  451.                 auto it = _queue_msgs.find(qname);
  452.                 if(it == _queue_msgs.end())
  453.                 {
  454.                     return ;
  455.                 }
  456.                 qmp = it->second;
  457.                 _queue_msgs.erase(it);
  458.             }
  459.             qmp->clear();
  460.         }
  461.  
  462.         bool insert(const std::string& qname, BasicProperties* bp, const std::string& body, bool _queue_is_durable)
  463.         {
  464.             QueueMessage::ptr qmp;
  465.             {
  466.                 std::unique_lock<std::mutex> lock(_mutex);
  467.                 auto it = _queue_msgs.find(qname);
  468.                 if(it == _queue_msgs.end())
  469.                 {
  470.                     ELOG("向队列 %s 新增消息失败，没有找到消息管理句柄", qname.c_str());
  471.                     return false;
  472.                 }
  473.                 qmp = it->second;
  474.             }
  475.             return qmp->insert(bp, body, _queue_is_durable); 
  476.         }
  477.  
  478.         MessagePtr front(const std::string& qname)
  479.         {
  480.             QueueMessage::ptr qmp;
  481.             {
  482.                 std::unique_lock<std::mutex> lock(_mutex);
  483.                 auto it = _queue_msgs.find(qname);
  484.                 if(it == _queue_msgs.end())
  485.                 {
  486.                     DLOG("获取 %s 队首数据失败：没有找到对应的消息句柄", qname.c_str() ); 
  487.                     return MessagePtr();
  488.                 }
  489.                 qmp = it->second;
  490.             }
  491.             //取出对应的相关队列的首部消息
  492.             return qmp->front();
  493.         }
  494.  
  495.         void ack(const std::string& qname, const std::string& msg_id)
  496.         {
  497.             QueueMessage::ptr qmp;
  498.             {
  499.                 std::unique_lock<std::mutex> lock(_mutex);
  500.                 auto it = _queue_msgs.find(qname);
  501.                 if(it == _queue_msgs.end())
  502.                 {
  503.                     DLOG("确认队列 %s 消息 %s 失败：没有找到消息管理句柄", qname.c_str(), msg_id.c_str());
  504.                     return ;
  505.                 }
  506.                 qmp = it->second;
  507.             }
  508.             qmp->remove(msg_id);
  509.             return ;
  510.         }
  511.  
  512.         size_t getable_count(const std::string& qname)
  513.         {
  514.             QueueMessage::ptr qmp;
  515.             {
  516.                 std::unique_lock<std::mutex> lock(_mutex);
  517.                 auto it = _queue_msgs.find(qname);
  518.                 if(it == _queue_msgs.end())
  519.                 {
  520.                     DLOG("获取队列 %s 待推送消息数量失败：没有找到消息管理句柄", qname.c_str());
  521.                     return 0;
  522.                 }
  523.                 qmp = it->second;
  524.             }
  525.             return qmp->getable_count();
  526.         }
  527.         size_t total_count(const std::string& qname)
  528.         {
  529.             QueueMessage::ptr qmp;
  530.             {
  531.                 std::unique_lock<std::mutex> lock(_mutex);
  532.                 auto it = _queue_msgs.find(qname);
  533.                 if(it == _queue_msgs.end())
  534.                 {
  535.                     DLOG("获取队列 %s 总持久化消息数量失败：没有找到消息管理句柄", qname.c_str());
  536.                     return 0;
  537.                 }
  538.                 qmp = it->second;
  539.             }
  540.             return qmp->total_count();
  541.         }
  542.         size_t durable_count(const std::string& qname)
  543.         {
  544.             QueueMessage::ptr qmp;
  545.             {
  546.                 std::unique_lock<std::mutex> lock(_mutex);
  547.                 auto it = _queue_msgs.find(qname);
  548.                 if(it == _queue_msgs.end())
  549.                 {
  550.                     DLOG("获取队列 %s 总持久化消息数量失败：没有找到消息管理句柄", qname.c_str());
  551.                     return 0;
  552.                 }
  553.                 qmp = it->second;
  554.             }
  555.             return qmp->durable_count();
  556.         }
  557.         size_t waitack_count(const std::string& qname)
  558.         {
  559.                  QueueMessage::ptr qmp;
  560.             {
  561.                 std::unique_lock<std::mutex> lock(_mutex);
  562.                 auto it = _queue_msgs.find(qname);
  563.                 if(it == _queue_msgs.end())
  564.                 {
  565.                     DLOG("获取队列 %s 总持久化消息数量失败：没有找到消息管理句柄", qname.c_str());
  566.                     return 0;
  567.                 }
  568.                 qmp = it->second;
  569.             }
  570.             return qmp->waitack_count();
  571.         }
  572.     private:
  573.         std::mutex _mutex;
  574.         std::string _basedir;
  575.         std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs;
  576.     };
  577.  
  578. }
  579.  
  580. #endif

交换机数据管理

要管理的数据

  • 交换机名称：唯一标识
  • 交换机类型：决定了消息的转发方式
  • 每一个队列绑定中有个binding_key, 每条消息中有一个routing_key
  • 直接交换：binding_key和routing_key相同，则将消息放入队列
  • 广播交换：将消息放入交换机绑定的所有队列中
  • 主题交换：routing_key与多个绑定队列的binding_key有个匹配机制，匹配成功了则放入
  • 持久化标志：决定了当前交换机信息是否需要持久化存储
  • 自动化删除标志：指的是关联了当前交换机的所有客户端都退出了，是否要自动删除交换机

要管理的操作

  • 创建交换机：本质上需要的是声明 ----- 强断言的思想，没有的话，就创建
  • 删除交换机：每个交换机都会绑定一个或多个队列（意味着会有一个或多个绑定信息，因此删除交换机需要删除相关绑定信息
  • 获取指定名称的交换机
  • 获取当前交换机名称

注意交换机有持久化的标识，所以在进行管理操作的时候，需要分情况进行，在内存中管理，以及在硬盘中管理

代码展示

  1. #include "../mqcommon/helper.hpp"
  2. #include "../mqcommon/logger.hpp"
  3. #include "../mqcommon/msg.pb.h"
  4. #include "../mqcommon/mq_proto.pb.h"
  5. #include <google/protobuf/map.h>
  6. #include <iostream>
  7. #include <unordered_map>
  8. #include <memory>
  9. #include <mutex>
  10.  
  11. namespace mymq
  12. {
  13.     // 1. 定义交换机类
  14.     struct Exchange
  15.     {
  16.         using ptr = std::shared_ptr<Exchange>; 
  17.         // 1. 交换机名称
  18.         std::string name;
  19.         // 2. 交换机类型
  20.         ExchangeType type;
  21.         // 3. 交换机持久化标志
  22.         bool durable;
  23.         // 4. 是否自动删除标志
  24.         bool auto_delete;
  25.         // 5. 其他参数
  26.         google::protobuf::Map<std::string, std::string> args;
  27.         //因为后面的函数需要构造一个Exchange的空对象
  28.         Exchange()
  29.         {}
  30.         
  31.         Exchange(const std::string &ename, ExchangeType etype, bool edurable, bool eauto_delete, const google::protobuf::Map<std::string, std::string> eargs)
  32.             : name(ename),
  33.               type(etype),
  34.               durable(edurable),
  35.               auto_delete(eauto_delete),
  36.               args(eargs)
  37.         {}
  38.  
  39.         // args存储键值对，在存储数据库的时候，会组织一个格式字符串进行存储 key=value&key=value.....
  40.         // 内部解析str_args字符串，将容器存储到成员中
  41.         void setArgs(const std::string &str_args)
  42.         {
  43.             //key=val&key=val
  44.             std::vector<std::string> sub_args;
  45.             StrHelper::split(str_args, "&", sub_args);
  46.             for(auto& str : sub_args)
  47.             {
  48.                 size_t pos = str.find('=');
  49.                 std::string key = str.substr(0, pos);
  50.                 std::string val = str.substr(pos + 1);
  51.                 args[key] = val;
  52.             }
  53.         }
  54.         // 将args中的内容进行序列化后，返回一个字符串
  55.         std::string getArgs()
  56.         {
  57.             std::string result;
  58.             for(auto e = args.begin(); e != args.end(); e++)
  59.             {
  60.                 result = e->first + "=" + e->second + "&" + result;
  61.             }
  62.             return result; 
  63.         }
  64.     };
  65.  
  66.     // 2. 定义交换机数据持久化管理类 -- 数据存储在sqlite数据库中
  67.     class ExchangeMapper
  68.     {
  69.     public:
  70.         ExchangeMapper(const std::string &dbfile)
  71.             :_sqlite_helper(dbfile)
  72.         {
  73.             std::string path = FileHelper::parentDirectory(dbfile);
  74.             FileHelper::createDirectory(path);
  75.             assert(_sqlite_helper.open());
  76.             createTable();
  77.         }
  78.  
  79.         void createTable()
  80.         {
  81.             #define CREATE_TABLE "create table if not exists exchange_table(\
  82.             name varchar(32) primary key, \
  83.             type int,\
  84.             durable int, \
  85.             auto_delete int ,\
  86.             args varchar(128));"
  87.             bool ret = _sqlite_helper.exec(CREATE_TABLE, nullptr, nullptr);
  88.             if(ret == false)
  89.             {
  90.                 DLOG("创建交换机数据库失败");
  91.                 abort();  //直接异常退出程序
  92.             }
  93.         }
  94.  
  95.         void removeTable()
  96.         {
  97.             #define DROP_TABLE "drop table if exists exchange_table;"
  98.             bool ret = _sqlite_helper.exec(DROP_TABLE, nullptr, nullptr);
  99.             if(ret == false)
  100.             {
  101.                 DLOG("删除交换机数据库失败");
  102.                 abort();  //直接异常退出程序
  103.             }
  104.         }
  105.  
  106.         bool insert(Exchange::ptr &exp)
  107.         {
  108.             #define INSERT_TABLE "insert into exchange_table values('%s', %d, %d, %d, '%s');"
  109.             std::string args_str = exp->getArgs();
  110.             char sql_str[4096] = {0};
  111.             sprintf(sql_str, INSERT_TABLE, exp->name.c_str(), exp->type, exp->durable, exp->auto_delete, args_str.c_str());    
  112.             bool ret = _sqlite_helper.exec(sql_str, nullptr, nullptr);
  113.             if(ret == false)
  114.             {
  115.                 DLOG("插入数据库失败");
  116.                 return false;
  117.             }
  118.             return true;
  119.         }
  120.  
  121.         void remove(const std::string &name)
  122.         {
  123.             std::stringstream ss;
  124.             ss << "delete from exchange_table where name = ";
  125.             ss << "'" << name << "';";
  126.             _sqlite_helper.exec(ss.str(), nullptr, nullptr);
  127.         }
  128.  
  129.  
  130.         using ExchangeMap = std::unordered_map<std::string, Exchange::ptr>;
  131.         ExchangeMap recovery()
  132.         {
  133.             ExchangeMap result;
  134.             std::string sql = "select name, type, durable, auto_delete, args from exchange_table;";
  135.             _sqlite_helper.exec(sql.c_str(), SelectCallBack, &result);
  136.             return result;
  137.         }
  138.     private:
  139.         static int SelectCallBack(void* args, int numcol, char** row, char** fields)
  140.         {
  141.             ExchangeMap* result = (ExchangeMap*)args;
  142.             auto exp = std::make_shared<Exchange>();
  143.             exp->name = row[0];
  144.             exp->type = (mymq::ExchangeType)std::stoi(row[1]);
  145.             exp->durable = (bool)std::stoi(row[2]);
  146.             exp->auto_delete = (bool)std::stoi(row[3]);
  147.             if(row[4])
  148.             {
  149.                 exp->setArgs(row[4]);
  150.             }
  151.             result->insert(std::make_pair(exp->name, exp));
  152.             return 0;
  153.         }
  154.     private:
  155.         SqliteHelper _sqlite_helper;
  156.     };
  157.  
  158.     // 3. 定义交换机在内存管理类
  159.     class ExchangeManager
  160.     {
  161.     public:
  162.         using ptr = std::shared_ptr<ExchangeManager>;
  163.         ExchangeManager(const std::string &dbfile)
  164.             :_mapper(dbfile)
  165.         {
  166.             _exchanges = _mapper.recovery();
  167.         }
  168.         // 声明交换机
  169.         bool declareExchange(const std::string &name, ExchangeType type, bool durable, bool auto_delete, const google::protobuf::Map<std::string, std::string> &args)
  170.         {
  171.             std::unique_lock<std::mutex> lock(_mutex);
  172.             auto it = _exchanges.find(name);
  173.             if(it != _exchanges.end())
  174.             {
  175.                 //如果交换机已经存在，那就直接返回，不需要重复新增
  176.                 return true;
  177.             }
  178.             auto exp = std::make_shared<Exchange>(name, type, durable, auto_delete, args);
  179.             if(durable == true)
  180.             {
  181.                 bool ret = _mapper.insert(exp);
  182.                 if(ret == false)
  183.                 {
  184.                     return false;
  185.                 }
  186.             }
  187.             _exchanges.insert(std::make_pair(name, exp));
  188.             return true;
  189.         }
  190.         // 删除交换机
  191.         void deleteExchange(const std::string &name)
  192.         {
  193.             std::unique_lock<std::mutex> lock(_mutex);
  194.             auto it = _exchanges.find(name);
  195.             if(it == _exchanges.end())
  196.             {
  197.                 //如果交换机不存在，那就直接返回
  198.                 return ;
  199.             }
  200.             if(it->second->durable == true)
  201.                 _mapper.remove(name);
  202.             _exchanges.erase(name);
  203.         }
  204.  
  205.         Exchange::ptr selectExchange(const std::string& name)
  206.         {
  207.             std::unique_lock<std::mutex> lock(_mutex);
  208.             auto it = _exchanges.find(name);
  209.             if(it == _exchanges.end())
  210.             {
  211.                 //如果交换机不存在，那就直接返回
  212.                 return nullptr;
  213.             }
  214.             return it->second;
  215.         }
  216.  
  217.         // 判断交换机是否存在
  218.         bool exists(const std::string &name)
  219.         {
  220.             std::unique_lock<std::mutex> lock(_mutex);
  221.             auto it = _exchanges.find(name);
  222.             if(it == _exchanges.end())
  223.             {
  224.                 //如果交换机不存在，那就直接返回
  225.                 return false;
  226.             }
  227.             return true;
  228.         }
  229.  
  230.         size_t size()
  231.         {
  232.             std::unique_lock<std::mutex> lock(_mutex);
  233.             return _exchanges.size();
  234.         }
  235.  
  236.         // 清理所有交换机数据
  237.         void clear()
  238.         {
  239.             std::unique_lock<std::mutex> lock(_mutex);
  240.             _mapper.removeTable();
  241.             _exchanges.clear();
  242.         }
  243.  
  244.     private:
  245.         std::mutex _mutex;
  246.         ExchangeMapper _mapper;
  247.         std::unordered_map<std::string, Exchange::ptr> _exchanges;
  248.     };
  249.  
  250. }
  251.

队列数据管理

要管理的数据

  1. 队列名称：唯一的标识
  2. 持久化存储标志：决定了是否将队列信息持久化存储起来，决定了重启后，这个队列是否存在
  3. 是否独占标志：独占就指的是，只有当前客户端自己能够订阅队列消息
  4. 自动删除标志：当订阅了当前队列的所有客户端退出后，是否删除队列

要管理的操作

  1. 创建队列
  2. 删除队列
  3. 获取指定队列信息
  4. 获取队列数据
  5. 获取所有队列名称：当系统重启后，需要重新加载数据，加载历史消息（消息以队列为单元存储在文件中）而加载消息需要直到队列名称，因为后边消息存储的时候，存储文件以队列名称进行的取名

代码展示

  1. #ifndef __M_QUEUE_H__
  2. #define __M_QUEUE_H__
  3.  
  4.  
  5. #include "../mqcommon/helper.hpp"
  6. #include "../mqcommon/logger.hpp"
  7. #include "../mqcommon/msg.pb.h"
  8. #include <iostream>
  9. #include <unordered_map>
  10. #include <memory>
  11. #include <mutex>
  12.  
  13. namespace mymq
  14. {
  15.     struct MsgQueue
  16.     {
  17.         using ptr = std::shared_ptr<MsgQueue>;   // 放在这个位置合适吗？
  18.         std::string name;
  19.         bool durable;
  20.         bool exclusive;
  21.         bool auto_delete;
  22.         google::protobuf::Map<std::string, std::string> args;
  23.  
  24.         MsgQueue()
  25.         {}
  26.  
  27.         MsgQueue(const std::string qname, bool qdurable, bool qexclusive, bool qauto_delete, const google::protobuf::Map<std::string, std::string> &qargs)
  28.             :name(qname),
  29.             durable(qdurable),
  30.             exclusive(qexclusive),
  31.             auto_delete(qauto_delete),
  32.             args(qargs)
  33.         {}
  34.  
  35.         void setArgs(const std::string& str_args)
  36.         {
  37.             std::vector<std::string> sub_args;
  38.             StrHelper::split(str_args, "&", sub_args);
  39.             for(auto sub_string : sub_args)
  40.             {
  41.                 size_t pos = sub_string.find("=");
  42.                 std::string key = sub_string.substr(0, pos);
  43.                 std::string value = sub_string.substr(pos + 1);
  44.                 args[key] = value;
  45.             }
  46.         }
  47.  
  48.         std::string getArgs()
  49.         {
  50.             std::string result;
  51.             for(auto str = args.begin(); str != args.end(); str++)
  52.             {
  53.                 result = str->first + "=" + str->second + result;
  54.             }
  55.             return result;
  56.         }
  57.     };
  58.  
  59.     using QueueMap = std::unordered_map<std::string, MsgQueue::ptr>;
  60.     class MsgQueueMapper
  61.     {
  62.     public:
  63.         MsgQueueMapper(const std::string& dbfile)
  64.             :_sqlite_helper(dbfile)
  65.         {
  66.             std::string path = FileHelper::parentDirectory(dbfile);
  67.             bool ret = FileHelper::createDirectory(path);
  68.             if(ret == false)
  69.             {
  70.                 ELOG("创建父目录失败: %s", strerror(errno));
  71.             }
  72.             _sqlite_helper.open();  //创建sqlite目录
  73.             createTable();
  74.         }
  75.  
  76.         void createTable()
  77.         {
  78.             std::stringstream sql;
  79.             sql << "create table if not exists queue_table(";
  80.             sql << "name varchar(32) primary key, ";
  81.             sql << "durable int, ";
  82.             sql << "exclusive int, ";
  83.             sql << "auto_delete int, ";
  84.             sql << "args varchar(128));";
  85.             _sqlite_helper.exec(sql.str(), nullptr, nullptr);
  86.         }
  87.  
  88.         void removeTable()
  89.         {
  90.             std::stringstream sql;
  91.             sql << "drop table if exists queue_table;";
  92.             _sqlite_helper.exec(sql.str(), nullptr, nullptr);
  93.         }
  94.  
  95.         bool insert(MsgQueue::ptr& queue)
  96.         {
  97.             std::stringstream sql;
  98.             sql << "insert into queue_table values(";
  99.             sql << "'" << queue->name << "',";
  100.             sql << queue->durable << ", ";
  101.             sql << queue->exclusive << ", ";
  102.             sql << queue->auto_delete << ", ";
  103.             sql << "'" << queue->getArgs() << "');";
  104.             return _sqlite_helper.exec(sql.str(), nullptr, nullptr);
  105.         }
  106.  
  107.         void remove(const std::string& name)
  108.         {
  109.             std::stringstream sql;
  110.             sql << "delete from queue_table where name = ";
  111.             sql << "'" << name << "';";
  112.             _sqlite_helper.exec(sql.str(), nullptr, nullptr);
  113.         }
  114.  
  115.  
  116.         QueueMap recovery()
  117.         {
  118.             QueueMap result;
  119.             std::stringstream sql;
  120.             sql << "select name, durable, exclusive, auto_delete, args from queue_table;";
  121.             _sqlite_helper.exec(sql.str(), SelectCallBack, &result);
  122.             return result;
  123.         }
  124.  
  125.     private:
  126.         static int SelectCallBack(void* args, int numcol, char** row, char** fields)
  127.         {
  128.             QueueMap* result = (QueueMap*)args;
  129.             MsgQueue::ptr mqp = std::make_shared<MsgQueue>();
  130.             mqp->name = row[0];
  131.             mqp->durable = std::stoi(row[1]);
  132.             mqp->exclusive = std::stoi(row[2]);
  133.             mqp->auto_delete = std::stoi(row[3]);
  134.             if(row[4]) mqp->setArgs(row[4]);
  135.             result->insert(std::make_pair(mqp->name, mqp));
  136.             return 0;
  137.         }
  138.     private:
  139.         SqliteHelper _sqlite_helper;
  140.     };
  141.  
  142.  
  143.     class MsgQueueManager
  144.     {
  145.     public:
  146.         using ptr = std::shared_ptr<MsgQueueManager>;
  147.         MsgQueueManager(const std::string& dbfile)
  148.             :_mapper(dbfile)
  149.         {
  150.             _msg_queue = _mapper.recovery();
  151.         }
  152.  
  153.         bool declareQueue(const std::string& name, bool durable, bool exclusive, bool auto_delete, const google::protobuf::Map<std::string, std::string>& args)
  154.         {
  155.             std::unique_lock<std::mutex> lock(_mutex);
  156.             auto it = _msg_queue.find(name);
  157.             if(it != _msg_queue.end())
  158.             {
  159.                 // 队列存在，直接返回true
  160.                 return true;
  161.             }
  162.             MsgQueue::ptr mqp = std::make_shared<MsgQueue>();
  163.             mqp->name = name;
  164.             mqp->durable = durable;
  165.             mqp->exclusive = exclusive;
  166.             mqp->auto_delete = auto_delete;
  167.             mqp->args = args;
  168.             if(durable ==true) //需要进行持久化存储
  169.             {
  170.                 bool ret = _mapper.insert(mqp);
  171.                 if(ret == false) return false;
  172.             }
  173.             _msg_queue.insert(make_pair(mqp->name, mqp));
  174.             return true;
  175.         }
  176.  
  177.         void deleteQueue(const std::string& name)
  178.         {
  179.             std::unique_lock<std::mutex> lock(_mutex);
  180.             auto it = _msg_queue.find(name);
  181.             if(it == _msg_queue.end())
  182.             {
  183.                 DLOG("未找到相关信息");
  184.                 return ;
  185.             }
  186.             if(it->second->durable == true)
  187.             {
  188.                 _mapper.remove(name);
  189.             }
  190.             _msg_queue.erase(name);
  191.         }
  192.  
  193.         MsgQueue::ptr selectQueue(const std::string& name)
  194.         {
  195.             std::unique_lock<std::mutex> lock(_mutex);
  196.             auto it = _msg_queue.find(name);
  197.             if(it == _msg_queue.end())
  198.             {
  199.                 DLOG("未找到相关信息");
  200.                 return MsgQueue::ptr();
  201.             }
  202.             return it->second;
  203.         }
  204.  
  205.         QueueMap AllQueues()
  206.         {
  207.             std::unique_lock<std::mutex> lock(_mutex);
  208.             return _msg_queue;
  209.         }
  210.  
  211.         bool exists(const std::string& name)
  212.         {
  213.             std::unique_lock<std::mutex> lock(_mutex);
  214.             auto it = _msg_queue.find(name);
  215.             if(it == _msg_queue.end())
  216.             {
  217.                 return false;
  218.             }
  219.             return true;
  220.         }
  221.  
  222.         size_t size()
  223.         {
  224.             std::unique_lock<std::mutex> lock(_mutex);
  225.             return _msg_queue.size();
  226.         }
  227.  
  228.         void clear()
  229.         {
  230.             std::unique_lock<std::mutex> lock(_mutex);
  231.             _msg_queue.clear();
  232.         }
  233.  
  234.     private:
  235.         std::mutex _mutex;
  236.         MsgQueueMapper _mapper;
  237.         QueueMap _msg_queue;
  238.     };
  239. }
  240.  
  241. #endif

绑定数据管理模块

描述将哪个队列与哪个交换机绑定到了一起

管理的数据

  • 交换机名称
  • 队列名称
  • binding_key：绑定密钥----描述在交换机的主题交换&直接交换的消息发布匹配规则

由数字， 字符，_，#， * 组成 binding_key， 例如：news.music.#

管理的操作

1. 添加绑定

2. 解除绑定

3. 获取交换机相关的所有绑定信息

        删除交换机的时候，要删除相关绑定信息

        当消息发布到交换机中，交换机得通过这些信息来将消息发布到指定队列

4. 获取队列相关的所有绑定信息

        删除队列的时候，要删除相关的绑定信息

5. 获取绑定信息数量

代码展示

  1. #ifndef _M_BINDING_H_
  2. #define _M_BINDING_H_
  3.  
  4. #include "../mqcommon/helper.hpp"
  5. #include "../mqcommon/msg.pb.h"
  6. #include "../mqcommon/logger.hpp"
  7. #include <iostream>
  8. #include <unordered_map>
  9. #include <memory>
  10. #include <mutex>
  11.  
  12. namespace mymq
  13. {
  14.     struct Binding
  15.     {
  16.         using ptr = std::shared_ptr<Binding>;
  17.         std::string exchange_name;
  18.         std::string msgqueue_name;
  19.         std::string binding_key;
  20.  
  21.         Binding() {}
  22.  
  23.         Binding(const std::string &ename, const std::string &msgqname, const std::string &key)
  24.             : exchange_name(ename),
  25.               msgqueue_name(msgqname),
  26.               binding_key(key)
  27.         {
  28.         }
  29.     };
  30.     // 队列与绑定信息是一一对应的(因为是给某一个交换机去绑定队列，因此一个交换机可能会有多个队列的绑定信息)
  31.     // 因此先定义一个队列名，与绑定信息的映射信息，这个是为了方便通过队名查找绑定信息
  32.     using MsgQueueBindingMap = std::unordered_map<std::string, Binding::ptr>;
  33.     // 然后定义一个交换机名称与队列绑定信息之间的映射关系，这个map包含了所有的绑定信息，并且以交换机为单元进行区分
  34.     using BindingMap = std::unordered_map<std::string, MsgQueueBindingMap>;
  35.  
  36.     // 采用上面两种结构，则删除交换机相关绑定信息的时候，不仅要删除交换机映射，还需要删除对应队列中的映射，否则对象得不到释放
  37.     class BindingMapper
  38.     {
  39.     public:
  40.         BindingMapper(const std::string& dbfile)
  41.             :_sqlite_helper(dbfile)
  42.         {
  43.             std::string path = FileHelper::parentDirectory(dbfile);
  44.             FileHelper::createDirectory(path);
  45.             _sqlite_helper.open();
  46.             createTable();
  47.         }
  48.  
  49.         void createTable()
  50.         {
  51.             std::stringstream sql;
  52.             sql << "create table if not exists binding_table(";
  53.             sql << "exchange_name varchar(32), ";
  54.             sql << "msgqueue_name varchar(32), ";
  55.             sql << "binding_key varchar(128));";
  56.             assert(_sqlite_helper.exec(sql.str(), nullptr, nullptr));
  57.         }
  58.  
  59.         void removeTable()
  60.         {
  61.             std::stringstream sql;
  62.             sql << "drop table if exists binding_table; ";
  63.             assert(_sqlite_helper.exec(sql.str(), nullptr, nullptr));
  64.         }
  65.  
  66.  
  67.         bool insert(Binding::ptr& binding)
  68.         {
  69.             std::stringstream sql; 
  70.             sql << "insert into binding_table values(";
  71.             sql << "'" << binding->exchange_name <<"',";
  72.             sql << "'" << binding->msgqueue_name << "',";
  73.             sql << "'" << binding->binding_key << "');";
  74.             
  75.             return _sqlite_helper.exec(sql.str(), nullptr, nullptr);
  76.         }
  77.  
  78.         bool remove(const std::string& ename, const std::string& qname)
  79.         {
  80.             std::stringstream sql;
  81.             sql << "delete from binding_table where ";
  82.             sql << "exchange_name = '" << ename << "' and ";
  83.             sql << "msgqueue_name = '" << qname << "';";
  84.             return _sqlite_helper.exec(sql.str(), nullptr, nullptr); 
  85.         }
  86.  
  87.         void removeExchangeBindings(const std::string& ename)
  88.         {
  89.             std::stringstream sql;
  90.             sql << "delete from binding_table where ";
  91.             sql << "exchange_name = '" << ename << "';";
  92.             _sqlite_helper.exec(sql.str(), nullptr, nullptr); 
  93.         }
  94.  
  95.         void removeMsgQueueBindings(const std::string& qname)
  96.         {
  97.             std::stringstream sql;
  98.             sql << "delete from binding_table where ";
  99.             sql << "msgqueue_name = '" << qname << "';";
  100.             _sqlite_helper.exec(sql.str(), nullptr, nullptr); 
  101.         }
  102.  
  103.         BindingMap recovery()
  104.         {
  105.             BindingMap result;
  106.             std::string sql = "select exchange_name, msgqueue_name, binding_key from binding_table;";
  107.             _sqlite_helper.exec(sql, SelectCallBack, &result);
  108.             return result;
  109.         }
  110.  
  111.     private:
  112.         static int SelectCallBack(void* args, int numcol, char** row, char** fileds)
  113.         {
  114.             BindingMap* result = (BindingMap*)args;
  115.             Binding::ptr bq = std::make_shared<Binding>();
  116.             bq->exchange_name = row[0];
  117.             bq->msgqueue_name = row[1];
  118.             bq->binding_key = row[2];
  119.             //为了防止 交换相关的绑定信息，因此不能直接创建队列映射，这样会覆盖历史数据
  120.             // 因此得先获取交换机对应的映射对象，往里边添加数据
  121.             // 但是，若这时候没有交换机对应的映射信息，因此这里的获取要使用引用（会保证不存在则自动创建）
  122.             MsgQueueBindingMap& qbm = (*result)[bq->exchange_name];
  123.             qbm.insert(std::make_pair(bq->msgqueue_name, bq));
  124.             return 0;
  125.         }
  126.     private:
  127.         SqliteHelper _sqlite_helper;
  128.     };
  129.  
  130.     class BindingManager
  131.     {
  132.     public:
  133.         using ptr = std::shared_ptr<BindingManager>;
  134.         BindingManager(const std::string dbfile)
  135.             :_mapper(dbfile)
  136.         {
  137.             _bindmap = _mapper.recovery();
  138.         }
  139.  
  140.         bool bind(const std::string& ename, const std::string& qname, const std::string& key ,bool durable)
  141.         {
  142.             std::unique_lock<std::mutex> lock(_mutex);
  143.             auto it = _bindmap.find(ename);
  144.             if(it != _bindmap.end() && it->second.find(qname) != it->second.end())
  145.             {
  146.                 return true;
  147.             }
  148.             //绑定是否是持久化
  149.             Binding::ptr bp = std::make_shared<Binding>(ename, qname, key);
  150.             // 磁盘数据持久化
  151.             if(durable == true)
  152.             {
  153.                 bool ret = _mapper.insert(bp);
  154.                 if(ret == false)
  155.                 {
  156.                     //未添加成功
  157.                     return false;
  158.                 }
  159.             }
  160.             // 一个交换机对应着多条绑定信息，所以先确定交换机，再将相关绑定信息插进去
  161.             MsgQueueBindingMap& mqbm = _bindmap[ename];
  162.             mqbm.insert(std::make_pair(bp->msgqueue_name, bp));
  163.             return true;
  164.         }
  165.  
  166.         void unBind(const std::string& ename, const std::string& qname)
  167.         {
  168.             std::unique_lock<std::mutex> lock(_mutex);
  169.             auto itBindMap = _bindmap.find(ename);
  170.             if(itBindMap == _bindmap.end())
  171.             {
  172.                 return;
  173.             }
  174.             auto itMsgQueueBindings = itBindMap->second.find(qname);
  175.             if(itMsgQueueBindings == itBindMap->second.end())
  176.             {
  177.                 return ;
  178.             }
  179.             _mapper.remove(ename, qname);
  180.             _bindmap[ename].erase(qname);
  181.         }
  182.  
  183.         void removeExchangeBindings(const std::string& ename)
  184.         {
  185.             std::unique_lock<std::mutex> lock(_mutex);
  186.             _mapper.removeExchangeBindings(ename);
  187.             _bindmap.erase(ename);
  188.         }
  189.  
  190.         void removeMsgqueueBindings(const std::string& qname)
  191.         {
  192.             std::unique_lock<std::mutex> lock(_mutex);
  193.             _mapper.removeMsgQueueBindings(qname);
  194.             for(auto it = _bindmap.begin(); it != _bindmap.end(); it++)
  195.             {
  196.                 it->second.erase(qname);
  197.             }
  198.         }
  199.  
  200.         MsgQueueBindingMap getExchangeBindings(const std::string& ename)
  201.         {
  202.             std::unique_lock<std::mutex> lock(_mutex);
  203.             auto it = _bindmap.find(ename);
  204.             if(it == _bindmap.end())
  205.             {
  206.                 return MsgQueueBindingMap();
  207.             }
  208.             return it->second;
  209.         }
  210.  
  211.         Binding::ptr getBinding(const std::string& ename, const std::string& qname)
  212.         {
  213.             std::unique_lock<std::mutex> lock(_mutex);
  214.             auto it_bindmap = _bindmap.find(ename);
  215.             if(it_bindmap == _bindmap.end())
  216.             {
  217.                 return Binding::ptr();
  218.             }
  219.             auto it_MsgQueueBindings = it_bindmap->second.find(qname);
  220.             if(it_MsgQueueBindings == it_bindmap->second.end())
  221.             {
  222.                 return Binding::ptr();
  223.             }
  224.             return it_MsgQueueBindings->second;
  225.         }
  226.  
  227.         bool exists(const std::string& ename, const std::string& qname)
  228.         {
  229.             std::unique_lock<std::mutex> lock(_mutex);
  230.             auto it_bindmap = _bindmap.find(ename);
  231.             if(it_bindmap == _bindmap.end())
  232.             {
  233.                 return false;
  234.             }
  235.             auto it_MsgQueueBindings = it_bindmap->second.find(qname);
  236.             if(it_MsgQueueBindings ==it_bindmap->second.end())
  237.             {
  238.                 return false;
  239.             }
  240.             return true;
  241.         }
  242.  
  243.         size_t size()
  244.         {
  245.             std::unique_lock<std::mutex> lock(_mutex);
  246.             size_t total_size = 0;
  247.             for(auto it = _bindmap.begin(); it != _bindmap.end(); it++)
  248.             {
  249.                 total_size += it->second.size();
  250.             }
  251.             return total_size;
  252.         }
  253.  
  254.         void clear()
  255.         {
  256.             std::unique_lock<std::mutex> lock(_mutex);
  257.             _mapper.removeTable();
  258.             _bindmap.clear();
  259.         }
  260.  
  261.  
  262.         
  263.     private:
  264.         std::mutex _mutex;
  265.         BindingMapper _mapper;
  266.         BindingMap _bindmap;
  267.     };
  268. }
  269.  
  270. #endif

现在虚拟机模块的所有需要管理的数据全部都安排成功了，接下来是虚拟机的代码展示

虚拟机代码展示

  1. #ifndef __M_HOST_H__
  2. #define __M_HOST_H__
  3.  
  4. #include "mq_exchange.hpp"
  5. #include "mq_queue.hpp"
  6. #include "mq_binding.hpp"
  7. #include "mq_message.hpp"
  8.  
  9. namespace mymq
  10. {
  11.     class VirtualHost
  12.     {
  13.         public:
  14.             using ptr = std::shared_ptr<VirtualHost>;
  15.             VirtualHost(const std::string& hname, const std::string& basedir, const std::string& dbfile)
  16.                 :_host_name(hname),
  17.                 _emp(std::make_shared<ExchangeManager>(dbfile)),
  18.                 _mqmp(std::make_shared<MsgQueueManager>(dbfile)),           
  19.                 _bmp(std::make_shared<BindingManager>(dbfile)),
  20.                 _mmp(std::make_shared<MessageManager>(basedir))
  21.             {}
  22.  
  23.             bool declareExchange(const std::string& name, ExchangeType type, bool durable, bool auto_delete, const google::protobuf::Map<std::string, std::string>& args)
  24.             {
  25.                 return _emp->declareExchange(name, type, durable, auto_delete, args);
  26.             }
  27.  
  28.             void deleteExchange(const std::string& name)
  29.             {
  30.                 //删除交换机的时候，需要将交换机相关的绑定信息也删除掉
  31.                 _bmp->removeExchangeBindings(name);
  32.                 return _emp->deleteExchange(name);
  33.             }
  34.  
  35.             bool existsExchange(const std::string& name)
  36.             {
  37.                 return _emp->exists(name);
  38.             }
  39.  
  40.             Exchange::ptr selectExchange(const std::string& ename)
  41.             {
  42.                 return _emp->selectExchange(ename);
  43.             }
  44.  
  45.             bool declareQueue(const std::string qname, bool qdurable, bool qexclusive, bool qauto_delete, const google::protobuf::Map<std::string, std::string> args)
  46.             {
  47.                 //初始化队列消息句柄（消息的存储管理）
  48.                 //队列的创建
  49.                 _mmp->initQueueMessage(qname);
  50.                 return _mqmp->declareQueue(qname, qdurable, qexclusive, qauto_delete, args);
  51.             }
  52.  
  53.             void deleteQueue(const std::string& qname)
  54.             {
  55.                 //删除的时候队列相关的数据有两个：队列的消息，队列的绑定消息
  56.                 //删除队列里的消息
  57.                 _mmp->destroyQueueMessage(qname);
  58.                 //删除队列的绑定消息
  59.                 _bmp->removeMsgqueueBindings(qname);
  60.                 //删除队列
  61.                 _mqmp->deleteQueue(qname);
  62.             }
  63.  
  64.             bool existsQueue(const std::string qname)
  65.             {
  66.                 return _mqmp->exists(qname);
  67.             }
  68.  
  69.             QueueMap allqueue()
  70.             {
  71.                 return _mqmp->AllQueues();
  72.             }
  73.  
  74.             bool bind(const std::string& ename, const std::string& qname, const std::string& key)
  75.             {
  76.                 Exchange::ptr ep = _emp->selectExchange(ename);
  77.                 if(ep.get() == nullptr)
  78.                 {
  79.                     DLOG("进行队列绑定失败，交换机%s不存在", ename.c_str());
  80.                     return false;
  81.                 }
  82.                 MsgQueue::ptr mqp = _mqmp->selectQueue(qname);
  83.                 if(mqp.get() == nullptr)
  84.                 {
  85.                     DLOG("进行队列绑定失败，队列%s不存在", qname.c_str());
  86.                 }
  87.                 return _bmp->bind(ename, qname,key, ep->durable && mqp->durable);
  88.             }
  89.  
  90.             void unBind(const std::string& ename, const std::string& qname)
  91.             {
  92.                 return _bmp->unBind(ename, qname);
  93.             }
  94.  
  95.             MsgQueueBindingMap exchangeBindings(const std::string& ename)
  96.             {
  97.                 return _bmp->getExchangeBindings(ename);
  98.             }
  99.  
  100.             bool existsBinding(const std::string& ename, const std::string& qname)
  101.             {
  102.                 return _bmp->exists(ename, qname);
  103.             }
  104.             
  105.             // 消息发布函数
  106.             bool basicPublish(const std::string& qname, BasicProperties* bp, const std::string& body)
  107.             {
  108.                 MsgQueue::ptr mqp = _mqmp->selectQueue(qname);
  109.                 if(mqp.get() == nullptr)
  110.                 {
  111.                     DLOG("发布消息失败， 队列%s不存在", qname.c_str());
  112.                     return false;
  113.                 }
  114.                 return _mmp->insert(qname, bp, body, mqp->durable);
  115.             }
  116.  
  117.             //获取消息
  118.             MessagePtr basicConsume(const std::string& name)
  119.             {
  120.                 return _mmp->front(name);
  121.             }
  122.  
  123.             // 消息应答
  124.             void basicAck(const std::string& qname, const std::string& msgid)
  125.             {
  126.                 return _mmp->ack(qname, msgid);
  127.             }
  128.  
  129.             void clear()
  130.             {
  131.                 _emp->clear();
  132.                 _mqmp->clear();
  133.                 _bmp->clear();
  134.                 _mmp->clear();
  135.             }
  136.  
  137.         private:
  138.             std::string _host_name;
  139.             ExchangeManager::ptr _emp;
  140.             MsgQueueManager::ptr _mqmp;
  141.             BindingManager::ptr _bmp;
  142.             MessageManager::ptr _mmp;
  143.     };
  144. }
  145.  
  146. #endif

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/从前慢现在也慢/article/detail/983363
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号