仿 RabbitMQ 实现消息队列项目：交换机路由、队列隔离与消息可靠投递

整体过程图展示:

• 这里可能还是有点蒙，下面我们展开一步步展示设计，当整个项目疏通后就秒懂了。

三。基于交换机模块设计及实现

模块介绍

当客户端发布一条消息到交换机后，这条消息，应该被入队到该交换机绑定的哪些队列中？交换路由模块就是决定这件事情的。

首先对于交换机有三个模式（对于绑定的队列递送而言）：

• 广播：将消息入队到该交换机的所有绑定队列中。（也就是全发送）
• 直接：将消息入队到绑定信息中 binding_key 与消息 routing_key 一致的队列中。（也就是直接匹配）
• 主题：将消息入队到绑定信息中 binding_key 与 routing_key 是匹配成功的队列中。（也就是符合规则就发送）

因此需要定义不同的枚举来表示（这里我们定义在 proto 里）：

enum ExchangeType {
    UNKNOWTYPE = 0;
    DIRECT = 1;
    FANOUT = 2;
    TOPIC = 3;
};

模块设计图：

详细设计代码

Exchange 模块代码

基于交换机模块测试

下面基于 gtest 的全局模式的测试对 exchange 模块的增删查进行测试：

测试代码：

#include "../mqserver/exchange.hpp"
#include <gtest/gtest.h>

ExchangeManager::ptr emp;
class ExchangeTest : public testing::Environment {
public:
    virtual void SetUp() override {
        emp = std::make_shared<ExchangeManager>("./data/meta.db");
    }
    virtual void TearDown() override {
        emp->Clear();
        std::cout << "最后的清理！！\n";
    }
};

//因为涉及文件的永久删除故测试应联合 TEST(queue_test, insert_test) 这个模块一起进行！
TEST(exchange_test, insert_test) {
    google::protobuf::Map<std::string, std::string> map;
    map.insert({"k1","v1"});
    emp->DeclareExchange("exchange1", msg::ExchangeType::DIRECT,true,false, map);
    emp->DeclareExchange("exchange2", msg::ExchangeType::DIRECT,true,false, map);
    emp->DeclareExchange("exchange3", msg::ExchangeType::DIRECT,true,false, map);
    emp->DeclareExchange("exchange4", msg::ExchangeType::DIRECT,true,false, map);
    ASSERT_EQ(emp->Size(),4);
}

TEST(exchange_test, select_test) {
    ASSERT_EQ(emp->ExchangeIsExists("exchange1"),true);
    ASSERT_EQ(emp->ExchangeIsExists("exchange2"),true);
    ASSERT_EQ(emp->ExchangeIsExists("exchange3"),true);
    ASSERT_EQ(emp->ExchangeIsExists("exchange4"),true);
    Exchange::ptr exp = emp->SelectExchange("exchange2");
    ASSERT_NE(exp.get(),nullptr);
    ASSERT_EQ(exp->name,"exchange2");
    ASSERT_EQ(exp->durable,true);
    ASSERT_EQ(exp->auto_delete,false);
    ASSERT_EQ(exp->type, msg::ExchangeType::DIRECT);
    ASSERT_EQ(exp->GetArgs(), std::string("k1=v1&"));
}

TEST(exchange_test, remove_test) {
    emp->DeleteExchange("exchange2");
    Exchange::ptr exp = emp->SelectExchange("exchange2");
    ASSERT_EQ(exp.get(),nullptr);
    ASSERT_EQ(emp->ExchangeIsExists("exchange2"),false);
}

int main(int argc,char*argv[]) {
    testing::InitGoogleTest(&argc, argv);
    testing::AddGlobalTestEnvironment(new ExchangeTest());
    int ret = RUN_ALL_TESTS();
    if(!ret) std::cout << "成功的一次全局测试\n";
    return 0;
}

测试效果：

对应 SQLite3 数据库查询（gtest 测试最后不进行整体 Clear 操作）：

• 全部符合预期。

四。基于队列模块设计及实现

模块介绍

• 这里与上面的交换机模块类似，即通过队列管理者文件内的队列集合以及内存中的队列集合（提供的接口也差不多）。

全局视角看待大致过程：

详细设计代码

MsgQueue 模块代码

基于队列模块测试

