基于Linux和C++实现的RabbitMQ风格消息队列：设计与实现

文章目录

0. 前言
1. 需求分析
- ① 核心部分
- ② 核心API
- ③ 交换机类型
- ④ 持久化
- ⑤ 网络通信
- ⑥ 消息应答
2. 模块划分
- Ⅰ 服务端模块
- - ① 持久化数据管理中心模块
  - ② 虚拟机管理模块
  - ③ 交换机路由模块
  - ④ 消费者管理模块
  - ⑤ 信道管理模块
  - ⑥ 连接管理模块
  - ⑦ Broker Server模块
- Ⅱ 客户端模块
- - ① 消费者管理
  - ② 信道请求模块
  - ③ 通信连接模块
3. 项目模块关系图
4. 项目创建
4. 公共模块实现
- 4.1 log日志打印工具
- 4.2 helper工具
- 4.3 线程池类
- 4.4 proto.proto：定义请求/响应参数
- 4.5 message.proto
5. 部分需求设计
- 5.1 网络通信协议设计
- 5.2 设计应用层协议
6. 服务器模块实现
- MQ Broker Server
7. 客户端模块实现
- ① 订阅者模块
- ② 信道管理模块
- ③ 异步工作线程实现
- ④ 连接管理模块
- ⑤ 生产者客户端
- ⑥ 消费者客户端

0. 前言

源码

1. 需求分析

① 核心部分

⽣产者 (Producer)
消费者 (Consumer)
中间⼈ (Broker)
发布 (Publish)
订阅 (Subscribe)

单个⽣产者，单个消费者：

在这里插入图片描述

多个⽣产者，多个消费者：

在这里插入图片描述

对于上面的内容，Broker Sevrer是最核心的部分，负责消息的存储和转发。

在AMQP(Advanced Message Queuing Protocol)-⾼级消息队列协议，⼀个提供统⼀消息服务的应⽤层标准⾼级消息队列协议，为⾯向消息的中间件设计，使得遵从该规范的客户端应⽤和消息中间件服务器的全功能互操作成为可能)模型中，也就是消息中间件服务器Broker中，⼜存在以下概念：

虚拟机(VirtualHost) ：类似于MySQL的database，是⼀个逻辑上的集合。⼀个BrokerServer上可以存在多个VirtualHost。
交换机(Exchange) ：⽣产者把消息先发送到Broker的Exchange上，再根据不同的规则,把消息转发给不同的Queue。
队列(Queue) ：真正⽤来存储消息的部分，每个消费者决定⾃⼰从哪个Queue上读取消息。
绑定(Binding) ：Exchange和Queue之间的关联关系，Exchange和Queue可以理解成"多对多"关系，使⽤⼀个关联表就可以把这两个概念联系起来
消息(Message) ：传递的内容

Exchange和Queue可以理解成"多对多"关系；
即⼀个Exchange可以绑定多个Queue(可以向多个Queue中转发消息)⼀个Queue也可以被多个Exchange绑定(⼀个Queue中的消息可以来⾃于多个Exchange)
如下图所示：

具体交互步骤：
在这里插入图片描述

上述数据结构，既需要在内存中存储，也需要在硬盘中存储
• 内存存储：⽅便使⽤
• 硬盘存储：重启数据不丢失

② 核心API

对于 Broker Server 来说, 要实现以下核⼼ API，通过这些 API 来实现消息队列的基本功能：

创建交换机 (exchangeDeclare)
销毁交换机 (exchangeDelete)
创建队列 (queueDeclare)
销毁队列 (queueDelete)
创建绑定 (queueBind)
解除绑定 (queueUnbind)
发布消息 (basicPublish)
订阅消息 (basicConsume)
确认消息 (basicAck)
取消订阅 (basicCancel)

另⼀⽅⾯, Producer 和 Consumer 则通过⽹络的⽅式, 远程调⽤这些 API, 实现 生产者消费者模型

关于 VirtualHost：对于 RabbitMQ 来说, VirtualHost 也是可以随意创建删除的

③ 交换机类型

RabbitMQ 中常见的四种交换机类型是：

Direct Exchange（直接交换机） ：
- 生产者在发送消息时，直接指定消息要发送到的队列名。消息的路由键（Routing Key）必须与队列绑定时指定的路由键完全匹配，消息才会被发送到相应的队列中。
Fanout Exchange（扇型交换机） ：

生产者发送的消息会被复制到该交换机绑定的所有队列中，不考虑消息的路由键。（消息的广播）

Topic Exchange（主题交换机）：

在绑定队列到主题交换机时，指定一个字符串作为绑定键（bindingKey）。生产者发送消息时指定一个字符串作为路由键（routingKey）。当路由键与绑定键按照一定的匹配规则匹配时，消息会被投递到相应的队列中。

Headers Exchange（头部交换机） ：

使用消息的头部信息来决定将消息发送到哪个队列。不依赖于路由键，而是根据消息头部的键值对匹配进行路由。

④ 持久化

Exchange, Queue, Binding, Message 等数据都有持久化需求
当程序重启 / 主机重启, 需要保证上述内容不丢失：

我们通过数据库来存储信息，对于该项目，使用sqlite轻量级数据库，便于操作使用；

