首页 > 其他分享 >exchange.hpp交换机模块

exchange.hpp交换机模块

时间：2024-08-15 20:53:16浏览次数：6  
标签：std args name Exchange hpp 交换机 ptr exchange

目录

一.Exchange模块介绍

二.Exchange类的实现

交换机类型

交换机类型的定义

Exchange类的相关接口

主要成员变量

构造函数

公有成员函数

setArgs 函数：

getArgs 函数：

三.ExchangeMapper类的实现

主要成员变量

构造函数

公有成员函数

createTable

dropTable

insert

remove

recover

exchangeMapCb

四.ExchangeManager类的实现

主要成员变量

构造函数

成员函数(增删查)

declareExchange 函数：

removeExchange 函数：

selectExchange 函数：

exists 函数：

clear 函数：

size 函数：

五.exchange.hpp所有代码

一.Exchange模块介绍

二.Exchange类的实现

交换机类型

对于 RabbitMQ 来说, 主要⽀持四种交换机类型： 
• Direct 
• Fanout 
• Topic 
• Header 
其中 Header 这种⽅式⽐较复杂, ⽐较少⻅。常⽤的是前三种交换机类型，项⽬中也主要实现这三种 
• Direct: ⽣产者发送消息时, 指定被该交换机绑定的所有队列中满足(RoutingKey==BindingKey)的
• Fanout: ⽣产者发送的消息会被复制到该交换机绑定的所有队列中  
• Topic: 绑定队列到交换机上时, 指定⼀个字符串为 BindingKey。发送消息指定⼀个字符串为 
RoutingKey。当 RoutingKey 和 BindingKey满⾜⼀定的匹配条件的时候, 则把消息投递到指定队列

交换机类型的定义

// 交换机类型
enum ExchangeType
{
    UNKNOWNTYPE = 0;
    DIRECT = 1;
    FANOUT = 2;
    TOPIC = 3;
};

// 消息投递模式（决定是否持久化）
enum DeliverMode
{
    NORMAL = 0;
    UNDURABLE = 1;
    DURABLE = 2;
};

Exchange类的相关接口

主要成员变量

  • _name：std::string 类型，表示交换机的名称。
  • _type：msg::ExchangeType 类型，表示交换机的类型。
  • _durable：bool 类型，表示交换机是否持久化。
  • _auto_del：bool 类型，表示交换机是否自动删除。
  • _args：google::protobuf::Map<std::string, std::string> 类型，用于存储交换机的其他参数。这个 Map 用于键值对存储交换机的自定义属性。
std::string _name;                                     // 交换机名称
msg::ExchangeType _type;                               // 交换机类型
bool _durable;                                         // 是否持久化
bool _auto_del;                                        // 是否自动删除
google::protobuf::Map<std::string, std::string> _args; // 其它参数

构造函数

   Exchange() {}
        Exchange(const std::string &name, msg::ExchangeType type,
                 bool durable,
                 bool auto_del,
                 const google::protobuf::Map<std::string,std::string> &args)
            : _name(name),
              _type(type),
              _durable(durable),
              _auto_del(auto_del),
              _args(args) {}

公有成员函数

setArgs 函数：
  • 用于从格式化字符串中解析出参数并存储到 _args 成员变量中。格式化字符串的格式是 key=val&key=val...，多个键值对通过 & 符号分隔。
  • 实现思路：首先将字符串按 & 进行分割，然后对于每个键值对，按 = 分割为键和值，并存入 _args 中。如果格式错误（找不到 =），则记录错误日志并终止程序。
void setArgs(const std::string &str_args)
{
    std::vector<std::string> args;
    size_t sz = StrHelper::split(str_args, "&", args);
    for (auto &kv : args)
    {
        size_t pos = kv.find("=");
        if (pos == std::string::npos)
        {
            ELOG("Exchange args format error:%s", kv.c_str());
            assert(0);
        }
        std::string key = kv.substr(0, pos);
        std::string val = kv.substr(pos + 1);
        _args[key] = val;
    }
}
getArgs 函数：
  • 将 _args 成员变量中的kv参数转换为格式化字符串，格式同上。它遍历 _args Map，将每个键值对拼接为字符串并以 & 符号分隔。