下面基于 gtest 的全局模式的测试对 queue 模块的增删查进行测试：

测试代码：

#include "../mqserver/msgqueue.hpp"
#include <gtest/gtest.h>

MsgqueueManager::ptr emp;
class MsgqueueTest : public testing::Environment {
public:
    virtual void SetUp() override {
        emp = std::make_shared<MsgqueueManager>("./data/meta.db");
    }
    virtual void TearDown() override {
        // emp->Clear();
        std::cout << "最后的清理！！\n";
    }
};

//因为涉及文件的永久删除故测试应联合 TEST(queue_test, insert_test) 这个模块一起进行！
TEST(msgqueue_test, insert_test) {
    google::protobuf::Map<std::string, std::string> map;
    map.insert({"k1","v1"});
    emp->DeclareMsgqueue("mq1",1,true,false, map);
    emp->DeclareMsgqueue("mq2",1,true,false, map);
    emp->DeclareMsgqueue("mq3",1,true,false, map);
    emp->DeclareMsgqueue("mq4",1,true,false, map);
    ASSERT_EQ(emp->Size(),4);
}

TEST(msgqueue_test, select_test) {
    ASSERT_EQ(emp->MsgqueueIsExists("mq1"),true);
    ASSERT_EQ(emp->MsgqueueIsExists("mq2"),true);
    ASSERT_EQ(emp->MsgqueueIsExists("mq3"),true);
    ASSERT_EQ(emp->MsgqueueIsExists("mq4"),true);
    Msgqueue::ptr exp = emp->SelectMsgqueue("mq2");
    ASSERT_NE(exp.get(),nullptr);
    ASSERT_EQ(exp->name,"mq2");
    ASSERT_EQ(exp->exclusive,1);
    ASSERT_EQ(exp->auto_delete,false);
    ASSERT_EQ(exp->GetArgs(), std::string("k1=v1&"));
}

TEST(msgqueue_test, remove_test) {
    emp->DeleteMsgqueue("mq2");
    Msgqueue::ptr exp = emp->SelectMsgqueue("mq2");
    ASSERT_EQ(exp.get(),nullptr);
    ASSERT_EQ(emp->MsgqueueIsExists("mq2"),false);
}

int main(int argc,char*argv[]) {
    testing::InitGoogleTest(&argc, argv);
    testing::AddGlobalTestEnvironment(new MsgqueueTest());
    int ret = RUN_ALL_TESTS();
    if(!ret) std::cout << "成功的一次全局测试\n";
    return 0;
}

测试效果：

对应 SQLite3 数据库查询（gtest 测试最后不进行整体 Clear 操作）：

• 符合预期。

五。基于绑定信息模块设计及实现

模块介绍：

这里我们设计的是绑定信息（一个交换机对应有一个和很多队列的一个绑定信息集合）。

绑定信息具体成员：

• 交换机名称。
• 队列名称。
• Binding_key(进行对应交换机路由匹配找队列有关)。

如下：

struct Binding {
    using ptr = std::shared_ptr<Binding>;
    // 一组绑定关系
    std::string exchange_name;
    std::string msgqueue_name;
    std::string binding_key;
    
    Binding(const std::string &ename,const std::string &qname,const std::string &key)
        :exchange_name(ename),msgqueue_name(qname),binding_key(key){}
};

下面还是基于上面的那俩模块进行的基于文件与内存级别实现（这里只有对应的交换机和队列都是持久化的，绑定信息才能持久化并存储文件中）。

其次就是这里我们实现的是一个交换机对应的是一个 map(绑定的队列与对应 binding 的映射)；但是为什么这么实现？

这里绑定采取这种映射而不是直接拿交换机映射 binding（这样每次删除指定队列先关绑定需要遍历 1 次交换机数组 + 每次对应的 binding）而采取交换机–Queue+binding 映射方式，只需要遍历一遍交换机直接拿到 second 进行删除 second[Queue]即可。

如下：

using MsgqueueBindingMap = std::unordered_map<std::string, Binding::ptr>; // MsgQueue---binding
using ExchangeMBMap = std::unordered_map<std::string, MsgqueueBindingMap>; // Exchange---mb

下面剩下的模式几乎和上面他俩相同：

