C++基于protobuf实现仿RabbitMQ消息队列---服务器模块认识2
项目git链接：mq/mqdemo/muduo/protobuf/protobuf_client.cpp · 耀空/项目mq - 码云 - 开源中国
使用技术简介：
开发主语言：C++
序列化框架：Protobuf 二进制序列化
网络通信：自定义应用层协议 + muduo 库（对 tcp 长连接的封装、并且使用 epoll 的事件驱动模式，实现高并发服务器与客户端）
源数据信息数据库： SQLite3
单元测试框架： Gtest
- 首先，我们要知道我们这个项目是如何运行的，然后才能划分不同的模块进行编写；根据上一篇项目设计中我们就可以知道了具体要分为哪几个模块；现在要把他们设计出，还要知道它们的功能有那些，要管理什么资源，模块之间的连续；这样之后就可以在server.hpp直接把他们整合到一起；
- 当然每个模块写完之后都有进行测试，这里使用gtest的功能；所有测试见：mq/mqtest · 耀空/项目mq - 码云 - 开源中国
队列消息
认识：
因为消息数据需要在网络中进行传输，因此消息的类型定义使用 protobuf 进行，因为 protobuf 中自带了序列化和反序列化功能，因此操作起来会简便一些。
注意：
消息的存储并没有使用数据库，因为消息长度通常不定，且有些消息可能会非常庞大，因此并不适合存储在数据库中，因此我们的处理方式（包括 RabbitMQ）是直接将消息存储在文件中进行管理（sqlite），而内存中管理的消息只需要记录好自己在文件中的所在位置和长度即可。
消息的结构如下：
以队列为单元进行管理
为了便于管理，消息管理以队列为单元进行管理，因此每个队列都会有自己独立的数 据存储文件；当然也可以信道为单元进行管理；（这个项目是以队列为单元进行管理）
区别就是
- 以信道为单元：一个信道关闭的时候所以关联的消费者都要删除；
- 以队列为单元：一个队列收到一个消息，需要找到订阅了队列消息的消费者进行消息推送
消息数据管理
消息数据管理（mq_msg.proto）
因为消息数据需要在网络中进行传输，因此消息的类型定义使用 protobuf 进行，因为 protobuf 中自带了序列化和反序列化功能，因此操作起来会简便一些。
所以要先在protobuf定义 消息属性；
消息信息
- 消息属性:
- ID: 消息的唯一标识
- 持久化标志：表示是否对消息进行持久化（还取决于队列的持久化标志）
- routing_key: 决定了当前消息要发布的队列（消息发布到交换机后，根据绑定队列的 binding_key 决定是否发布到指定队列）
- 消息主体：消息内容
- 有效标志
--- 以下是服务端为了管理所添加的信息
- 存储偏移量：消息以队列为单元存储在文件中，这个偏移量，是当前消息相对于文件起始位置的偏移量
- 消息长度：从偏移量位置取出指定长度的消息（解决粘包问题）是否
- 有效标志：标识当前消息是否已经被删除（删除一条消息，并不会每次直接将后边的数据拷贝到前边，而只是重置了标志，当一个文件中，有效消息占据总消息比例不到 50%，且数据量超过 2000，则进行垃圾回收，重新整理文件数据...）
最好还是参考代码看；
代码见
消息数据内容管理代码
https://gitee.com/yaokong123/project-mq/blob/master/mq/mqcommon/mq_msg.proto
消息的管理
管理方式：以队列为单元进行管理（因为消息的所有操作都是以队列为单元的）
上面都已经说明了
管理的数据：
- 消息链表：保存所有的待推送消息
- 待确认消息 hash：消息推送给客户端后，会等待客户端进行消息确认，收到确认后，才会真正删除消息
- 持久化消息 hash：假设消息都会进行持久化存储，操作过程中会存在垃圾回收操作，但是垃圾回收会改变消息的存储位置。但是内存中的消息也会存储消息的实际存储位置，垃圾回收后就不一致了因此每次垃圾回收后，都需要用新的位置，去更新持久化消息的信息垃圾回收：将有效消息读取出来，然后重新截断文件，将消息连续写入文件中（文件中都是有效消息）
- 持久化的有效消息数量
- 持久化的总的消息数量：决定了什么时候进行垃圾回收。
- 持久化数据管理句柄
管理操作：
- 向队列新增消息
- 获取队首消息：获取消息后，就会将消息从待推送消息链表删除（不再是待发送消息，而是待确认消息），加入到待确认消息中
- 对消息进行确认：从待确认消息中移除消息，并进行持久化数据的删除
- 恢复队列历史消息：主要是在构造函数中进行（只有在重启的时候才会进行）
- 垃圾回收（消息持久化子模块完成）：持久化文件中有效消息比例小于 50%，且总消息数量超过 200 进行垃圾回收。
- 删除队列相关消息文件：当一个队列被删除了，那它的消息也就没有存在的意义了。
代码见：
mq/mqsever/mq_message.hpp · 耀空/项目mq - 码云 - 开源中国https://gitee.com/yaokong123/project-mq/blob/master/mq/mqsever/mq_message.hpp
队列消息管理
管理的数据：
这不必多说
- 持久化管理队列消息的路径
- 队列消息局部与其名称的映射
管理的操作
- 初始化队列消息结构：初始化新建队列的消息管理结构，并创建消息存储文件
- 移除队列消息结构：删除队列的消息管理结构，以及消息存储文件
- 向队列新增消息：向指定队列新增消息
- 对队列消息进行确认：确认指定队列待确认消息（删除）
- 获取指定队列队首消息：按序获得消息，让消息得以有序消费
- 获得指定队列的消息数量；
代码见：
MessageManager类，队列消息管理https://gitee.com/yaokong123/project-mq/blob/master/mq/mqsever/mq_message.hpp
虚拟机管理
（VirtualHost）
虚拟机模块是对上述交换机 + 队列 + 绑定 + 消息 的整合，并基于数据之间的关联关系进行联合操作
流程图：
从这个broker服务器的简单流程图，数据在这三大模块之间应该是如何有联系的；
要管理的数据：
就是要整合的模块的数据管理句柄
- 交换机数据管理句柄
- 队列数据管理句柄
- 绑定信息数据管理句柄
- 消息数据管理句柄
要管理的操作：
- 声明 / 删除交换机：在删除交换机的时候要删除相关的绑定信息
- 声明 / 删除队列：在删除队列的时候，要删除相关的绑定信息以及消息数据
- 队列的绑定 / 解除绑定：绑定的时候，必须交换机和队列是存在的
- 获取指定队列的消息：
- 对指定队列的指定消息进行确认：
- 获取交换机相关的所有绑定信息：一条消息要发布给指定交换机的时候，交换机获取所有绑定信息，来确定消息要发布到哪个队列。
代码见：
mq/mqsever/mq_host.hpp · 耀空/项目mq - 码云 - 开源中国https://gitee.com/yaokong123/project-mq/blob/master/mq/mqsever/mq_host.hpp
路由匹配模块
认识
客户端将消息发布到指定的交换机，交换机这时候要考虑这条数据该放入到哪些与自 己绑定的队列中，而这个考量是通过交换机类型以及匹配规则来决定的
路由匹配模块：决定了一条消息是否能够发布到指定的队
在每个队列跟交换机的绑定信息中，都有一个 binding_key：这是队列发布的匹配规则
在每条要发布的消息中，都有一个 routing_key：是消息的发布规则
要实现的交换机有三种交换类型：直接，广播，主题
- 广播：直接将消息发布给交换机的所有绑定队列
- 直接：routing_key 与 binding_key 完全一致则匹配成功
- 主题：binding_key 中是匹配规则 news.music.#，routing_key 是消息规则 news.music.pop，匹配成功才能发布
路由匹配模块本质上来说，没有要管理的数据，只有向外提供的路由匹配操作：
- 提供一个判断 routing_key 与 binding_key 是否能够匹配成功的接口
- 判断 routing_key 是否符合规定：格式约定：只能由 数字，字母，_ . 构成
- 判断 binding_key 是否符合规定：格式越是：只能由 数字，字母，_ . # * 构成
广播，直接交换类型很容易实现，主题如何实现呢？
主题如何实现呢？
这其实就是一个算法题，
定义一个二维数组来标记每次匹配的结果，通过最终数组末尾位置的结果来查看是否 整体匹配成功
这个是一个动态规划
具体思路就是：使用 routing_key 中的每个单词，与 binding_key 中的单词进行逐个匹配，根据匹配结果来标记数组内容，最终以数组中的末尾标记来确定是否匹配成功。
- dp[i][j] 表示routing_key第i个单词与binding_key第j个单词 是否匹配
- 对于一般的单词，这个动态规划是否能继续和 上组个单词匹配的结果有关，也就是dp[i][j] = dp[i-1][j-1]
- 对于#通配符的特殊，是否能继续，不仅和[i-1][j-1]有关系，还和[i][j-1]有关系，如图
- 动态规划的初始化：第0组的第一个因该是1，一定是能传承到下一组匹配的；
binding_key = "#.aaa"; routing_key = "aaa"
可以按照这个思路自己匹配一次，下图是一个例子;
具体代码见：
mq/mqsever/mq_route.hpp · 耀空/项目mq - 码云 - 开源中国https://gitee.com/yaokong123/project-mq/blob/master/mq/mqsever/mq_route.hpp
本文地址：https://www.yitenyun.com/4852.html
