文章目录
- 一.编写思路
- 二.代码实践
一.编写思路
- 定义交换机类型
- 直接交换
- 广播交换
- 主题交换
- 定义交换机
- 名字
- 类型
- 是否持久化
- 定义交换机持久化类(持久化到 sqlite3)
- 构造函数(只能成功，不能失败)
如果数据库(文件)不存在则创建;打开数据库; 打开 exchange_table 数据库表- 插入交换机
- 移除交换机
- 将数据库中的交换机恢复到内存中
传入一个哈希表，key 为名字，value 为交换机的智能指针，填充该哈希表- 定义交换机管理类(包含内存管理和持久化管理)
- 构造函数：从数据库中恢复交换机
- 声明交换机
- 移除交换机
- 获取交换机
二.代码实践
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;}} ;
}