std::string getArgs()
{
    std::string ret;
    for (auto &kv : _args)
    {
        ret += kv.first + "=" + kv.second + "&";
    }
    return ret;
}

三.ExchangeMapper类的实现

ExchangeMapper 类负责将 Exchange 对象的属性持久化到数据库中，并提供了对这些数据的插入、删除和查询等操作。它通过 SqliteHelper 类与 SQLite 数据库进行交互，确保 Exchange 对象的信息能够在数据库中保存和恢复。

主要成员变量

  • sql_helper：SqliteHelper 类型，用于管理与 SQLite 数据库的连接和操作

构造函数

  • ExchangeMapper(const std::string &dbname)：构造函数接受一个数据库文件的路径 dbname，并使用 SqliteHelper 进行数据库连接的初始化。如果数据库文件所在路径不存在，则会自动创建。同时，构造函数会调用 createTable 方法创建存储 Exchange 信息的表。

公有成员函数

createTable

用于在数据库中创建存储 Exchange 对象信息的表 exchange_table。表中包含以下字段：

args：存储其他参数的字符串。

auto_del：是否自动删除。

durable：是否持久化。

type：交换机类型。

name：交换机名称，作为主键。

void createTable()
{
    std::stringstream sql;
    sql << "create table if not exists exchange_table\
    ('name' varchar(64) primary key,\
    type int,durable int,auto_del int,args varchar(128));";
    if (!sql_helper.exec(sql.str(), nullptr, nullptr))
    {
        ELOG("ExchangeMapper create table error!");
        assert(0);
    }
}
dropTable
  • 用于删除 exchange_table 表
void dropTable()
{
    std::stringstream sql;
    sql << "drop table if exists exchange_table;";
    if (!sql_helper.exec(sql.str(), nullptr, nullptr))
    {
        ELOG("ExchangeMapper drop table error!");
        assert(0);
    }
}
insert
  • 将一个 Exchange 对象插入到 exchange_table 表中。它将对象的各个成员变量转化为 SQL 语句中的对应字段值，并执行插入操作。
void insert(Exchange::ptr &exchange_ptr)
{
    std::stringstream sql;
    sql << "insert into exchange_table values('" << exchange_ptr->_name << "'," << exchange_ptr->_type << "," << exchange_ptr->_durable << ","
        << exchange_ptr->_auto_del << ",'"
        << exchange_ptr->getArgs() << "');";
    if (!sql_helper.exec(sql.str(), nullptr, nullptr))
    {
        ELOG("ExchangeMapper insert error!");
        assert(0);
    }
}
remove
  • 从 exchange_table 表中删除指定名称的 Exchange 对象
void remove(const std::string &name)
{
    std::stringstream sql;
    sql << "delete from exchange_table where name='" << name << "';";
    if (!sql_helper.exec(sql.str(), nullptr, nullptr))
    {
        ELOG("ExchangeMapper remove error!");
        assert(0);
    }
}
recover
  • 从 exchange_table 表中查询所有的 Exchange 对象，并将它们恢复为 Exchange 对象存储在一个哈希表（exchange_map）中返回。该函数使用一个回调函数 exchangeMapCb 处理查询结果，将每一行记录转化为 Exchange 对象。

using exchange_map = std::unordered_map<std::string, Exchange::ptr>;

exchange_map recover()
{
    exchange_map ret;
    std::stringstream sql;
    sql << "select * from exchange_table;";
    if (!sql_helper.exec(sql.str(), exchangeMapCb, &ret))
    {
        ELOG("ExchangeMapper recover error!");
        assert(0);
    }
    return ret;
}
exchangeMapCb
  • 该静态回调函数在 recover 函数中被调用，用于将数据库查询结果中的每一行转化为一个 Exchange 对象，并将其存储到 exchange_map 中。
