一、模块设计分析

我们以交换机为例来分析，队列跟绑定信息和它的设计思路是一样的

1.模块划分

这个对外管理其实就是对交换机、队列、绑定信息的：
增、删、查（没有改，不提供改操作，你可以删了重新新增，但是不能改）
因为多线程环境下改交换机，队列，绑定信息破坏系统的原子性和一致性，导致数据不一致的状态

而且RabbitMQ的设计哲学就是保持简单和可预测。通过提供明确和有限的操作接口，RabbitMQ能够确保其行为在不同场景下都是可预测的，从而降低了使用和维护的难度

可以看作是一种代码健壮性，可维护性和YAGNI原则（You Aren’t Gonna Need It 你不会需要它）的贯彻

2.功能需求

1. Exchange
   - 【1】交换机名称
   - 【2】交换机类型
   - 【3】持久化标志
   - 【4】 自动删除标志
   - 【5】其他参数【<key,value>的一个google::protobuf::Map】

2. ExchangeMapper

   1. 对外接口
      • 【1】创建/删除交换机表
      • 【2】新增/删除数据
      • 【3】查询所有交换机数据（用于恢复持久化的交换机数据）
   2. 管理对象
      • 【1】SqliteHelper的句柄

3. ExchangeManager

   1. 对外接口
      • 【1】声明/删除交换机
      • 【2】获取指定交换机
      • 【3】销毁所有交换机
   2. 管理对象
      • 【1】保证线程安全的互斥锁
      • 【2】<交换机名称，交换机对象智能指针>的一个哈希表
      • 【3】ExchangeMapper句柄

二、交换机模块的实现

1.交换机结构体的实现

之前说过，交换机类型需要持久化，因为需要通过网络IO，因此这个其他参数的Map也需要持久化
所以需要用protobuf当中的Map

namespace ns_Exchange
{const std::string inner_sep = "=";const std::string outer_sep = "&";
}struct Exchange
{using ptr = std::shared_ptr<Exchange>;Exchange() = default;Exchange(const std::string &ename, ExchangeType etype, bool edurable, bool eauto_delete,const google::protobuf::Map<std::string, std::string> &eargs): name(ename), type(etype), durable(edurable), auto_delete(eauto_delete), args(eargs) {}std::string name;ExchangeType type;bool durable;bool auto_delete;google::protobuf::Map<std::string, std::string> args;// 因为交换机要持久化,而sqlite里面没有kv结构类型,所以需要进行序列化和反序列化,存字符串// 定义格式 key=val&key=val// 因此需要我们提供args的序列化和反序列化操作std::string getArgs(){std::string ret;for (auto &kv : args){ret += kv.first + ns_Exchange::inner_sep + kv.second;ret += ns_Exchange::outer_sep;}int sz = ns_Exchange::outer_sep.size();while (!ret.empty() && sz--){ret.pop_back();}return ret;}void setArgs(const std::string &str){// 使用StringHelper的字符串切割函数将key=val&key=val切分为key=val key=val// 然后按照=进行分割即可std::vector<std::string> vec;StringHelper::split(str, ns_Exchange::outer_sep, &vec);for (auto &elem : vec){size_t pos = elem.find(ns_Exchange::inner_sep);args.insert({elem.substr(0, pos), elem.substr(pos + 1)});}}
};

就是一个结构体定义+俩构造函数+Map的序列和反序列化（自定义小协议解决粘包）

2.交换机持久化管理模块的实现

using ExchangeMap = std::unordered_map<std::string, Exchange::ptr>;const int arg_len = 512;class ExchangeMapper
{
public:// 构造当中传入数据库名称ExchangeMapper(const std::string &dbfile): _helper(dbfile){if (!_helper.open()){default_fatal("交换机持久化管理模块初始化失败, 因为数据库打开失败");abort();}if (!createTable()){default_fatal("交换机持久化管理模块初始化失败, 因为表的创建失败");abort();}}// 因为SqliteHelper在析构的时候会自动调用其内部的close，所以不用管~ExchangeMapper() = default;// 创建/删除表bool createTable(){// 为了代码健壮性std::ostringstream create_sql;create_sql << "create table if not exists exchange_table(\name varchar(32) primary key,\type int,\durable int,\auto_delete int,\args varchar(";create_sql << arg_len << "));";if (!_helper.exec(create_sql.str(), nullptr, nullptr)){default_fatal("创建交换机数据表失败");return false;}return true;}bool dropTable(){static std::string drop_sql = "drop table if exists exchange_table;";if (!_helper.exec(drop_sql, nullptr, nullptr)){default_fatal("删除交换机数据表失败");return false;}return true;}// 插入/删除数据bool insert(const Exchange::ptr &ep){if (ep.get() == nullptr){default_fatal("无法将该数据插入交换机表当中，因为该数据为空指针");return false;}std::ostringstream insert_sql;insert_sql << "insert into exchange_table values(";insert_sql << "'" << ep->name << "',";insert_sql << ep->type << ",";insert_sql << ep->durable << ",";insert_sql << ep->auto_delete << ",";insert_sql << "'" << ep->getArgs() << "');";if (!_helper.exec(insert_sql.str(), nullptr, nullptr)){default_error("插入交换机数据失败，exchange_name: %s",ep->name.c_str());return false;}return true;}bool erase(const std::string &ename){std::ostringstream erase_sql;erase_sql << "delete from exchange_table where name=" << "'" << ename << "';";if (!_helper.exec(erase_sql.str(), nullptr, nullptr)){default_error("删除交换机数据失败，exchange_name: %s",ename.c_str());return false;}return true;}// 查询所有交换机数据ExchangeMap recovery(){static std::string select_sql = "select * from exchange_table;";ExchangeMap emap;if (!_helper.exec(select_sql, &ExchangeMapper::SelectCallback, &emap)){default_error("查询交换机所有数据失败");return ExchangeMap();}return emap;}private:static int SelectCallback(void *arg, int column, char **row, char **field){ExchangeMap *eptr = static_cast<ExchangeMap *>(arg);Exchange::ptr ep = std::make_shared<Exchange>();ep->name = row[0];ep->type = static_cast<ExchangeType>(std::stoi(row[1])); // 字符串 -> 整形 -> 枚举ep->durable = static_cast<bool>(std::stoi(row[2]));ep->auto_delete = static_cast<bool>(std::stoi(row[3]));ep->setArgs(row[4]);eptr->insert(std::make_pair(ep->name, ep));return 0;}SqliteHelper _helper;
};