⑤ 网络通信

生产者和消费者都是客户端程序，Broker 则是作为服务器，通过网络进行通信。

在⽹络通信的过程中, 客户端部分要提供对应的 api, 来实现对服务器的操作：

创建 Connection
关闭 Connection
创建 Channel
关闭 Channel
创建队列 (queueDeclare)
销毁队列 (queueDelete)
创建交换机 (exchangeDeclare)
销毁交换机 (exchangeDelete)
创建绑定 (queueBind)
解除绑定 (queueUnbind)
发布消息 (basicPublish)
订阅消息 (basicConsume)
确认消息 (basicAck)
取消订阅（basicCancel）

可以看出来：在 Broker 的基础上, 客户端还要增加 Connection 操作和 Channel 操作；
• Connection 对应⼀个 TCP 连接；
• Channel 则是 Connection 中的逻辑通道；

⼀个 Connection 中可以包含多个 Channel。 Channel 和 Channel 之间的数据是独⽴的，不会相互⼲扰。这样做主要是为了能够更好的复⽤ TCP 连接, 达到⻓连接的效果, 避免频繁的创建关闭 TCP 连接。

Connection 可以理解成⼀根⽹线，而Channel则是⽹线⾥具体的线缆

⑥ 消息应答

被消费的消息, 需要进行应答。应答模式分成两种：

• 自动应答 : 消费者只要消费了消息, 就算应答完毕了，Broker 直接删除这个消息
• 手动应答 ：消费者⼿动调⽤应答接⼝，Broker 收到应答请求之后，删除这个消息

手动应答为了保证消息确实被消费者处理成功；在⼀些对于数据可靠性要求高的场景较为常用；

2. 模块划分

Ⅰ 服务端模块

① 持久化数据管理中心模块

在数据管理模块中管理交换机，队列，队列绑定，消息等部分数据数据。

交换机管理：
- 管理信息：名称，类型，是否持久化标志，是否(⽆⼈使⽤时)⾃动删除标志，其他参数，…
- 管理操作：恢复历史信息，声明，删除，获取，判断是否存在
队列管理：
- 管理信息：名称，是否持久化标志，是否独有标志，是否(⽆⼈使⽤时)⾃动删除标志，其他参数，…
- 管理操作：恢复历史信息，声明，删除，获取，判断是否存在
绑定管理：
- 管理信息：交换机名称，队列名称，绑定主题
- 管理操作：恢复历史信息，绑定，解绑，解除交换机关联绑定信息，解除队列关联绑定信息，获取交换机关联绑定信息
消息管理：
- 管理信息
  - 属性：消息ID，路由主题，持久化模式标志
  - 消息内容
  - 有效标志（持久化需要）
  - 持久化位置（内存中）
  - 持久化消息⻓度（内存中）
- 管理操作：恢复历史信息，向指定队列新增消息，获取指定队列队⾸消息，确认移除消息

这几个核⼼概念数据都需要在内存和硬盘中存储的。

以内存存储为主，主要是保证快速查找信息进⾏处理
以硬盘存储为辅，主要是保证服务器重启之后，之前的信息都可以正常保持

② 虚拟机管理模块

因为交换机/队列/绑定都是基于虚拟机为单元整体进⾏操作的，因此虚拟机是对以上数据管理模块的整
合模块。

虚拟机管理信息：
- 交换机数据管理模块句柄
- 队列数据管理模块句柄
- 绑定数据管理模块句柄
- 消息数据管理模块句柄
虚拟机对外操作：
- 提供虚拟机内交换机声明，交换机删除操作。
- 提供虚拟机内队列声明，队列删除操作。
- 提供虚拟机内交换机-队列绑定，解除绑定操作。
- 获取交换机相关绑定信息
虚拟机管理操作：
- 创建虚拟机
- 查询虚拟机
- 删除虚拟机

③ 交换机路由模块

当客户端发布⼀条消息到交换机后，这条消息应该被入队到该交换机绑定的哪些队列？ 这个功能由交换路由模块决定。

在绑定信息中有⼀个binding_key，⽽每条发布的消息中有⼀个routing_key，能否⼊队取决于两个要素：交换机类型 和 key

⼴播： 将消息⼊队到该交换机的所有绑定队列中
直接： 将消息⼊队到绑定信息中binding_key与消息routing_key⼀致的队列中
主题： 将消息⼊队到绑定信息中binding_key与routing_key是匹配成功的队列中

binding_key

是由数字字⺟下划线构成的, 并且使⽤ . 分成若⼲部分。

例如： news.music.# ，这⽤于表⽰交换机绑定的当前队列是⼀个⽤于发布⾳乐新闻的队列。

⽀持 * 和 # 两种通配符，但是 * # 只能作为 . 切分出来的独⽴部分, 不能和其他数字字⺟混⽤,
- ⽐如 a.*.b 是合法的， a.*a.b 是不合法的
- * 可以匹配任意⼀个单词（注意是单词不是字⺟）
- # 可以匹配任意零个或者多个单词（注意是单词不是字⺟

routing_key
是由数据、字⺟和下划线构成，并且可以使⽤ . 划分成若⼲部分。

例如： news.music.pop ，这⽤于表⽰当前发布的消息是⼀个流⾏⾳乐的新闻