static int exchangeMapCb(void *args, int rows, char **ret, char **fields)
{
    Exchange::ptr exchange_ptr = std::make_shared<Exchange>();
    exchange_map *map_ptr = (exchange_map *)args;
    exchange_ptr->_name = ret[0];
    exchange_ptr->_type = (msg::ExchangeType)std::stoi(ret[1]);
    exchange_ptr->_durable = std::stoi(ret[2]);
    exchange_ptr->_auto_del = std::stoi(ret[3]);
    if (ret[4])
    {
        exchange_ptr->setArgs(ret[4]);
    }
    map_ptr->insert(std::make_pair(exchange_ptr->_name, exchange_ptr));
    return 0;
}

四.ExchangeManager类的实现

ExchangeManager 类用于管理系统中的 Exchange 对象，并提供声明、删除、查询、判断存在性以及清理等功能。该类在内部通过一个 ExchangeMapper 实例与数据库交互，以实现 Exchange 对象的持久化和恢复。

主要成员变量

  • _mutex：std::mutex 类型，用于确保多线程环境下对 _exchanges 数据结构的安全访问。
  • _mapper：ExchangeMapper 类型，用于与数据库进行交互，负责 Exchange 对象的持久化操作。
  • _exchanges：exchange_map 类型，用于在内存中存储所有当前管理的 Exchange 对象

构造函数

  • ExchangeManager(const std::string &dbname)：构造函数接受一个数据库文件的路径 dbname，通过 ExchangeMapper 从数据库中恢复已存储的 Exchange 对象到内存中的 _exchanges 哈希表。

成员函数(增删查)

declareExchange 函数：

声明一个 Exchange 对象，如果该名称的 Exchange 已存在则直接返回 true。如果不存在，则创建一个新的 Exchange 对象并插入到 _exchanges 中，并在持久化标志 _durable 为 true 时，将其插入到数据库中。

bool declareExchange(const std::string &name, msg::ExchangeType type,
                     bool durable,
                     bool auto_del,
                     google::protobuf::Map<std::string, std::string> args)
{
    std::unique_lock<std::mutex> lock(_mutex);
    auto it = _exchanges.find(name);
    if (it != _exchanges.end())
    {
        ILOG("ExchangeManager declareExchange:%s already exists!", name.c_str());
        return true;
    }
    Exchange::ptr exchange_ptr = std::make_shared<Exchange>(name, type, durable, auto_del, args);
    _exchanges.insert(std::make_pair(name, exchange_ptr));
    if (exchange_ptr->_durable == true)
        _mapper.insert(exchange_ptr);
    return true;
}
removeExchange 函数：
  • 删除一个指定名称的 Exchange 对象。如果该对象存在且是持久化的，则同时从数据库中删除该记录。
bool removeExchange(const std::string &name)
{
    std::unique_lock<std::mutex> lock(_mutex);
    auto it = _exchanges.find(name);
    if (it == _exchanges.end())
    {
        ILOG("ExchangeManager removeExchange not found!");
        return true;
    }
    _exchanges.erase(it);
    if (it->second->_durable)
        _mapper.remove(name);
    return true;
}
selectExchange 函数：
  • 获取指定名称的 Exchange 对象的智能指针，如果未找到则返回一个空的智能指针
Exchange::ptr selectExchange(const std::string &name)
{
    std::unique_lock<std::mutex> lock(_mutex);
    auto it = _exchanges.find(name);
    if (it == _exchanges.end())
        return Exchange::ptr();
    else
        return it->second;
}
exists 函数：
bool exists(const std::string &name)
{
    std::unique_lock<std::mutex> lock(_mutex);
    auto it = _exchanges.find(name);
    return it != _exchanges.end();
}
clear 函数：

清理内存中的 _exchanges 哈希表，并删除数据库中的 exchange_table 表。

void clear()
{
    std::unique_lock<std::mutex> lock(_mutex);
    _exchanges.clear();
    _mapper.dropTable();
}
size 函数：
size_t size()
{
    std::unique_lock<std::mutex> lock(_mutex);
    return _exchanges.size();
}

五.exchange.hpp所有代码

#pragma once