其实就是写SQL语句，执行，打印错误日志。
那个selectCallback跟我们之前写的本质是同一个思想，其实就是传入一个大型容器
然后在内部创建填好一个小容器，然后把小容器放到大容器当中即可
甚至上面那个还更简单

int SelectCallback(void *arg, int column, char **rows, char **fields)
{vector<vector<string>> *total_elem = static_cast<vector<vector<string>> *>(arg);vector<string> one_elem;if (total_elem->empty()){for (int i = 0; i < column; i++){one_elem.push_back(fields[i]);}total_elem->push_back(move(one_elem));}for (int i = 0; i < column; i++){one_elem.push_back(rows[i]);}total_elem->push_back(move(one_elem));return 0;
}

3.交换机对外管理模块实现

对外管理其实就是增删查，也没啥难写的

class ExchangeManager
{
public:using ptr = std::shared_ptr<ExchangeManager>;ExchangeManager(const std::string &dbfile): _mapper(dbfile){// 恢复所有交换机_emap = _mapper.recovery();}bool declareExchange(const std::string &ename, ExchangeType etype, bool edurable, bool eauto_delete,const google::protobuf::Map<std::string, std::string> &eargs){// 0. 加锁std::unique_lock<std::mutex> ulock(_mutex);if (_emap.count(ename)) // 有该交换机return true;// 1. 构建Exchange::ptr对象（这里也需要锁，因为判断+创建才是个完整的声明交换机的操作，要保证他的原子性）Exchange::ptr ep = std::make_shared<Exchange>(ename, etype, edurable, eauto_delete, eargs);// 2. 看是否需要持久化if (edurable){if (!_mapper.insert(ep)){default_error("声明交换机持久化数据失败, 交换机名称：%s",ename.c_str());return false;}}// 3. 放到_emap当中_emap.insert(std::make_pair(ename, ep));return true;}bool eraseExchange(const std::string &ename){// 0. 加锁std::unique_lock<std::mutex> ulock(_mutex);// 1. 在emap中查找该交换机auto iter = _emap.find(ename);if (iter == _emap.end())return true;// 2. 看该交换机是否持久化了Exchange::ptr ep = iter->second;if (ep->durable){// 3. 持久化删除if (!_mapper.erase(ename)){default_error("删除交换机持久化数据失败, 交换机名称: %s",ename.c_str());return false;}}// 4. 内存删除_emap.erase(iter);return true;}ExchangeMap getAllExchanges(){std::unique_lock<std::mutex> ulock(_mutex);return _emap;}Exchange::ptr getExchange(const std::string &ename){std::unique_lock<std::mutex> ulock(_mutex);auto iter = _emap.find(ename);if (iter == _emap.end()){return Exchange::ptr();}return iter->second;}// ExchangeManager的析构函数不许调用clear// 因为持久化数据的清理只能由用户决定void clear(){std::unique_lock<std::mutex> ulock(_mutex);_mapper.dropTable();_emap.clear();}bool exists(const std::string &ename){std::unique_lock<std::mutex> ulock(_mutex);return _emap.count(ename) > 0;}private:ExchangeMapper _mapper;std::mutex _mutex;ExchangeMap _emap;
};

声明、删除交换机时的查找不能复用exists函数

其实上面就是增删查而已
大家都写过这么多代码了，这种代码没啥难的，仔细看看就搞定了
他们仨的核心其实都是一样的，所以我们放在一起来介绍

为何持久化管理模块无需事务

因为它们都只是涉及到一条简单的SQL语句而已,并不是多条组成
也不是复杂的多表连接等等,操作本身就是原子的,要么成功,要么失败
所以无需事务来保证ACID 原子性,一致性,隔离性,持久性
天然就有这四种特性

4.测试代码

用GTest的全局测试框架来搞

#include <gtest/gtest.h>
#include "../mqserver/exchange.hpp"
// 因为全局测试套件更方便,所以我们用它
using namespace ns_mq;ExchangeManager::ptr emp;class ExchangeTest : public testing::Environment
{
public:virtual void SetUp(){emp = std::make_shared<ExchangeManager>("test.db");}virtual void TearDown(){emp->clear();}
};TEST(exchange_test, recovery_test)
{ASSERT_EQ(emp->exists("exchange1"), true);ASSERT_EQ(emp->exists("exchange2"), true);ASSERT_EQ(emp->exists("exchange3"), false);ASSERT_EQ(emp->exists("exchange4"), false);
}TEST(exchange_test, insert_test)
{google::protobuf::Map<std::string, std::string> emap;emap.insert({"k1", "v1"});emap.insert({"k2", "v2"});emp->declareExchange("exchange1", FANOUT, true, false, emap);emp->declareExchange("exchange2", DIRECT, true, false, emap);emp->declareExchange("exchange3", DIRECT, true, false, emap);emp->declareExchange("exchange4", DIRECT, false, false, emap);ASSERT_EQ(emp->exists("exchange1"), true);ASSERT_EQ(emp->exists("exchange2"), true);ASSERT_EQ(emp->exists("exchange3"), true);ASSERT_EQ(emp->exists("exchange4"), true);
}TEST(exchange_test, select_test)
{Exchange::ptr eptr = emp->getExchange("exchange1");ASSERT_EQ(eptr->type, FANOUT);ASSERT_EQ(eptr->durable, true);ASSERT_EQ(eptr->auto_delete, false);
}TEST(exchange_test, erase_test)
{bool ret = emp->eraseExchange("exchange3");ASSERT_EQ(ret, true);ASSERT_EQ(emp->exists("exchange1"), true);ASSERT_EQ(emp->exists("exchange2"), true);ASSERT_EQ(emp->exists("exchange3"), false);ASSERT_EQ(emp->exists("exchange4"), true);
}int main(int argc, char *argv[])
{testing::AddGlobalTestEnvironment(new ExchangeTest);testing::InitGoogleTest(&argc, argv);return RUN_ALL_TESTS();
}

这代码也不难，毕竟我们之前都用过GTest了，大家按自己的方式来测试就行了

三、队列模块的实现

1.功能需求

1. MsgQueue
   - 【1】 队列名称
   - 【2】 持久化标志
   - 【3】 是否独占标志（true：只有创建该队列的消费者才能订阅该队列）
   - 【4】 自动删除标志
   - 【5】 其他参数【<key,value>的一个google::protobuf::Map】

2. MsgQueueMapper
   1. 对外接口
   - 【1】 创建/删除表
   - 【2】 新增/删除数据
   - 【3】 查询所有队列数据
   2. 管理对象
   - 【1】 SqliteHelper句柄

3. MsgQueueManager
   1. 对外接口
   - 【1】 声明/删除队列
   - 【2】 查询指定队列
   - 【3】 销毁所有队列
   2. 管理对象
   - 【1】 互斥锁
   - 【2】 <队列名称，队列数据智能指针>的一个哈希表
   - 【3】MsgQueueMapper句柄

2.队列结构体的实现

namespace ns_MsgQueue
{const std::string inner_sep = "=";const std::string outer_sep = "&";
}
struct MsgQueue
{using ptr = std::shared_ptr<MsgQueue>;MsgQueue() = default;MsgQueue(const std::string &qname, bool qdurable, bool qexclusive, bool qauto_delete,const google::protobuf::Map<std::string, std::string> &qargs): name(qname), durable(qdurable), exclusive(qexclusive), auto_delete(qauto_delete), args(qargs) {}std::string name;bool durable;bool exclusive;bool auto_delete;google::protobuf::Map<std::string, std::string> args;std::string getArgs(){std::string ret;for (auto &kv : args){ret += kv.first + ns_MsgQueue::inner_sep + kv.second;ret += ns_MsgQueue::outer_sep;}int sz = ns_MsgQueue::outer_sep.size();while (!ret.empty() && sz--){ret.pop_back();}return ret;}void setArgs(const std::string &str){std::vector<std::string> vec;StringHelper::split(str, ns_MsgQueue::outer_sep, &vec);for (auto &elem : vec){size_t pos = elem.find(ns_MsgQueue::inner_sep);args.insert({elem.substr(0, pos - 1), elem.substr(pos + 1)});}}
};

跟交换机雷同，也就是成员有一点不一样而已

3.队列持久化管理模块的实现

using ExchangeMap = std::unordered_map<std::string, Exchange::ptr>;const int earg_len = 512;class ExchangeMapper
{
public:// 构造当中传入数据库名称ExchangeMapper(const std::string &dbfile): _helper(dbfile){if (!_helper.open()){default_fatal("交换机持久化管理模块初始化失败, 因为数据库打开失败");abort();}if (!createTable()){default_fatal("交换机持久化管理模块初始化失败, 因为表的创建失败");abort();}}// 因为SqliteHelper在析构的时候会自动调用其内部的close，所以不用管~ExchangeMapper() = default;// 创建/删除表bool createTable(){// 为了代码健壮性std::ostringstream create_sql;create_sql << "create table if not exists exchange_table(\name varchar(32) primary key,\type int,\durable int,\auto_delete int,\args varchar(";create_sql << earg_len << "));";if (!_helper.exec(create_sql.str(), nullptr, nullptr)){default_fatal("创建交换机数据表失败");return false;}return true;}bool dropTable(){static std::string drop_sql = "drop table if exists exchange_table;";if (!_helper.exec(drop_sql, nullptr, nullptr)){default_fatal("删除交换机数据表失败");return false;}return true;}// 插入/删除数据bool insert(const Exchange::ptr &ep){if (ep.get() == nullptr){default_fatal("无法将该数据插入交换机表当中，因为该数据为空指针");return false;}std::ostringstream insert_sql;insert_sql << "insert into exchange_table values(";insert_sql << "'" << ep->name << "',";insert_sql << ep->type << ",";insert_sql << ep->durable << ",";insert_sql << ep->auto_delete << ",";insert_sql << "'" << ep->getArgs() << "');";if (!_helper.exec(insert_sql.str(), nullptr, nullptr)){default_fatal("插入交换机数据失败，exchange_name: %s",ep->name.c_str());return false;}return true;}bool erase(const std::string &ename){std::ostringstream erase_sql;erase_sql << "delete from exchange_table where name=" << "'" << ename << "';";if (!_helper.exec(erase_sql.str(), nullptr, nullptr)){default_error("删除交换机数据失败，exchange_name: %s",ename.c_str());return false;}return true;}// 查询所有交换机数据ExchangeMap recovery(){static std::string select_sql = "select * from exchange_table;";ExchangeMap emap;if (!_helper.exec(select_sql, &ExchangeMapper::SelectCallback, &emap)){default_error("查询交换机所有数据失败");return ExchangeMap();}return emap;}private:static int SelectCallback(void *arg, int column, char **row, char **field){ExchangeMap *eptr = static_cast<ExchangeMap *>(arg);Exchange::ptr ep = std::make_shared<Exchange>();ep->name = row[0];ep->type = static_cast<ExchangeType>(std::stoi(row[1])); // 字符串 -> 整形 -> 枚举ep->durable = static_cast<bool>(std::stoi(row[2]));ep->auto_delete = static_cast<bool>(std::stoi(row[3]));ep->setArgs(row[4]);eptr->insert(std::make_pair(ep->name, ep));return 0;}SqliteHelper _helper;
};

跟交换机雷同

4.队列对外管理模块的实现

class MsgQueueManager
{
public:using ptr = std::shared_ptr<MsgQueueManager>;MsgQueueManager(const std::string &dbfile): _mapper(dbfile){_mqp = _mapper.recovery();}bool declareMsgQueue(const std::string &qname, bool qdurable, bool qexclusive, bool qauto_delete,const google::protobuf::Map<std::string, std::string> &qargs){std::unique_lock<std::mutex> ulock(_mutex);if (_mqp.count(qname))return true;MsgQueue::ptr mp = std::make_shared<MsgQueue>(qname, qdurable, qexclusive, qauto_delete, qargs);if (qdurable){if (!_mapper.insert(mp)){default_error("声明队列持久化数据失败, 队列名: %s",qname.c_str());return false;}}_mqp.insert(std::make_pair(qname, mp));return true;}bool eraseMsgQueue(const std::string &qname){std::unique_lock<std::mutex> ulock(_mutex);auto iter = _mqp.find(qname);if (iter == _mqp.end())return true;MsgQueue::ptr mp = iter->second;if (mp->durable){if (!_mapper.erase(mp->name)){default_error("删除队列持久化数据失败, 队列名: %s",mp->name.c_str());return false;}}_mqp.erase(iter);return true;}MsgQueueMap getAllMsgQueue(){std::unique_lock<std::mutex> ulock(_mutex);return _mqp;}MsgQueue::ptr getMsgQueue(const std::string &qname){std::unique_lock<std::mutex> ulock(_mutex);auto iter = _mqp.find(qname);if (iter == _mqp.end())return MsgQueue::ptr();return iter->second;}void clear(){std::unique_lock<std::mutex> ulock(_mutex);_mapper.dropTable();_mqp.clear();}bool exists(const std::string &qname){std::unique_lock<std::mutex> ulock(_mutex);return _mqp.count(qname) > 0;}private:std::mutex _mutex;MsgQueueMap _mqp;MsgQueueMapper _mapper;
};

没啥好说的,跟交换机一样的

5.测试代码

#include "../mqserver/msgqueue.hpp"
#include <gtest/gtest.h>using namespace ns_mq;MsgQueueManager::ptr mqmp;class MsgQueueTest : public testing::Environment
{
public:virtual void SetUp(){mqmp = std::make_shared<MsgQueueManager>("test.db");}virtual void TearDown(){mqmp->clear();}
};TEST(msgqueue_test, recovery_test)
{ASSERT_EQ(mqmp->exists("queue1"), true);ASSERT_EQ(mqmp->exists("queue2"), true);ASSERT_EQ(mqmp->exists("queue3"), false);ASSERT_EQ(mqmp->exists("queue4"), false);
}TEST(msgqueue_test, insert_test)
{google::protobuf::Map<std::string, std::string> mymap;mymap.insert({"k1", "v1"});mymap.insert({"k2", "v2"});mqmp->declareMsgQueue("queue1", true, false, false, mymap);mqmp->declareMsgQueue("queue2", true, false, false, mymap);mqmp->declareMsgQueue("queue3", true, false, false, mymap);mqmp->declareMsgQueue("queue4", false, true, true, mymap);ASSERT_EQ(mqmp->exists("queue1"), true);ASSERT_EQ(mqmp->exists("queue2"), true);ASSERT_EQ(mqmp->exists("queue3"), true);ASSERT_EQ(mqmp->exists("queue4"), true);
}TEST(msgqueue_test, select_test)
{MsgQueue::ptr mqp = mqmp->getMsgQueue("queue4");ASSERT_EQ(mqp->durable, false);ASSERT_EQ(mqp->exclusive, true);ASSERT_EQ(mqp->auto_delete, true);
}TEST(msgqueue_test, erase_test)
{mqmp->eraseMsgQueue("queue3");ASSERT_EQ(mqmp->exists("queue1"), true);ASSERT_EQ(mqmp->exists("queue2"), true);ASSERT_EQ(mqmp->exists("queue3"), false);ASSERT_EQ(mqmp->exists("queue4"), true);
}int main(int argc, char *argv[])
{testing::AddGlobalTestEnvironment(new MsgQueueTest);testing::InitGoogleTest(&argc, argv);return RUN_ALL_TESTS();
}

四、绑定信息管理模块的实现

1.功能需求

依然是这样：

1. Binding
   - 【1】 交换机名称
   - 【2】 队列名称
   - 【3】 binding_key
   - 【4】持久化标志

2. BindingMapper

