当前位置： 首页 > article >正文

【RabbitMQ 项目】服务端数据管理模块之交换机管理

article 2024/11/12 13:22:34

文章目录

  • 一.编写思路
  • 二.代码实践

一.编写思路

  1. 定义交换机类型
    1. 直接交换
    2. 广播交换
    3. 主题交换
  2. 定义交换机
    1. 名字
    2. 类型
    3. 是否持久化
  3. 定义交换机持久化类(持久化到 sqlite3)
    1. 构造函数(只能成功，不能失败)
      如果数据库(文件)不存在则创建;打开数据库; 打开 exchange_table 数据库表
    2. 插入交换机
    3. 移除交换机
    4. 将数据库中的交换机恢复到内存中
      传入一个哈希表，key 为名字，value 为交换机的智能指针，填充该哈希表
  4. 定义交换机管理类(包含内存管理和持久化管理)
    1. 构造函数：从数据库中恢复交换机
    2. 声明交换机
    3. 移除交换机
    4. 获取交换机

二.代码实践

Exchange.hpp:

#pragma once
#include "../common/Util.hpp"
#include <string>
#include <memory>
#include <unordered_map>
#include <functional>

namespace ns_data
{
    // 声明
    class Exchange;

    using ExchangePtr = std::shared_ptr<Exchange>;

    /***********
     * 定义交换机类型
     * **************/
    enum class ExchangeType
    {
        DIRECT = 0, // 直接交换
        FANOUT = 1, // 广播交换
        TOPIC = 2   // 主题交换
    };

    /**********
     * 定义交换机
     * ***************/
    struct Exchange
    {
        std::string _name;
        ExchangeType _type;
        bool _isDurable;

        Exchange()
        {
        }

        Exchange(const std::string &name, ExchangeType type, bool isDurable)
            : _name(name),
              _type(type),
              _isDurable(isDurable)
        {
        }
    };

    /**************
     * 定义交换机持久化类
     * ****************/
    class ExchangeMapper
    {
    private:
        ns_util::Sqlite3Util _sqlite3;

    public:
        ExchangeMapper(const std::string &dbName)
            : _sqlite3(dbName)
        {
            // 如果数据库不存在就创建
            if (!ns_util::FileUtil::createFile(dbName))
            {
                LOG(FATAL) << "creat database" << dbName << endl;
                exit(1);
            }
            if (!_sqlite3.open())
            {
                LOG(FATAL) << "open database " << dbName << " fail" << endl;
                exit(1);
            }
            createTable();
        }

        /************
         * 功能:插入交换机
         * *************/
        bool insertExchange(ExchangePtr exchangePtr)
        {
            char insertSql[1024];
            sprintf(insertSql, "insert into exchange_table values('%s', '%d', '%d');", (exchangePtr->_name).c_str(),
                    static_cast<int>(exchangePtr->_type), exchangePtr->_isDurable);
            if (!_sqlite3.exec(insertSql, nullptr, nullptr))
            {
                LOG(WARNING) << "insert Exchange fail, Exchange: " << exchangePtr->_name << endl;
                return false;
            }
            return true;
        }
        /**********
         * 功能：移除交换机
         * *************/
        void removeExchange(const std::string &exchangeName)
        {
            char removeSql[1024];
            sprintf(removeSql, "delete from exchange_table where name='%s'", exchangeName.c_str());
            if (!_sqlite3.exec(removeSql, nullptr, nullptr))
            {
                LOG(WARNING) << "remvoe Exchange fail, Exchange: " << exchangeName << endl;
            }
        }

        /***********
         * 功能：把数据库中的所有交换机恢复到内存中
         * *************/
        void recoverExchange(std::unordered_map<std::string, ExchangePtr> *mapPtr)
        {
            char selectSql[1024] ;
            sprintf(selectSql, "select * from exchange_table;");
            if (!_sqlite3.exec(selectSql, selectCallback, mapPtr))
            {
                LOG(FATAL) << "recover Exchange fail" << endl;
                exit(1);
            }
        }