④ 消费者管理模块

消费者管理是以队列为单元的，因为每个消费者都会在开始的时候订阅⼀个队列的消息，当队列中有消息后，会将队列消息轮询推送给订阅了该队列的消费者。

因此操作流程通常是，从队列关联的消息管理中取出消息，从队列关联的消费者中取出⼀个消费者，然后将消息推送给消费者（这就是发布订阅中负载均衡的⽤法）。

消费者信息：
- 标识
- 订阅队列名称
- ⾃动应答标志（决定了⼀条消息推送给消费者后，是否需要等待收到确认后再删除消息）
- 消息处理回调函数指针（⼀个消息发布后调⽤回调，选择消费者进⾏推送…）
  - void(const std::string& tag, const BasicProperties& p, const std::string& body)
消费者管理：
- 添加，删除，轮询获取指定队列的消费者，移除队列所有消费者等操作

⑤ 信道管理模块

在AMQP模型中，通信连接除了Connection概念，还有Channel的概念，Channel是针对Connection连接的⼀个更细粒度的通信信道，多个Channel可以使用同⼀个通信连接Connection进⾏通信，但是同⼀个Connection的Channel之间相互独⽴。

信道模块就是再次将上述模块进⾏整合提供服务的模块：

管理信息：
- 信道ID
- 信道关联的消费者
- 信道关联的连接
- 信道关联的虚拟机
- ⼯作线程池（⼀条消息被发布到队列后，需要将消息推送给订阅了对应队列的消费者，过程由
  线程池完成）
管理操作：
- 提供声明&删除交换机操作（删除交换机的同时删除交换机关联的绑定信息）
- 提供声明&删除队列操作（删除队列的同时，删除队列关联的绑定信息，消息，消费者信息）
- 提供绑定&解绑队列操作
- 提供订阅&取消订阅队列消息操作
- 提供发布&确认消息操作

⑥ 连接管理模块

该项目的服务器是通过muduo库来实现底层通信的，此处的连接管理，主要是对muduo库中的Connection进⾏⼆次封装管理，并额外提供项⽬所需操作。

管理信息：
- 连接关联的信道
- 连接关联的muduo库Connection
管理操作：
- 新增连接，删除连接，获取连接，打开信道，关闭信道。

⑦ Broker Server模块

整合以上所有模块，并搭建⽹络通信服务器，实现与客⼾端⽹络通信，能够识别客⼾端请求，并提供客户端请求的处理服务。

管理信息：

虚拟机管理模块句柄
消费者管理模块句柄
连接管理模块句柄
⼯作线程池句柄
muduo库通信所需元素

Ⅱ 客户端模块

① 消费者管理

消费者在客户端的存在感较低，从用户的使用角度，创建⼀个信道就可以通过信道完成所有的操作，因此对于消费者的认知更多是在订阅时传入⼀个消费者标识，且项目实现仅是⼀个信道只能创建订阅⼀个队列，也即只能创建⼀个消费者，⼀⼀对应更弱化了消费者的存在。

消费者信息：
- 标识
- 订阅队列名称
- ⾃动应答标志（决定了⼀条消息推送给消费者后，是否需要等待收到确认后再删除消息）
- 消息处理回调函数指针（⼀个消息发布后调⽤回调，选择消费者进⾏推送…）
消费者管理
- 添加，删除，轮询获取指定队列的消费者，移除队列所有消费者等操作

② 信道请求模块

与服务端的信道类似，客⼾端这边在AMQP模型中，也是除了通信连接Connection概念外，还有⼀个Channel的概念，Channel是针对Connection连接的⼀个更细粒度的通信信道，多个Channel可以使⽤同⼀个通信连接Connection进⾏通信，但是同⼀个Connection的Channel之间相互独⽴。

信道管理信息：
- 信道ID
- 信道关联的通信连接
- 信道关联的消费者
- 请求对应的响应信息队列（这⾥队列使⽤hash表，以便于查找指定的响应）
- 互斥锁&条件变量（⼤部分的请求都是阻塞操作，发送请求后需要等到响应才能继续，但是muduo库的通信是异步的，因此需要我们⾃⼰在收到响应后，通过判断是否是等待的指定响应
  来进⾏同步）
信道管理操作：
- 提供创建信道操作
- 提供删除信道操作
- 提供声明交换机操作（强断⾔-有则OK，没有则创建）
- 提供删除交换机
- 提供创建队列操作（强断⾔-有则OK，没有则创建）
- 提供删除队列操作
- 提供交换机-队列绑定操作
- 提供交换机-队列解除绑定操作
- 提供添加订阅操作
- 提供取消订阅操作
- 提供发布消息操作

③ 通信连接模块

向⽤⼾提供⼀个⽤于实现⽹络通信的Connection对象，从其内部可创建出粒度更轻的Channel对象，⽤于与服务端进行网络通信。

管理信息：
- 连接关联的实际⽤于通信的muduo::net::Connection连接
- 连接关联的信管理句柄（实现信道的增删查）
- 连接关联的EventLoop异步循环⼯作线程
- 异步⼯作线程池（⽤于对收到服务器推送过来的消息进⾏处理的线程池）
管理操作：
- 提供创建Channel信道的操作
- 提供删除Channel信道的操作