#include "../common_mq/helper.hpp"
#include "../common_mq/logger.hpp"
#include "../common_mq/msg.pb.h"
#include <string>
#include <unordered_map>
#include <mutex>
#include <memory>
#include <cassert>
#include <cstring>
#include <google/protobuf/map.h>
namespace mq
{
    struct Exchange
    {
        using ptr = std::shared_ptr<Exchange>;
        Exchange() {}
        Exchange(const std::string &name, msg::ExchangeType type,
                 bool durable,
                 bool auto_del,
                 const google::protobuf::Map<std::string, std::string> &args)
            : _name(name),
              _type(type),
              _durable(durable),
              _auto_del(auto_del),
              _args(args) {}

        // args存储键值对，在存储数据库的时候，会组织一个格式字符串进行存储 key=val & key=val....
        // 内部解析str_args字符串，将内容存储到成员中
        void setArgs(const std::string &str_args)
        {
            std::vector<std::string> args;
            size_t sz = StrHelper::split(str_args, "&", args);
            for (auto &kv : args)
            {
                size_t pos = kv.find("=");
                if (pos == std::string::npos)
                {
                    ELOG("Exchange args format error:%s", kv.c_str());
                    assert(0);
                }
                std::string key = kv.substr(0, pos);
                std::string val = kv.substr(pos + 1);
                _args[key] = val;
                // ILOG("Exchange args:%s", kv.c_str());
            }
        }
        std::string getArgs()
        {
            std::string ret;
            for (auto &kv : _args)
            {
                ret += kv.first + "=" + kv.second + "&";
            }

            return ret;
        }

        std::string _name;                                     // 交换机名称
        msg::ExchangeType _type;                               // 交换机类型
        bool _durable;                                         // 是否持久化
        bool _auto_del;                                        // 是否自动删除
        google::protobuf::Map<std::string, std::string> _args; // 其它参数
};

    using exchange_map = std::unordered_map<std::string, Exchange::ptr>;

    // 在数据库中 存储exchange信息
    class ExchangeMapper
    {
    public:
        ExchangeMapper(const std::string &dbname)
            : sql_helper(dbname)
        {
            // 数据库有path即可，open时自动创建文件
            std::string path = FileHelper::getParentDirName(dbname);
            FileHelper::createDir(path);
            if (!sql_helper.open())
            {
                ELOG("ExchangeMapper open error!");
                assert(0);
            }
            // std::cout<<"创建表！！！"<<std::endl;
            createTable();
        }
        ~ExchangeMapper() {}
        // 1.创建/删除Exchange表
        void createTable()
        {
            // std::cout<<"创建表！！！"<<std::endl;
            std::stringstream sql;
            sql << "create table if not exists exchange_table\
            ('name' varchar(64) primary key,\
            type int,durable int,auto_del int,args varchar(128));";
            if (!sql_helper.exec(sql.str(), nullptr, nullptr))
            {
                ELOG("ExchangeMapper create table error!");
                assert(0);
            }
        }
        void dropTable()
        {
            std::stringstream sql;
            sql << "drop table if exists exchange_table;";
            if (!sql_helper.exec(sql.str(), nullptr, nullptr))
            {
                ELOG("ExchangeMapper drop table error!");
                assert(0);
            }
        }
        // 2.插入/删除Exchange
        void insert(Exchange::ptr &exchange_ptr)
        {
            std::stringstream sql;
            sql << "insert into exchange_table values('" << exchange_ptr->_name << "'," << exchange_ptr->_type << "," << exchange_ptr->_durable << ","
                << exchange_ptr->_auto_del << ",'"
                << exchange_ptr->getArgs() << "');";
            if (!sql_helper.exec(sql.str(), nullptr, nullptr))
            {
                ELOG("ExchangeMapper insert error!");
                assert(0);
            }
        }
        void remove(const std::string &name)
        {
            std::stringstream sql;
            sql << "delete from exchange_table where name='" << name << "';";
            if (!sql_helper.exec(sql.str(), nullptr, nullptr))
            {
                ELOG("ExchangeMapper remove error!");
                assert(0);
            }
        }

        // 3.查询所有的Exchange，返回一个哈希表
        // 将数据库中的Exchange恢复到内存中，组织一个map并返回
        exchange_map recover()
        {
            exchange_map ret;
            std::stringstream sql;
            sql << "select * from exchange_table;";
            if (!sql_helper.exec(sql.str(), exchangeMapCb, &ret))
            {
                ELOG("ExchangeMapper recover error!");
                assert(0);
            }
            return ret;
        }