        /*************
         * 功能:删除交换机数据表(仅用于调试)
         * ***********/
        void removeTable()
        {
            const std::string removeSql = "drop table if exists exchange_table";
            if (!_sqlite3.exec(removeSql, nullptr, nullptr))
            {
                LOG(FATAL) << "remvoe exchange_table fail" <<  endl;
                exit(1);
            }
        }

    private:
        /*************
         * 功能：创建存储Exchange的数据表
         * *********/
        void createTable()
        {
            std::string createTableSql = "create table if not exists exchange_table(\
                name varchar(32) primary key,\
                type int,\
                durable int);";
            if (!_sqlite3.exec(createTableSql, nullptr, nullptr))
            {
                LOG(FATAL) << "创建交换机数据库表失败" << endl;
                exit(1);
            }
        }

        static int selectCallback(void* arg, int colNum, char** line, char** fields)
        {
            auto mapPtr = static_cast<std::unordered_map<std::string, ExchangePtr>*>(arg);

            std::string name = line[0];
            ExchangeType type = static_cast<ExchangeType>(std::stoi(line[1]));
            bool isDurable = line[2];

            mapPtr->insert({name, std::make_shared<Exchange>(name, type, isDurable)});
            return 0;
        }
    };

    /***********
     * 定义交换机管理类：包含内存管理和持久化管理两方面
     * *************/
    class ExchangeManager
    {
    private:
        ExchangeMapper _mapper;
        std::unordered_map<std::string, ExchangePtr> _exchanges;
    public:
        ExchangeManager(const std::string& dbName)
            :_mapper(dbName)
        {
            _mapper.recoverExchange(&_exchanges);
        }

        /**************
         * 功能：声明交换机：无则创建，有则什么都不管
         * *************/
        bool declareExechange(const std::string &name, ExchangeType type, bool isDurable)
        {
            if (_exchanges.count(name))
            {
                return true;
            }

            auto exchangePtr = std::make_shared<Exchange>(name, type, isDurable);
            _exchanges[name] = exchangePtr;

            if (isDurable)
            {
                return _mapper.insertExchange(exchangePtr);
            }

            return true;
        }

        /**********
         * 移除交换机：包括内存和磁盘两个地方
         * ************/
        void removeExchange(const std::string &name)
        {
            auto it = _exchanges.find(name);
            if (it == _exchanges.end())
            {
                return;
            }

            if (it->second->_isDurable)
            {
                _mapper.removeExchange(name);
            }
            _exchanges.erase(name);
        }

        /*************
         * 获取指定交换机
         * ***********/
        ExchangePtr getExchange(const std::string& name)
        {
            if (_exchanges.count(name) == 0)
            {
                return nullptr;
            }
            return _exchanges[name];
        }

        /***********
         * 清理所有交换机(仅调试)
         * ************/
        void clearExchanges()
        {
            _exchanges.clear();
            _mapper.removeTable();
        }
    };
}

Util.hpp:

#pragma once
#include "Log.hpp"
#include <string>
#include <sqlite3.h>
#include <iostream>
#include <sys/stat.h>
#include <sys/types.h>
#include <fcntl.h>
#include <unistd.h>
using namespace ns_log;
namespace ns_util
{
    class Sqlite3Util
    {
    private:
        std::string _dbfile;
        sqlite3 *_handler;
        bool _isOpen;

    public:
        Sqlite3Util(const std::string &dbfile)
            : _dbfile(dbfile),
              _handler(nullptr),
              _isOpen(false)
        {
            open();
        }

        ~Sqlite3Util()
        {
            close();
        }

        bool open(int safeLevel = SQLITE_OPEN_FULLMUTEX)
        {
            if (_isOpen)
            {
                return true;
            }

            // 可读可写，不存在就创建，默认串行化访问
            int ret = sqlite3_open_v2(_dbfile.c_str(), &_handler, SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | safeLevel,
                                      nullptr);
            if (ret != SQLITE_OK)
            {
                LOG(WARNING) << sqlite3_errmsg(_handler) << endl;
                return false;
            }

            _isOpen = true;
            return true;
        }
        // int sqlite3_exec(sqlite3*, char *sql, int (*callback) (void* arg,int colNum ,char** lines,char** fields), void* arg, char **err)
        bool exec(const std::string &sql, int (*cb)(void *, int, char **, char **), void *arg)
        {
            if (sqlite3_exec(_handler, sql.c_str(), cb, arg, nullptr) != SQLITE_OK)
            {
                LOG(WARNING) << "execute fail: error: " << sqlite3_errmsg(_handler) << endl;
                return false;
            }
            return true;
        }
        void close()
        {
            if (_handler)
            {
                sqlite3_close_v2(_handler);
            }
        }
    };