**基于内存 + 文件管理：**内存管理模块就是进行绑定信息的增删，查取并恢复进内存，根据队列名/交换机名，进行相关绑定信息删除。管理者就是对外提供调用接口：绑定，解绑，根据交换机名/队列名移除绑定信息，获取交换机所有相关绑定信息 map，获取特定绑定信息等。

全局视角看待大致过程：

详细设计代码

Binding 模块代码

基于绑定信息模块测试

下面基于 gtest 的全局模式的测试对 binding 模块的增删查进行测试：
测试代码：

#include "../mqserver/binding.hpp"
#include <gtest/gtest.h>

BindingManager::ptr bmp;
class bindingTest : public testing::Environment {
public:
    virtual void SetUp() override {
        bmp = std::make_shared<BindingManager>("./data/meta.db");
    }
    virtual void TearDown() override {
        // bmp->Clear();
    }
};

//因为涉及文件的永久删除故测试应联合 TEST(queue_test, insert_test) 这个模块一起进行！
TEST(queue_test, insert_test) {
    bmp->Bind("exchange1","queue1","news.music.#",true);
    bmp->Bind("exchange1","queue2","news.sport.#",true);
    bmp->Bind("exchange1","queue3","news.gossip.#",true);
    bmp->Bind("exchange2","queue1","news.music.pop",true);
    bmp->Bind("exchange2","queue2","news.sport.football",true);
    bmp->Bind("exchange2","queue3","news.gossip.#",true);
    ASSERT_EQ(bmp->Size(),6);
}

TEST(queue_test, select_test) {
    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);
    ASSERT_EQ(bmp->Exists("exchange3","queue3"),false);
    Binding::ptr bp = bmp->GetBinding("exchange1","queue1");
    ASSERT_NE(bp.get(),nullptr);
    ASSERT_EQ(bp->exchange_name, std::string("exchange1"));
    ASSERT_EQ(bp->msgqueue_name, std::string("queue1"));
    ASSERT_EQ(bp->binding_key, std::string("news.music.#"));
}

TEST(queue_test, remove_exchange_test) {
    bmp->RemoveExchangeBindings("exchange1");
    ASSERT_EQ(bmp->Exists("exchange1","queue1"),false);
    ASSERT_EQ(bmp->Exists("exchange1","queue2"),false);
    ASSERT_EQ(bmp->Exists("exchange1","queue3"),false);
}

int main(int argc,char*argv[]) {
    testing::AddGlobalTestEnvironment(new bindingTest);
    testing::InitGoogleTest(&argc, argv);
    int ret = RUN_ALL_TESTS();
    if(!ret) std::cout << "成功的一次全局测试\n";
    std::shared_ptr<bindingTest>();
    return 0;
}

基于插入，查找，操作测试：

数据库内：

• 符合预期。

文件恢复到内存的操作测试：

解除绑定测试：

• 数据库中确实被移除了，符合预期。

根据交换机与队列分别进行移除测试：

• 内存效果与数据库效果都符合预期。

六。基于消息模块设计及实现

模块介绍

• 消息包含：有效载荷（属性 内容 持久化标志），长度（在文件中的长度），偏移量（在文件中）。

具体如下（proto 文件）：

enum DeliveryMode {
    UNKNOWMODE = 0;
    UNDURABLE = 1;
    DURABLE = 2;
};
//这里 BasicProperties 在外面而 Payload 进行嵌套：
//BasicProperties 需要外界单独创建穿进去进行消息插入故能外界直接创建，但是 payload 只能通过先创建 Message 对象在进行创建更合乎情理
message BasicProperties {
    string id = 1;
    DeliveryMode delivery_mode = 2;
    string routing_key = 3;
};
message Message {
    message Payload {
        BasicProperties properties = 1;
        string body = 2;
        string valid = 3;
    };
    Payload payload = 1;
    uint32 offset = 2; //对应有效载荷的文件偏移量（方便移除写 0 覆盖）
    uint32 length = 3; //有效载荷长度 (方便移除写 0 覆盖)
};

本模块相对于前面三个确实有点复杂（代码也是比较多的），下面梳理下一些要点：

以队列形式管理该队列的内存消息 + 文件消息：

消息是否持久化与它要插入的队列是否持久化有关。

1·文件操作：