    private:
        static int exchangeMapCb(void *args, int rows, char **ret, char **fields)
        {
            Exchange::ptr exchange_ptr = std::make_shared<Exchange>();
            exchange_map *map_ptr = (exchange_map *)args;
            exchange_ptr->_name = ret[0];
            exchange_ptr->_type = (msg::ExchangeType)std::stoi(ret[1]);
            exchange_ptr->_durable = std::stoi(ret[2]);
            exchange_ptr->_auto_del = std::stoi(ret[3]);
            // 避免args为空时出错
            if (ret[4])
            {
                // ILOG("有其它参数，setArgs");
                // std::cout<<"ret[4]"<<ret[4]<<std::endl;
                exchange_ptr->setArgs(ret[4]);
            }
            else
            {
                ILOG("没有其它参数");
            }

            map_ptr->insert(std::make_pair(exchange_ptr->_name, exchange_ptr));
            return 0;
        }

    private:
        SqliteHelper sql_helper;
    };
    // 对外Exchange管理类
    class ExchangeManager
    {
    private:
        std::mutex _mutex;
        ExchangeMapper _mapper;
        exchange_map _exchanges;

    public:
        using ptr = std::shared_ptr<ExchangeManager>;
        ExchangeManager(const std::string &dbname)
            : _mapper(dbname)
        {
            _exchanges = _mapper.recover();
            // std::cout<<"recover后的个数"<<_exchanges.size()<<std::endl;
        }
        // 1.声明交换机，没有则创建
        bool declareExchange(const std::string &name, msg::ExchangeType type,
                             bool durable,
                             bool auto_del,
                             google::protobuf::Map<std::string, std::string> args)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // std::cout<<"声明之前----"<<std::endl;
            // std::cout<<_exchanges.size()<<std::endl;

            auto it = _exchanges.find(name);
            if (it != _exchanges.end()) // 找到了
            {
                std::cout << _exchanges[name]->_args.size() << std::endl;
                ILOG("ExchangeManager declareExchange:%s already exists!", name.c_str());
                return true;
            }
            Exchange::ptr exchange_ptr = std::make_shared<Exchange>(name, type, durable, auto_del, args);
            // std::cout<<"打印的是构建出的对象"<<std::endl;
            // const auto &tmp = exchange_ptr->_args;
            // for(const auto &tmp : exchange_ptr->_args)
            // {
            //     std::cout<<"key:"<<tmp.first<<" value:"<<tmp.second<<std::endl;
            // }
            _exchanges.insert(std::make_pair(name, exchange_ptr));
            // 若持久化则插入数据库
            if (exchange_ptr->_durable == true)
                _mapper.insert(exchange_ptr);
            return true;
        }
        // 2.删除交换机
        bool removeExchange(const std::string &name)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _exchanges.find(name);
            if (it == _exchanges.end()) // 没找到
            {
                ILOG("ExchangeManager removeExchange not found!");
                return true;
            }
            // 删除交换机
            _exchanges.erase(it);
            // 若持久化则在数据库中也remove
            if (it->second->_durable)
                _mapper.remove(name);
            return true;
        }
        // 3.获取指定交换机/判断是否存在
        Exchange::ptr selectExchange(const std::string &name)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _exchanges.find(name);
            if (it == _exchanges.end()) // 没找到返回空的ptr
                return Exchange::ptr();
            else
                return it->second;
        }
        bool exists(const std::string &name)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _exchanges.find(name);
            if (it == _exchanges.end()) // 没找到
                return false;
            else
                return true;
        }
        // 4.清理数据结构
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _exchanges.clear();
            _mapper.dropTable();
        }
        size_t size()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _exchanges.size();
        }
    };
};

标签：std,args,name,Exchange,hpp,交换机,ptr,exchange
From： https://blog.csdn.net/lzfnb666/article/details/141124098