    class FileUtil
    {
    public:
        static void getParentDirectory(const std::string &pathName, std::string *dirPtr)
        {
            // 从后往前找，找到第一个“/”
            auto pos = pathName.rfind('/');
            if (pos == std::string::npos)
            {
                *dirPtr = "./";
                return;
            }
            *dirPtr = pathName.substr(0, pos);
        }

        static bool createDirectory(const std::string &dirName)
        {
            // 从第一个父目录开始，逐层创建
            size_t prev = 0;
            while (true)
            {
                auto pos = dirName.find('/', prev);
                if (pos == std::string::npos)
                {
                    break;
                }

                std::string dir = dirName.substr(0, pos);
                int ret = mkdir(dir.c_str(), 0775);
                if (ret != 0 && errno != EEXIST)
                {
                    LOG(WARNING) << "创建目录" << dir << "失败, error: " << strerror(errno) << endl;
                    return false;
                }

                prev = pos + 1;
            }

            int ret = mkdir(dirName.c_str(), 0775);
            if (ret != 0 && errno != EEXIST)
            {
                LOG(WARNING) << "创建目录" << dirName << "失败, error: " << strerror(errno) << endl;
                return false;
            }

            return true;
        }

        static bool createFile(const std::string pathName)
        {
            //先创建它的父目录
            std::string parentDir;
            getParentDirectory(pathName, &parentDir);
            if (!createDirectory(parentDir))
            {
                LOG(WARNING) << "创建文件失败，因为创建父目录失败" << endl;
                return false;
            }
            //再创建文件
            int fd = open(pathName.c_str(), O_CREAT, 0775);
            if (fd == -1)
            {
                LOG(WARNING) << "创建文件失败, error: " << strerror(errno) << endl;
                return false;
            }
            close(fd);
            return true;
        }

        static bool removeFileOrDir(const std::string& name)
        {
            
            if (remove(name.c_str()) != 0)
            {
                LOG(WARNING) << "remove " << name << " fail, error: " << strerror(errno) << endl;
                return false;
            }
            return true;
        }
    } ;
}

http://www.kler.cn/a/312244.html

相关文章：

  • mysql 更改 字段长度
  • 深入了解Git、GitHub、GitLab及其应用技巧
  • Redis安装(Windows环境)
  • 数据库基础(14) . MySQL存储过程
  • Docker使用docker-compose一键部署nacos、Mysql、redis
  • C语言 | Leetcode C语言题解之第559题N叉树的最大深度
  • prometheus监控linux虚拟机
  • 操作系统迁移(CentOs -＞ Ubuntu)
  • Wacom 和 Splashtop 携手共赴 IBC 2024 展会，宣布向欧洲市场隆重推出 Wacom Bridge
  • XSS漏洞挖掘利用总结（附个人常用payload）
  • MyBatis 分批次执行（新增，修改，删除）
  • ROS激光雷达介绍
  • WPF中图片的宫格显示
  • TPDO触发条件如何满足？
  • AI学习指南深度学习篇-Adam的Python实践
  • 如何配置和使用自己的私有 Docker Registry
  • python的6种常用数据结构
  • 3.大语言模型LLM的公开资源（API、语料库、算法库）
  • Python中的树与图：构建复杂数据结构的艺术
  • 图论三元环（并查集的高级应用）
  • 天润融通创新功能，将无效会话转化为企业新商机
  • 青柠视频云——视频丢包（卡顿、花屏、绿屏）排查
  • Python 集合的魔法：解锁高效数据处理的秘密
  • 工厂模式（一）：简单工厂模式
  • Web后端服务平台解析漏洞与修复、文件包含漏洞详解
  • 【Git原理与使用】多人协作与开发模型（2）