   1. 对外接口
      • 【1】 创建/删除Binding表
      • 【2】 新增Binding数据
      • 【3】 持久化信息恢复
      • 【4】删除指定Binding数据
      • 【5】删除某个交换机对应的Binding数据
      • 【6】删除某个队列对应的Binding数据
   2. 成员
      -【1】 SqliteHelper句柄

3. BindingManager

   1. 对外接口
      【1】新增Binding数据
      【2】删除指定Binding数据
      【3】删除某个交换机对应的Binding数据
      【4】删除某个队列对应的Binding数据
      【5】获取指定Binding数据
      【6】获取某个交换机对应的所有Binding数据【因为交换机收到消息之后要根据绑定信息进行消息的分发】
   2. 成员
      【1】互斥锁
      【2】<交换机名称，交换机对应的所有绑定信息>的哈希表
      【3】BindingMapper句柄
      注意：

4. 删除交换机时要删除该交换机对应的所有Binding数据

5. 删除队列时也要删除该队列对应的所有Binding数据

2.Binding结构体实现

struct Binding
{using ptr = std::shared_ptr<Binding>;Binding() = default;Binding(const std::string &ename, const std::string &qname, const std::string &Binding_key, bool Durable): exchange_name(ename), queue_name(qname), binding_key(Binding_key), durable(Durable) {}std::string exchange_name;std::string queue_name;std::string binding_key;bool durable;
};

3.版本一

1.绑定信息持久化管理类

我要维护两个哈希表~
交换机的哈希表是用来拿到交换机的所有绑定信息从而进行消息路由的
队列的哈希表是用来删除队列的所有绑定信息的

如果没有队列的哈希表，那么删除队列所有的绑定信息时就需要遍历整个交换机哈希表的所有Binding::ptr

using BindingMap = std::unordered_map<std::string, std::vector<Binding::ptr>>;class BindingMapper
{using SelectCallback = int (*)(void *, int, char **, char **);public:BindingMapper(const std::string &dbfile): _helper(dbfile){if (!_helper.open()){default_fatal("绑定信息持久化管理类初始化失败，因为数据库打开失败");abort();}if (!createTable()){default_fatal("绑定信息持久化管理类初始化失败，因为创建表失败");abort();}}bool createTable(){static std::string create_sql = "create table if not 	exists binding_table(\exchange_name varchar(32),\queue_name varchar(32),\binding_key varchar(32),\durable int);";if (!_helper.exec(create_sql, nullptr, nullptr)){default_fatal("持久化管理模块建表失败");return false;}return true;}bool dropTable(){static std::string drop_sql = "drop table if exists binding_table;";if (!_helper.exec(drop_sql, nullptr, nullptr)){default_error("绑定信息持久化管理类删除表失败");return false;}return true;}bool insert(const Binding::ptr &bp){std::ostringstream insert_sql;insert_sql << "insert into binding_table values (";insert_sql << "'" << bp->exchange_name << "',";insert_sql << "'" << bp->queue_name << "',";insert_sql << "'" << bp->binding_key << "',";insert_sql << bp->durable << ");";if (!_helper.exec(insert_sql.str(), nullptr, nullptr)){default_error("新增持久化绑定信息失败，交换机名称：%s , 队列名称：%s",bp->exchange_name.c_str(),bp->queue_name.c_str());return false;}return true;}bool erase(const std::string &ename, const std::string &qname){std::ostringstream delete_sql;delete_sql << "delete from binding_table where exchange_name=";delete_sql << "'" << ename << "' and queue_name=";delete_sql << "'" << qname << "';";if (!_helper.exec(delete_sql.str(), nullptr, nullptr)){default_error("删除指定持久化绑定信息失败，交换机名称：%s , 队列名称：%s",ename.c_str(),qname.c_str());return false;}return true;}bool removeExchangeBinding(const std::string &ename){return _removeBinding("exchange_name", ename);}bool removeMsgQueueBinding(const std::string &qname){return _removeBinding("queue_name", qname);}BindingMap recoveryExchangeBinding(){return _recoveryBinding(&BindingMapper::SelectExchangeBindingCallback, "查询交换机对应持久化绑定信息失败");}BindingMap recoveryMsgQueueBinding(){return _recoveryBinding(&BindingMapper::SelectMsgQueueBindingCallback, "查询队列对应持久化绑定信息失败");}// 重构部分的公共代码
private:static int _selectCallback(void *arg, char **rows, const std::string &name){BindingMap *bmptr = static_cast<BindingMap *>(arg);std::vector<Binding::ptr> &vec = (*bmptr)[name]; //[]不存在则创建 一定要用引用,这样写代码可读性更好Binding::ptr bp = std::make_shared<Binding>();bp->exchange_name = rows[0];bp->queue_name = rows[1];bp->binding_key = rows[2];bp->durable = static_cast<bool>(std::stoi(rows[3]));vec.push_back(bp);return 0;}bool _removeBinding(const std::string &colname, const std::string name){std::ostringstream remove_sql;remove_sql << "delete from binding_table where " << colname << "=";remove_sql << "'" << name << "';";if (!_helper.exec(remove_sql.str(), nullptr, nullptr)){default_error("删除持久化绑定信息失败，%s ：%s",colname.c_str(),name.c_str());return false;}return true;}BindingMap _recoveryBinding(SelectCallback cb, const std::string &errmsg = ""){static std::string select_sql = "select * from binding_table;";BindingMap bmp;if (!_helper.exec(select_sql, cb, &bmp)){default_error("%s",errmsg.c_str());return BindingMap();}return bmp;}// 回调函数代码
private:static int SelectExchangeBindingCallback(void *arg, int column, char **rows, char **fields){return _selectCallback(arg, rows, rows[0]); // rows[0]是交换机名称}static int SelectMsgQueueBindingCallback(void *arg, int column, char **rows, char **fields){return _selectCallback(arg, rows, rows[1]); // rows[1]是队列名称}SqliteHelper _helper;
};

这其实就是SQL语句编写，然后又重构了一下而已

2.绑定信息管理类

class BindingManager
{
public:BindingManager() = default;using ptr = std::shared_ptr<BindingManager>;BindingManager(const std::string &dbfile): _mapper(dbfile){_exchange_map = _mapper.recoveryExchangeBinding();_queue_map = _mapper.recoveryMsgQueueBinding();}bool bind(const std::string &ename, const std::string &qname, const std::string &binding_key, bool durable){std::unique_lock<std::mutex> ulock(_mutex);// 1. 查找是否存在// 我们只需要在交换机这里检查是否存在即可,因为我们到时候路由的时候也是只根据交换机的哈希表进行路由的auto iter = _exchange_map.find(ename);if (iter != _exchange_map.end()){std::vector<Binding::ptr> &vec = iter->second;for (auto &bp : vec){if (bp->queue_name == qname){return true;}}}// 2. 构建bpBinding::ptr bp = std::make_shared<Binding>(ename, qname, binding_key, durable);// 3. 看是否需要持久化(因为只有当交换机持久化了+队列持久化了// 我们的绑定信息持久化才有意义,而我们不想增加模块间的耦合度,因此在虚拟机模块处理这个事情)if (durable){if (!_mapper.insert(bp)){default_error("绑定信息持久化失败: exchange_name: %s , queue_name: %s",ename.c_str(),qname.c_str());return false;}}// 4. push_back它_exchange_map[ename].push_back(bp);_queue_map[qname].push_back(bp);return true;}bool unBind(const std::string &ename, const std::string &qname){// 0. 加锁std::unique_lock<std::mutex> ulock(_mutex);// 1. 查找是否存在auto iter = _exchange_map.find(ename);if (iter == _exchange_map.end())return true;// 2. 从交换机当中删除bool ok = false;std::vector<Binding::ptr> &evec = iter->second;for (auto iter_bp = evec.begin(); iter_bp != evec.end(); ++iter_bp){Binding::ptr bp = *iter_bp;if (bp->queue_name == qname){// 看是否需要持久化if (bp->durable){if (!_mapper.erase(ename, qname)){default_error("解除持久化绑定信息失败");return false;}}ok = true;evec.erase(iter_bp);break;}}// 3. 从队列当中删除if (!ok)return true;std::vector<Binding::ptr> &qvec = _queue_map[qname];for (auto iter_bp = qvec.begin(); iter_bp != qvec.end(); ++iter_bp){Binding::ptr bp = *iter_bp;if (bp->exchange_name == ename){qvec.erase(iter_bp);break;}}return true;}std::vector<Binding::ptr> getExchangeBindings(const std::string &ename){std::unique_lock<std::mutex> ulock(_mutex);auto iter = _exchange_map.find(ename);if (iter != _exchange_map.end()){return iter->second;}return std::vector<Binding::ptr>();}Binding::ptr getBinding(const std::string &ename, const std::string &qname){std::unique_lock<std::mutex> ulock(_mutex);auto iter = _exchange_map.find(ename);if (iter != _exchange_map.end()){std::vector<Binding::ptr> &vec = iter->second;for (auto &bp : vec){if (bp->queue_name == qname){return bp;}}}return Binding::ptr();}bool exists(const std::string &ename, const std::string &qname){// 0. 加锁std::unique_lock<std::mutex> ulock(_mutex);// 1. 查找是否存在// 只要交换机哈希表当中有这个绑定信息,此时我们认为该绑定信息存在,因为交换机进行数据分发时,看的是交换机哈希表上的绑定信息auto iter = _exchange_map.find(ename);if (iter != _exchange_map.end()){std::vector<Binding::ptr> &vec = iter->second;for (auto &bp : vec){if (bp->queue_name == qname){return true;}}}return false;}bool removeExchangeBindings(const std::string &ename){std::unique_lock<std::mutex> ulock(_mutex);auto iter = _exchange_map.find(ename);if (iter != _exchange_map.end()){// 1. 从表中删除if (!_mapper.removeExchangeBinding(ename)){default_error("移除该交换机所有持久化绑定信息失败");return false;}std::vector<Binding::ptr> &evec = iter->second;for (auto &bp : evec){std::string qname = bp->queue_name;std::vector<Binding::ptr> &qvec = _queue_map[qname];// 小心迭代器失效问题for (auto it = qvec.begin(); it != qvec.end();){if ((*it)->exchange_name == ename){qvec.erase(it);}else++it;}}_exchange_map.erase(iter);return true;}return true;}bool removeMsgQueueBindings(const std::string &qname){std::unique_lock<std::mutex> ulock(_mutex);auto iter = _queue_map.find(qname);if (iter != _queue_map.end()){// 1. 从表中删除if (!_mapper.removeMsgQueueBinding(qname)){default_error("移除该队列所有持久化绑定信息失败");return false;}std::vector<Binding::ptr> &qvec = iter->second;for (auto &bp : qvec){std::string ename = bp->exchange_name;std::vector<Binding::ptr> &evec = _exchange_map[ename];// 小心迭代器失效问题for (auto it = evec.begin(); it != evec.end();){if ((*it)->queue_name == qname){evec.erase(it);}else++it;}}_queue_map.erase(iter);return true;}return true;}void clear(){std::unique_lock<std::mutex> ulock(_mutex);_mapper.dropTable();_exchange_map.clear();_queue_map.clear();}private:std::mutex _mutex;BindingMapper _mapper;BindingMap _exchange_map;BindingMap _queue_map;
};

仔细看的话，其实也没啥难的，就是STL容器的使用，写的时候注意力集中一点，仔细一点就问题不大

缺点是删除的效率实在是太低了
原因：Binding::ptr在每个BindingMap当中都存在  且  两个哈希表没有建立联系

4.版本二

using MsgQueueBindingMap = std::unordered_map<std::string, Binding::ptr>;
using BindingMap = std::unordered_map<std::string, MsgQueueBindingMap>;

1.绑定信息持久化管理类

class BindingMapper
{using SelectCallback = int (*)(void *, int, char **, char **);public:BindingMapper(const std::string &dbfile): _helper(dbfile){if (!_helper.open()){default_fatal("绑定信息持久化管理类初始化失败，因为数据库打开失败");abort();}if (!createTable()){default_fatal("绑定信息持久化管理类初始化失败，因为创建表失败");abort();}}bool createTable(){static std::string create_sql = "create table if not exists binding_table(\exchange_name varchar(32),\queue_name varchar(32),\binding_key varchar(32),\durable int);";if (!_helper.exec(create_sql, nullptr, nullptr)){default_fatal("持久化管理模块建表失败");return false;}return true;}bool dropTable(){static std::string drop_sql = "drop table if exists binding_table;";if (!_helper.exec(drop_sql, nullptr, nullptr)){default_error("绑定信息持久化管理类删除表失败");return false;}return true;}bool insert(const Binding::ptr &bp){std::ostringstream insert_sql;insert_sql << "insert into binding_table values (";insert_sql << "'" << bp->exchange_name << "',";insert_sql << "'" << bp->queue_name << "',";insert_sql << "'" << bp->binding_key << "',";insert_sql << bp->durable << ");";if (!_helper.exec(insert_sql.str(), nullptr, nullptr)){default_error("新增持久化绑定信息失败，交换机名称：%s , 队列名称：%s",bp->exchange_name.c_str(),bp->queue_name.c_str());return false;}return true;}bool erase(const std::string &ename, const std::string &qname){std::ostringstream delete_sql;delete_sql << "delete from binding_table where exchange_name=";delete_sql << "'" << ename << "' and queue_name=";delete_sql << "'" << qname << "';";if (!_helper.exec(delete_sql.str(), nullptr, nullptr)){default_error("删除指定持久化绑定信息失败，交换机名称：%s , 队列名称：%s",ename.c_str(),qname.c_str());return false;}return true;}bool removeExchangeBinding(const std::string &ename){return _removeBinding("exchange_name", ename);}bool removeMsgQueueBinding(const std::string &qname){return _removeBinding("queue_name", qname);}BindingMap recoveryExchangeBinding(){static std::string select_sql = "select * from binding_table;";BindingMap bmp;if (!_helper.exec(select_sql, &BindingMapper::SelectExchangeBindingCallback, &bmp)){default_error("查询交换机对应持久化绑定信息失败");return BindingMap();}return bmp;}// 重构部分的公共代码
private:bool _removeBinding(const std::string &colname, const std::string name){std::ostringstream remove_sql;remove_sql << "delete from binding_table where " << colname << "=";remove_sql << "'" << name << "';";if (!_helper.exec(remove_sql.str(), nullptr, nullptr)){default_error("删除持久化绑定信息失败，%s : %s",colname.c_str(),name.c_str());return false;}return true;}// 回调函数代码
private:static int SelectExchangeBindingCallback(void *arg, int column, char **rows, char **fields){BindingMap *bmptr = static_cast<BindingMap *>(arg);MsgQueueBindingMap &mqbm = (*bmptr)[rows[0]]; //[]不存在则创建 一定要用引用,这样写代码可读性更好Binding::ptr bp = std::make_shared<Binding>();bp->exchange_name = rows[0];bp->queue_name = rows[1];bp->binding_key = rows[2];bp->durable = static_cast<bool>(std::stoi(rows[3]));mqbm.insert(std::make_pair(bp->queue_name, bp));return 0;}SqliteHelper _helper;
};

这个比上个版本简单不少

2.绑定信息管理类

class BindingManager
{
public:BindingManager() = default;using ptr = std::shared_ptr<BindingManager>;BindingManager(const std::string &dbfile): _mapper(dbfile){_exchange_map = _mapper.recoveryExchangeBinding();}bool bind(const std::string &ename, const std::string &qname, const std::string &binding_key, bool durable){// 0. 加锁std::unique_lock<std::mutex> ulock(_mutex);// 1. 查找是否存在MsgQueueBindingMap &mqbm = _exchange_map[ename];if (mqbm.count(qname))return true;// 2. 构建bpBinding::ptr bp = std::make_shared<Binding>(ename, qname, binding_key, durable);// 3. 看是否持久化if (durable){if (!_mapper.insert(bp)){default_error("绑定信息持久化失败，交换机名称：%s , 队列名称：%s",ename.c_str(),qname.c_str());return false;}}// 4. 直接插入mqbm.insert(std::make_pair(qname, bp));return true;}bool unBind(const std::string &ename, const std::string &qname){// 0. 加锁std::unique_lock<std::mutex> ulock(_mutex);// 1. 查找auto eiter = _exchange_map.find(ename);if (eiter == _exchange_map.end())return true;MsgQueueBindingMap &mqbm = eiter->second;auto qiter = mqbm.find(qname);if (qiter == mqbm.end())return true;// 2. 看是否持久化Binding::ptr bp = qiter->second;if (bp->durable){if (!_mapper.erase(ename, qname)){default_error("解绑失败，交换机名称：%s ，队列名称：%s",ename.c_str(),qname.c_str());return false;}}// 3. 直接删除mqbm.erase(qname);return true;}MsgQueueBindingMap getExchangeBindings(const std::string &ename){std::unique_lock<std::mutex> ulock(_mutex);return _exchange_map[ename];}Binding::ptr getBinding(const std::string &ename, const std::string &qname){std::unique_lock<std::mutex> ulock(_mutex);MsgQueueBindingMap &mqbm = _exchange_map[ename];auto iter = mqbm.find(qname);if (iter == mqbm.end())return Binding::ptr();return iter->second;}bool exists(const std::string &ename, const std::string &qname){std::unique_lock<std::mutex> ulock(_mutex);MsgQueueBindingMap &mqbm = _exchange_map[ename];auto iter = mqbm.find(qname);if (iter == mqbm.end())return false;return true;}bool removeExchangeBindings(const std::string &ename){std::unique_lock<std::mutex> ulock(_mutex);if (!_mapper.removeExchangeBinding(ename)){default_error("删除交换机表失败");return false;}_exchange_map.erase(ename); // 因为队列那张表是依附于交换机这样表而存在的，所以我们直接暴力删表即可return true;}bool removeMsgQueueBindings(const std::string &qname){std::unique_lock<std::mutex> ulock(_mutex);if (!_mapper.removeMsgQueueBinding(qname)){default_error("删除队列表失败");return false;}// 遍历整张交换机表，删除队列信息for (auto &kv : _exchange_map){MsgQueueBindingMap &mqbm = kv.second;mqbm.erase(qname);}return true;}void clear(){std::unique_lock<std::mutex> ulock(_mutex);_mapper.dropTable();_exchange_map.clear();}private:std::mutex _mutex;BindingMapper _mapper;BindingMap _exchange_map;
};

这个效率就杠杠的了，唯一有点削微慢一丝丝的就是这个删除某个队列所有绑定信息时要遍历整个交换机表，但是因为MsgQueueBindingMap是哈希表，所以效率也挺高的

比版本快太多了

5.测试

#include "../mqserver/binding_upper_version.hpp"
#include <gtest/gtest.h>
using namespace ns_mq;BindingManager::ptr bmp;class BindingTest : public testing::Environment
{
public:virtual void SetUp(){bmp = std::make_shared<BindingManager>("test.db");}virtual void TearDown(){bmp->clear();}
};TEST(binding_test, recovery_test)
{ASSERT_EQ(bmp->exists("exchange1", "queue1"), false);ASSERT_EQ(bmp->exists("exchange1", "queue2"), false);ASSERT_EQ(bmp->exists("exchange1", "queue3"), false);ASSERT_EQ(bmp->exists("exchange2", "queue1"), false);ASSERT_EQ(bmp->exists("exchange2", "queue2"), true);ASSERT_EQ(bmp->exists("exchange2", "queue3"), true);
}TEST(binding_test, insert_test)
{ASSERT_NE(bmp.get(), nullptr);bmp->bind("exchange1", "queue1", "news.sport.basketball", true);bmp->bind("exchange1", "queue2", "news.sport.basketball", true);bmp->bind("exchange1", "queue3", "news.sport.basketball", true);bmp->bind("exchange2", "queue1", "news.music.pop", true);bmp->bind("exchange2", "queue2", "news.music.pure", true);bmp->bind("exchange2", "queue3", "news.music.#", true);ASSERT_EQ(bmp->exists("exchange1", "queue1"), true);ASSERT_EQ(bmp->exists("exchange1", "queue2"), true);ASSERT_EQ(bmp->exists("exchange1", "queue3"), true);ASSERT_EQ(bmp->exists("exchange2", "queue1"), true);ASSERT_EQ(bmp->exists("exchange2", "queue2"), true);ASSERT_EQ(bmp->exists("exchange2", "queue3"), true);
}TEST(binding_test, select_test)
{Binding::ptr bp = bmp->getBinding("exchange2", "queue2");ASSERT_NE(bp.get(), nullptr);ASSERT_EQ(bp->binding_key, std::string("news.music.pure"));auto bm = bmp->getExchangeBindings("exchange1");ASSERT_EQ(bm.size(), 3);
}TEST(binding_test, erase_test)
{bmp->removeMsgQueueBindings("queue1");ASSERT_EQ(bmp->exists("exchange1", "queue1"), false);ASSERT_EQ(bmp->exists("exchange1", "queue2"), true);ASSERT_EQ(bmp->exists("exchange1", "queue3"), true);ASSERT_EQ(bmp->exists("exchange2", "queue1"), false);ASSERT_EQ(bmp->exists("exchange2", "queue2"), true);ASSERT_EQ(bmp->exists("exchange2", "queue3"), true);bmp->removeExchangeBindings("exchange1");ASSERT_EQ(bmp->exists("exchange1", "queue1"), false);ASSERT_EQ(bmp->exists("exchange1", "queue2"), false);ASSERT_EQ(bmp->exists("exchange1", "queue3"), false);ASSERT_EQ(bmp->exists("exchange2", "queue1"), false);ASSERT_EQ(bmp->exists("exchange2", "queue2"), true);ASSERT_EQ(bmp->exists("exchange2", "queue3"), true);
}int main(int argc, char *argv[])
{testing::AddGlobalTestEnvironment(new BindingTest);testing::InitGoogleTest(&argc, argv);return RUN_ALL_TESTS();
}

以上就是项目第四弹：交换机、队列、绑定信息管理模块分析与代码实现的全部内容