3. 项目模块关系图

请添加图片描述

4. 项目创建

Linux机器上创建mq项目，并规划开发⽬录，利用Makefile组织项目。

demo：编写⼀些功能⽤例时所在的⽬录
common：公共模块代码（线程池，数据库访问，⽂件访问，⽇志打印，pb相关，以及其他的⼀些琐碎功能模块代码）
client: 客⼾端模块代码
server: 服务器模块代码
test：单元测试
third: ⽤到的第三⽅库存放⽬录

4. 公共模块实现

4.1 log日志打印工具

我们实现一个日志打印工具，使可以在代码编写时通过 LOG(format, ...) 的形式进行日志打印。

#ifndef __M_LOG_H__
#define __M_LOG_H__

#include <iostream>
#include <ctime>

#define DBG_LEVEL 0
#define INF_LEVEL 1
#define ERR_LEVEL 2
#define DEFAULT_LEVEL DBG_LEVEL
#define LOG(level_str, level, format, ...) {\
    if(level >= DEFAULT_LEVEL) {\
        time_t t = time(nullptr);\
        struct tm* ptm = localtime(&t);
        char time_str[32];
        strftime(time_str, 31, "%H:%M:%S", ptm);\
        printf("[%s][%s][%s:%d]\t" format "\n", lev_str, time_str, __FILE__, __LINE__, ##__VA_ARGS__);\
    }\
}

#define DLOG(format, ...) LOG("DBG", DBG_LEVEL, format, ##__VA_ARGS__)
#define ILOG(format, ...) LOG("INF", INF_LEVEL, format, ##__VA_ARGS__)
#define ELOG(format, ...) LOG("ERR", ERR_LEVEL, format, ##__VA_ARGS__)

#endif

4.2 helper工具

helper模块 用于工具类用于实现一些通用的功能：

由于代码实现较长，对于helper工具的实现，主要在于以下几个类：

sqlite数据库相关操作（SqliteHelper）
- 判断库是否存在
- 创建并打开库 / 关闭库 / 删除库
- 启动 / 提交 / 回滚事务
- 执行语句
字符串相关操作（StrHelper）
- 提供字符串分割功能
UID相关操作（UUIDHelper）

UUID(Universally Unique Identifier), 也叫通⽤唯⼀识别码，通常由32位16进制数字字符组成。

UUID的标准型式包含32个16进制数字字符，以连字号分为五段，形式为8-4-4-4-12的32个字符，如：550e8400-e29b-41d4-a716-446655440000。

在这⾥，uuid⽣成，我们采⽤⽣成8个随机数字，加上8字节序号，共16字节数组⽣成32位16进制字符的组合形式来确保全局唯⼀的同时能够根据序号来分辨数据。

文件相关操作（FileHelper）
- ⽂件是否存在判断
- ⽂件⼤⼩获取
- ⽂件读/写
- ⽂件创建/删除
- ⽬录创建/删除
```
class SqliteHelper{};
class StrHelper{};
class UUIDHelper{};
class FileHelper{};
```

4.3 线程池类

该线程池类，能够管理并执行多个任务。线程池在构造时启动指定数量的线程，每个线程不断从任务队列中取任务并执行。通过 push 方法，用户可以将任务添加到队列中，任务会被封装为 std::packaged_task 并返回一个 std::future 对象，以便在任务完成后获取结果。

class ThreadPool
{
public:
    using ptr = std::shared_ptr<ThreadPool>;
    using Functor = std::function<void(void)>;

    ThreadPool(int thread_count = 1);
    ~ThreadPool();

    void stop();

    template<typename F, typename ...Args>
    auto push(F&& f, Args&& ...args) -> std::future<decltype(f(args...))>;

private:
    void entry();

private:
    std::atomic<bool> _stop;
    std::vector<Functor> _taskpool; // 任务池
    std::mutex _mutex; // 互斥锁
    std::condition_variable _cv;
    std::vector<std::thread> _threads; // 存储线程
};

4.4 proto.proto：定义请求/响应参数

该文件定义了客户端服务器通信时所用到的请求和响应；由于参数需要进⾏⽹络传输以及序列化，所以需要将参数定义在pb⽂件中。

比如：打开信道与关闭信道的请求：

message openChannelRequest{
    string rid = 1;
    string cid = 2;
};
message closeChannelRequest{
    string rid = 1;
    string cid = 2;
};

一个创建交换机的请求可以理解为下图：

在这里插入图片描述

4.5 message.proto

同理于proto.proto，通过该文件定义消息队列中的核心部分，用于网络传输时的序列化与反序列化。

// 定义交换类型枚举
enum ExchangeType {
    UNKNOWTYPE = 0; // 未知类型
    DIRECT = 1; // 直连交换类型
    FANOUT = 2; // 扇出交换类型
    TOPIC = 3; // 主题交换类型
};

// 定义投递模式枚举
enum DeliveryMode {
    UNKNOWMODE = 0; // 未知模式
    UNDURABLE = 1; // 非持久模式
    DURABLE = 2; // 持久模式
};