相关文章

  • 思科三层交换机实现EIGIP路由协议6
    #路由协议实现##任务六三层交换机实现EIGIP路由协议6##1配置计算机的IP地址、子网掩码和网关#2配置Switch-A的名称及其接口IP地址Switch(config)#hostnameSwitch-ASwitch-A(config)#iproutingSwitch-A(config)#intg0/1Switch-A(config-if)#noswitchportSwitch-......
  • 什么是核心交换机？这样回答太到位了
    号主：老杨丨11年资深网络工程师，更多网工提升干货，请关注公众号：网络工程师俱乐部你们好，我的网工朋友。无论是企业内部通信还是对外服务，稳定高效的数据传输都是成功的关键。在这个背景下，核心交换机作为网络基础设施的中枢神经，扮演着至关重要的角色，相信大家在工作中也经常打交道......
  • HCL学习——交换机端口安全技术
    本篇记录学习HCL的笔记。【2023年】H3CNE认证网络工程师完整培训视频教程_上https://www.bilibili.com/video/BV1Dg411i7yM?p=22&spm_id_from=pageDriver&vd_source=ecbebcd4db8fad7f74c518d13e78b165 802.1x技术 如果随便拿台pc插到交换机上岂不是就能接入本公司网络呢，80......
  • 3.3交换机的三种交换方式
    一、存储转发（Store-and-Forward）存储转发是最常见的交换方式，交换机会复制整个数据帧到缓冲区，然后计算CRC（循环冗余检查，CyclicRedundancyCheck）。工作原理接收数据帧：交换机接收到整个数据帧。错误检测：交换机计算帧校验序列（FCS），检查数据帧是否有错误。转发决定：如果数据帧没有错......
  • 3.2交换机交换的基本原理
    一、 交换机的基本概念A、数据帧1.数据帧的组成部分一个典型的数据帧由多个字段组成，每个字段都有特定的用途。1)帧头（Header）帧头位于数据帧的开头，包含了必要的控制信息，用于指导帧的传输和处理。帧头通常包括以下几个字段：目的地址（DestinationAddress）：指向接收数据帧的......
  • 在 Windows 上使用 LCX（Local Channel eXchange）来进行本地和远程转发，此大纲旨在提供顶
    LCX（LocalChanneleXchange）通常指的是一种用于网络协议中的本地和远程转发技术。如果你在谈论的是与LCX相关的网络配置，它可能涉及不同的上下文，例如在通信协议或网络交换中。本地和远程转发的基本概念：本地转发(LocalForwarding):本地转发将本地计算机上的一个端口转发到......
  • IgniteFAQ-13-GridDhtPartitionsExchangeFuture : Failed to reinitialize local part
    报错2024-08-0815:29:02.532ERROR39656---[ange-worker-#49].c.d.d.p.GridDhtPartitionsExchangeFuture:Failedtoreinitializelocalpartitions(rebalancingwillbestopped):GridDhtPartitionExchangeId[topVer=AffinityTopologyVersion[topVer=1,minorTopVe......
  • 思科三层交换机的配置3
    #交换机的VLAN配置##任务三三层交换机的配置#三层交换机即内置了路由功能的交换机，在转发数据帧的同时，还可以在不同网段之间路由数据包。在交换式局域网中，三层交换机可以配置多个虚拟vlan端口（SVI）作为vlan内PC设备的网关，同时转发数据包，实现不同vlan之间的通信。#1配置计算......
  • 思科交换机的VLAN划分1
    #交换机的VLAN配置##任务一交换机的VLAN划分#vlan中的网络用户是通过lan交换机来通信的。一个vlan中的成员看不到另一个vlan中的成员。因此vlan划分能有效地控制网络广播风暴，提高网络的安全可靠性，是有效的网络监控、数据流量控制的手段，还能实现不同地理位置的部门间的局域网......
  • 6.交换机及其配置
    6.1交换机的基本概念#路由器工作在：第三层网络层，跨局域网传输交换机工作在：第二层数据链路层，在局域网内传输在PC2ping一下PC3的IP交换机自动就会记录他们的MAC地址和端口号6.2交换表的内容小型交换机大型交换机6.3交换机的交换模式直通-快速转发：知道地......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99