• 每个队列有个文件用于存储消息（四字节长度 + 有效载荷（属性 内容 是否有效））。
• 删除操作：这里只是往文件对应位置标记"0"为无效（把传进来的 message 对应位置标记无效再写回去），而有恢复操作：就是把文件有效的数据（写入文件后对应位置是"1"的消息）转移到临时文件再重命名成对应消息文件即可），这种方式避免了文件频繁被操作带来不便（这里恢复操作是符合特定要求（持久化的消息总量大于 2000，且其中有效比例低于 50% 则需要持久化））。
• 其他细节见代码。

2·queuemessage 部分：

• 这里提供插入 删除 恢复 回收 取出消息 等操作（持久化的话，文件也跟着操作）。
• 对于删除：每次删除完（也就是标记完），都要去检查对应的文件有无效消息占比，进行是否回收等（这里恢复不仅要文件，还要与内存同步（比如文件变了对应的 msg 的偏移量等就变了，需要及时同步更新内存的。））
• 其他细节见代码。

3·消息总的管理者：

• 提供初始化与销毁队列，插入队列消息 取出队列消息 对队列消息进行应答等操作。
• 这里接口其实就是对之前封装的进行调用即可，具体见代码。

下图展示了关系图：

详细设计代码

Message 模块代码

基于消息模块测试

下面基于 gtest 的全局模式的测试对 message 模块的增删查进行测试：
测试代码：

#include "../mqserver/message.hpp"
#include <gtest/gtest.h>

MessageManager::ptr mp;
class MsgqueueTest : public testing::Environment {
public:
    virtual void SetUp() override {
        mp = std::make_shared<MessageManager>("./message");
    }
    virtual void TearDown() override {
        // mp->Clear();
        std::cout << "最后的清理！！\n";
    }
};

//先要 insert 让他写入 queue1 文件，再次跑动加载：
TEST(message_test2, recovery_test) {
    mp->InitQueueMessage("queue1");
    ASSERT_EQ(mp->GetAll_count("queue1"),4);
    Message::ptr msg1 = mp->Front("queue1");
    ASSERT_NE(msg1.get(),nullptr);
    ASSERT_EQ(msg1->payload().body(), std::string("Hello World-1"));
    ASSERT_EQ(mp->GetAll_count("queue1"),3);
    ASSERT_EQ(mp->Waitack_count("queue1"),1);
    Message::ptr msg2 = mp->Front("queue1");
    ASSERT_NE(msg2.get(),nullptr);
    ASSERT_EQ(msg2->payload().body(), std::string("Hello World-2"));
    ASSERT_EQ(mp->GetAll_count("queue1"),2);
    ASSERT_EQ(mp->Waitack_count("queue1"),2);
    Message::ptr msg3 = mp->Front("queue1");
    ASSERT_NE(msg3.get(),nullptr);
    ASSERT_EQ(msg3->payload().body(), std::string("Hello World-3"));
    ASSERT_EQ(mp->GetAll_count("queue1"),1);
    ASSERT_EQ(mp->Waitack_count("queue1"),3);
    Message::ptr msg4 = mp->Front("queue1");
    ASSERT_NE(msg4.get(),nullptr);
    ASSERT_EQ(msg4->payload().body(), std::string("Hello World-4"));
    ASSERT_EQ(mp->GetAll_count("queue1"),0);
    ASSERT_EQ(mp->Waitack_count("queue1"),4);
}

int main(int argc,char*argv[]) {
    testing::InitGoogleTest(&argc, argv);
    testing::AddGlobalTestEnvironment(new MsgqueueTest());
    int ret = RUN_ALL_TESTS();
    if(!ret) std::cout << "成功的一次全局测试\n";
    return 0;
}

测试效果：

基于插入的测试：

• 内存级别与文件级别都符合预期。

基于取出 + 顺序的测试（这里没有确认以及更新故文件消息不变）：

• 全部符合预期。

基于确认应答后文件对应被标记操作测试：

• 文件中对应的消息确实也被标记 0 了。

下面测试下从文件中恢复这四条消息，然后判断下对应取出的顺序等是否合理：

• 因为取出后没确认应答，故文件中还是标记的 1，符合预期。

七。本篇小结

本文介绍了这个项目的 server 大模块的几个分支，基于书写到测试并未发现问题，后续将更新对应的 路由匹配 consumer channel connection broker（服务器）等模块。