// 定义基础属性消息
message BasicProperties {
    string id = 1; // 消息ID
    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; // 偏移量
    uint32 length = 3; // 长度
};

5. 部分需求设计

5.1 网络通信协议设计

需求确认

这个章节我们考虑客⼾端和服务器之间的通**信⽅式。回顾MQ的交互模型：

在这里插入图片描述

整个过程中，⽣产者和消费者都是客⼾端，它们都需要通过⽹络和Broker Server进⾏通信。具体通信的过程使⽤Muduo库来实现，使⽤TCP作为通信的底层协议，同时在这个基础上⾃定义应⽤层协议，完成客⼾端对服务器功能的远端调⽤。我们要实现的远端调⽤接⼝包括：

创建 Channel
关闭 Channel
创建 Exchange
删除 Exchange
创建 Queue
删除 Queue
创建 Binding
删除 Binding
发送 Message
订阅 Message
发送 Ack
返回 Message (服务器 -> 客户端)

5.2 设计应用层协议

使⽤⼆进制的⽅式设计应⽤层协议。因为MQMessage的消息体是使⽤Protobuf进⾏序列化的，本⾝是按照⼆进制存储的，所以不太适合⽤json等⽂本格式来定义协议

在这里插入图片描述

len：4个字节，表示整个报文的长度。
nameLen：4个字节，表示 typeName 数组的长度。
typeName：字节数组，占 nameLen 个字节，表示请求/响应报文的类型名，用于将消息分发到对应的远端接口调用中。
protobufData：字节数组，占 len - nameLen - 8 个字节，表示请求/响应参数数据，通过 Protobuf 序列化后的二进制。
checkSum：4个字节，表示整个消息的校验和，用于校验请求/响应报文的完整性。

6. 服务器模块实现

MQ Broker Server

在这里插入图片描述

BrokerServer模块是对整体服务器所有模块的整合，接收客户端的请求，并提供服务。

基于前边实现的简单的翻译服务器代码，进⾏改造，只需要实现服务器内部提供服务的各个业务接⼝即可。

在各个业务处理函数中，也⽐较简单，创建信道后，每次请求过来后，找到请求对应的信道句柄，通过句柄调⽤前边封装好的处理接⼝进⾏请求处理，最终返回处理结果

根据上图，总体的Broker服务器需要依靠很多内容，即模块划分部分的子模块：

在这里插入图片描述
对于其他文件，在需求分析与模块划分部分详细的解释了需要实现的功能，在源码处查看；

对于 mq_broker.hpp：在代码实现上，BrokerServer 应该有以下成员变量：

_server：Muduo库提供的一个通用TCP服务器，我们可以封装这个服务器进行TCP通信。
_baseloop：主事件循环器，用于响应IO事件和定时器事件，主loop主要是为了响应监听描述符的IO事件。
_codec：一个protobuf编解码器，我们在TCP服务器上设计了一个应用层协议，这个编解码器主要负责实现应用层协议的解析和封装，下边具体讲解。
_dispatcher：一个消息分发器，当Socket接收到一个报文消息后，我们需要按照消息的类型，即上面提到的 typeName，进行消息分发，会将不同类型的消息分发到相应的处理函数中，下边具体讲解。
_consumer：服务器中的消费者信息管理句柄。
_threadpool：异步工作线程池，主要用于队列消息的推送工作。
_connections：连接管理句柄，管理当前服务器上的所有已经建立的通信连接。
_virtual_host：服务器持有的虚拟主机。队列、交换机、绑定、消息等数据都是通过虚拟主机管理。

BrokerServer 通过给分发器绑定不同的回调函数，而函数

typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
BrokerServer(int port, const std::string& basedir) 
: _server(&_loop, muduo::net::InetAddress("0.0.0.0", port), "BrokerServer", muduo::net::TcpServer::kReusePort),
/*1*/_dispatcher(std::bind(&BrokerServer::onUnknown, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
/*2*/_codec_ptr(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),\
/*3*/_virtual_host(std::make_shared<VirtualHost>(HOSTNAME, basedir, basedir + DBFILE)),
/*4*/_connection_manager(std::make_shared<ConnectionManager>()),
/*5*/_consumer_manager(std::make_shared<ConsumerManager>()),
/*6*/_threadpool(std::make_shared<ThreadPool>())
{
    QueueMap qm = _virtual_host->allQueues();
    for(auto& q : qm) {
        _consumer_manager->initQueueConsumer(q.first);
    }

    // 注册业务请求处理函数
    /*openChannelRequest*/_dispatcher.registerMessageCallback<aiyimu::openChannelRequest>(std::bind(&BrokerServer::onOpenChannel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    /*closeChannelRequest*/_dispatcher.registerMessageCallback<aiyimu::closeChannelRequest>(std::bind(&BrokerServer::onCloseChannel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    /*declareExchangeRequest*/_dispatcher.registerMessageCallback<aiyimu::declareExchangeRequest>(std::bind(&BrokerServer::onDeclareExchange, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    /*deleteExchangeRequest*/_dispatcher.registerMessageCallback<aiyimu::deleteExchangeRequest>(std::bind(&BrokerServer::onDeleteExchange, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    /*declareQueueRequest*/_dispatcher.registerMessageCallback<aiyimu::declareQueueRequest>(std::bind(&BrokerServer::onDeclareQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    /*deleteQueueRequest*/_dispatcher.registerMessageCallback<aiyimu::deleteQueueRequest>(std::bind(&BrokerServer::onDeleteQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    /*queueBindRequest*/_dispatcher.registerMessageCallback<aiyimu::queueBindRequest>(std::bind(&BrokerServer::onQueueBind, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    /*queueUnbindRequest*/_dispatcher.registerMessageCallback<aiyimu::queueUnBindRequest>(std::bind(&BrokerServer::onQueueUnbind, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    /*basicPublishRequest*/_dispatcher.registerMessageCallback<aiyimu::basicPublishRequest>(std::bind(&BrokerServer::onBasicPublish, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    /*basicAckRequest*/_dispatcher.registerMessageCallback<aiyimu::basicAckRequest>(std::bind(&BrokerServer::onBasicAck, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    /*basicConsumeRequest*/_dispatcher.registerMessageCallback<aiyimu::basicConsumeRequest>(std::bind(&BrokerServer::onBasicConsume, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    /*basicCancelRequest*/_dispatcher.registerMessageCallback<aiyimu::basicCancelRequest>(std::bind(&BrokerServer::onBasicCancel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

    _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec_ptr.get(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    _server.setConnectionCallback(std::bind(&BrokerServer::onConnection, this, std::placeholders::_1));
}

void start()
{
    _server.start();
    _loop.loop();
}

7. 客户端模块实现

在这里插入图片描述

在RabbitMQ中，提供服务的是信道，因此在客户端的实现中，弱化了Client客户端的概念，也就是说在RabbitMQ中并不会向用户展示网络通信的概念出来，而是以一种提供服务的形式来体现。

实现思想类似于普通的功能接口封装，一个接口实现一个功能，接口内部完成向客户端请求的过程，但是对外并不需要体现出客户端与服务端通信的概念，用户需要什么服务就调用什么接口就行。

基于以上的思想，客户端的实现共分为四大模块：

订阅者模块：
- 并不直接对⽤户展⽰，在客户端体现的作⽤就是对⻆⾊的描述，表⽰这是⼀个消费者；
信道模块
- 直接⾯向⽤户，内部包含多个向外提供的服务接口，⽤户需要什么服务，调⽤对应接口即可。
- 其包含交换机声明/删除，队列声明/删除，绑定/解绑，消息发布/确认，订阅/解除订阅等服务。
连接模块
- 体现⽹络通信概念，⽤于打开/关闭信道。
异步线程模块

尽管客户端部分并不对外体现网络通信的概念，但本质上内部还是包含有网络通信的，因此既然有网络通信，那么就必须包含有一个网络通信IO事件监控线程模块，用于进行客户端连接的IO事件监控，以便于在事件触发后进行IO操作。
在客户端部分存在一个情况，当一个信道作为消费者而存在的时候，服务端会向信道推送消息，而用户这边需要对收到的消息进行不同的业务处理，而这个消息的处理需要一个异步的工作线程池来完成。
因此异步线程模块包含两个部分：
- 客户端连接的IO事件监控线程
- 推送过来的消息异步处理线程

基于以上模块，实现⼀个客户端的流程如下：

实例化异步线程对象
实例化连接对象
通过连接对象，创建信道
根据信道获取⾃⼰所需服务
关闭信道
关闭连接

① 订阅者模块

对于该模块，与服务端并无太大差别，客⼾端这边虽然订阅者的存在感微弱了很多，但是还是有的，当进行队列消息订阅的时候，会伴随着⼀个订阅者对象的创建。

该文件主要定义了一个消费者对象以及相关的回调函数，当创建消费者时使用。

namespace mq
{
    using ConsumerCallback = std::function<void(const std::string, const BasicProperties* bp, const std::string)>;

    struct Consumer 
    {
        using ptr = std::shared_ptr<Consumer>;
        std::string tag; // 消费者标识
        std::string qname; // 订阅的队列名称
        bool auto_ack; // 自动确认标志
        ConsumerCallback callback; // 消息处理回调函数

        Consumer() {
            DLOG("new consumer: %p", this);
        }

        Consumer(const std::string& consumer_tag, const std::string& queue_name, bool ack_flag, ConsumerCallback cb)
        : tag(consumer_tag), qname(queue_name), auto_ack(ack_flag), callback(cb)
        {
            DLOG("new consumer: %p", this);
        }

        ~Consumer() {
            DLOG("delete consumer: %p", this);
        }
    }; 
};

② 信道管理模块

客户端的信道功能与服务端几乎一致。无论是客户端的信道还是服务端的信道，都是为了向用户提供具体服务而存在的。只是服务端的信道是为了响应客户端的请求而提供服务，而客户端的信道则是为了允许用户向服务端发送请求并获取相应服务。

下面的代码中，对与Channel以创建信道举例：创建请求后，向服务器发送创建信道的请求，随后等待响应；

class Channel
{
public:
    using ptr = std::shared_ptr<Channel>;
    Channel(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec):
    _cid(UUIDHelper::uuid()), _conn(conn), _codec_ptr(codec) {}

    ~Channel() { basicCancel(); }
        
    std::string cid() { return _cid; }

    // 创建信道
    bool openChannel()
    {
        std::string rid = UUIDHelper::uuid();
        std::cout << "rid: " << rid << std::endl;
        // 创建请求
        openChannelRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        // 发送请求

        _codec_ptr->send(_conn, req);
        // 等待相应
        basicCommonResponsePtr resp = waitResponse(rid);

        return resp->ok();
    }

basicCommonResponsePtr waitResponse(const std::string& rid)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _cv.wait(lock, [&rid, this]() {
            if (_basic_resp.find(rid) == _basic_resp.end()) {
                // DLOG("_basic_resp中不存在rid: %s", rid.c_str());
                return false;
            }
            // DLOG("_basic_resp中存在rid: %s", rid.c_str());
            return true;
        });

        if (_basic_resp.find(rid) == _basic_resp.end()) {
            DLOG("等待响应超时");
        }

        basicCommonResponsePtr resp = _basic_resp[rid];
        _basic_resp.erase(rid);

        if (resp == nullptr) {
            DLOG("响应为空");
        }

        return resp;
    }

	// ... 其余函数

private:
        std::string _cid;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec_ptr;
        Consumer::ptr _consumer;
        std::mutex _mutex;
        std::condition_variable _cv;
        std::unordered_map<std::string, basicCommonResponsePtr> _basic_resp;
};

class ChannelManager
{
public:
    using ptr = std::shared_ptr<ChannelManager>;
    ChannelManager() {}
    Channel::ptr createChannel(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        auto channel = std::make_shared<Channel>(conn, codec);
        _channels.insert(std::make_pair(channel->cid(), channel));
        return channel;
    }

    void removeChannel(const std::string& cid)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _channels.erase(cid);
    }

    Channel::ptr getChannel(const std::string& cid)
    {
        // DLOG("获取信道: %s", cid.c_str());
        std::unique_lock<std::mutex> lock(_mutex);
        auto it = _channels.find(cid);
        if(it == _channels.end()) {
            DLOG("信道不存在: %s", cid.c_str());
            return Channel::ptr();
        }

        return it->second;
    }

private:
    std::unordered_map<std::string, Channel::ptr> _channels;
    std::mutex _mutex;  
};

③ 异步工作线程实现

客⼾端这边存在两个异步⼯作线程，

⼀个是muduo库中客⼾端连接的异步循环线程EventLoopThread，
⼀个是当收到消息后进⾏异步处理的⼯作线程池

这两项都不是以连接为单元进⾏创建的，⽽是创建后，可以⽤以多个连接中，因此单独进⾏封装。

AsyncWorker类通过这些成员来处理异步任务，利用事件循环和线程池机制提高操作效率。

namespace mq
{
    class AsyncWorker
    {
    public:
        using ptr = std::shared_ptr<AsyncWorker>;
        muduo::net::EventLoopThread _loopthread;
        ThreadPool _threadpool; // 线程池
    };
}

解释这个代码的大制作有

④ 连接管理模块

在客户端这边，RabbitMQ 弱化了客户端的概念，因为用户所需的服务都是通过信道来提供的。操作流程转换为先创建连接，通过连接创建信道，再通过信道提供服务。该模块同样是针对muduo库客户端连接的二次封装，向用户提供创建信道的接口。创建信道后，可以通过信道来获取指定服务。

namespace mq
{
    class Connection
    {
    public:
        using ptr = std::shared_ptr<Connection>;
        Connection(const std::string& server_ip, int server_port, const AsyncWorker::ptr& worker) 
        : _latch(1), _client(worker->_loopthread.startLoop(), muduo::net::InetAddress(server_ip, server_port), "client"), 
            _dispatcher(std::bind(&Connection::onUnknownMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
            _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
            _worker(worker), 
            _channel_manager(std::make_shared<ChannelManager>()) 
        {
            _dispatcher.registerMessageCallback<basicCommonResponse>(std::bind(&Connection::basicResponse, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
                
            _dispatcher.registerMessageCallback<basicConsumeResponse>(std::bind(&Connection::consumeResponse, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(),
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _client.setConnectionCallback(std::bind(&Connection::onConnection, this, std::placeholders::_1));      

            // 连接客户端
            _client.connect();
            _latch.wait();//阻塞等待，直到连接建立成功
        }

        // 打开信道
        Channel::ptr openChannel() {
            Channel::ptr channel = _channel_manager->createChannel(_conn, _codec);
            bool ret = channel->openChannel();
            if(!ret) {
                DLOG("打开信道失败");
                return Channel::ptr();
            }
            return channel;
        }

        // 关闭信道
        void closeChannel(Channel::ptr& channel) {
            channel->closeChannel();
            _channel_manager->removeChannel(channel->cid());
        }

    private:
        // 基本相应
        void basicResponse(const muduo::net::TcpConnectionPtr& conn, const basicCommonResponsePtr& msg, muduo::Timestamp) {
            // 1. 寻找信道
            Channel::ptr channel = _channel_manager->getChannel(msg->cid());
            if(channel.get() == nullptr) {
                DLOG("信道信息为空");
                return;
            }

            // 2. 将得到的响应对象添加到信道的基础相应
            channel->pushBasicResponse(msg);
        }

        void consumeResponse(const muduo::net::TcpConnectionPtr& conn, const basicConsumeResponsePtr& msg, muduo::Timestamp) {
            // 1. 寻找信道
            Channel::ptr channel = _channel_manager->getChannel(msg->cid());
            if(channel.get() == nullptr) {
                DLOG("信道信息为空");
                return;
            }

            // 2. 封装异步任务，放入线程池
            _worker->_threadpool.push([channel, msg](){
                channel->consume(msg);
            });
        }

        void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn, const MessagePtr& msg, muduo::Timestamp) {
            LOG_INFO << "收到未知消息: " << msg->DebugString();
            conn->shutdown(); // 关闭连接
        }

        void onConnection(const muduo::net::TcpConnectionPtr& conn) {
            if(conn->connected()) {
                _latch.countDown(); // 唤醒主线程的阻塞
                _conn = conn;
            } else {
                LOG_INFO << "连接断开";
                _conn.reset();
            }
        }

    private:
        muduo::CountDownLatch _latch; // 实现同步
        muduo::net::TcpConnectionPtr _conn; // 客户端对应的连接
        muduo::net::TcpClient _client; // 客户端
        ProtobufDispatcher _dispatcher; // 消息分发器
        ProtobufCodecPtr _codec; // 协议处理器

        AsyncWorker::ptr _worker; // 工作线程
        ChannelManager::ptr _channel_manager; // 频道管理器
    };
}

⑤ 生产者客户端

生产者客户端通过以下步骤，完成连接到服务器并生产消息：

实例化异步工作线程对象
实例化连接对象
创建信道
通过信道：
- 声明交换机、队列、绑定交换机与队列
- 生产消息

int main()
{
    // 1. 实例化 异步工作线程 对象
    aiyimu::AsyncWorker::ptr worker = std::make_shared<aiyimu::AsyncWorker>();
    // 2. 实例化 连接 对象
    aiyimu::Connection::ptr conn = std::make_shared<aiyimu::Connection>("127.0.0.1", 8081, worker);
    // 3. 通过连接 创建 信道
    aiyimu::Channel::ptr channel = conn->openChannel();
    // 4. 通过信道服务 完成操作
    // Ⅰ声明交换机，类型为广播
    google::protobuf::Map<std::string, std::string> args;
    channel->declareExchange("exchange1", aiyimu::ExchangeType::TOPIC, true, false, args);

    // Ⅱ声明n个队列
    channel->declareQueue("queue1", true, false, false, args);

    // Ⅲ绑定队列到交换机
    channel->queueBind("exchange1", "queue1", "news.music.#");
    // 5. 循环 发布消息 给交换机
    for(int i = 0; i < 10; ++i) 
    {
        aiyimu::BasicProperties bp;
        bp.set_id(aiyimu::UUIDHelper::uuid());
        bp.set_delivery_mode(aiyimu::DeliveryMode::DURABLE);
        bp.set_routing_key("news.music.pop");
        channel->basicPublish("exchange1", &bp, "hello World-" + std::to_string(i));
    }
   

    // 6. 关闭信道
    conn->closeChannel(channel);

    return 0;
}

⑥ 消费者客户端

客户端通过以下步骤，完成连接到服务器并订阅消息：

实例化异步工作线程对象
实例化连接对象
创建信道
通过信道：
- 声明交换机、队列、绑定交换机与队列
- 消费信息

#include "mq_connection.hpp"

void cb(aiyimu::Channel::ptr& channel, const std::string& consumer_tag,
    const aiyimu::BasicProperties* bp, const std::string& body)
{
    std::cout << "消费者 " << consumer_tag << " 消费了消息: " << body << std::endl;
    if (bp != nullptr) {
        channel->basicAck(bp->id());
    } else {
        std::cerr << "Error: BasicProperties is null" << std::endl;
    }
}

int main(int argc, char* argv[])
{
    if(argc != 2) {
        std::cout << "Usage: ./consumer_client queue_name\n";
        return -1; 
    }

    // 1. 实例化 异步工作线程 对象
    aiyimu::AsyncWorker::ptr worker = std::make_shared<aiyimu::AsyncWorker>();
    // 2. 实例化 连接 对象
    aiyimu::Connection::ptr conn = std::make_shared<aiyimu::Connection>("127.0.0.1", 8081, worker);
    // 3. 通过连接 创建 信道
    aiyimu::Channel::ptr channel = conn->openChannel();
    // 4. 通过信道 完成 需求
    // Ⅰ声明交换机exchange，类型为广播
    google::protobuf::Map<std::string, std::string> args;
    channel->declareExchange("exchange1", aiyimu::ExchangeType::TOPIC, false, true, args);
    // Ⅱ声明队列queue
    channel->declareQueue("queue1", true, false, false, args);
    
    // Ⅲ绑定队列到交换机
    channel->queueBind("exchange1", "queue1", "news.music.#");
    
    DLOG("消费消息");
    auto functor = std::bind(cb, channel, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
    channel->basicConsume("consumer1", argv[1], false, functor);

    while(1) {
        std::this_thread::sleep_for(std::chrono::seconds(3));
    }

    
    conn->closeChannel(channel);

    return 0;
}