【项目日记】仿RabbitMQ实现消息队列---模块设计

人工智能
2025-08-28 06:24:01

你要的答案不在书本里，也不能靠别人来解决，除非你想一辈子当小孩。你必须在自我内部找到答案，感受到该做的正确事情。 --- 《献给阿尔吉侬的花束》---

仿RabbitMQ实现消息队列 1 数据管理模块1.1 交换机数据管理模块1.2 队列数据管理模块1.3 绑定数据管理模块1.4 消息数据管理模块1.5 虚拟机数据管理模块 2 功能模块2.1 路由匹配模块2.2 消费者管理模块2.3 信道管理模块2.4 连接管理模块 3 服务器模块4 客户端模块5 模块关系图

1 数据管理模块 1.1 交换机数据管理模块

交换机需要管理的成员：描述了一个交换机应该有什么数据

交换机名称：唯一标识交换机类型：决定消息的转发方式，每个队列绑定中有个binding_key ，每条消息中有个routing_key : 直接交换Direct：binding_key 与routing_key 一致，将消息放入队列中。广播交换Fanout：将消息放入交换机绑定的所有队列中。主题交换Topic：routing_key与多个绑定队列的binding_key有个匹配规则，匹配成功则放入持久化标志：决定了当前交换机信息是否需要持久化储存。自动删除标志：关联了当前交换机的所有客户端都退出了，是否需要自动删除交换机。交换机其他参数

对交换机的管理操作

创建交换机：本质上是需要声明，强断言的思想：有就直接用，没有就创建删除交换机：每个交换机都会绑定一个或多个队列（意味着会有一个或多个绑定信息），因此删除交换机需要删除相关绑定信息。获取指定名称交换机：获取当前交换机数量： 1.2 队列数据管理模块

队列要管理的数据：

队列名称：队列的唯一标识持久化储存标志：决定了是否将队列信息持久化的储存起来，决定了重启后，这个队列还是是否存在，也决定了队列中的消息是否需要持久化。是否独占标志：独占指的是只有当前客户端自己可以订阅队列消息自动删除标志：当订阅了当前队列的所有客户端退出后，是否删除队列。其他参数

对队列进行的管理操作依旧是增删查：

创建队列删除队列获取指定队列信息获取队列数量获取所有队列名称：当系统重启后，需要重新加载数据，加载历史消息（消息以队列为单元储存在文件中），而加载消息需要知道队列名称！因为后边消息储存的时候，储存文件已队列名称进行的取名。

一个队列如果持久化标志为false，则意味着重启后队列就没了，也就没有客户端能够订阅队列的信息，因此这个队列的消息如果持久化储存了，是没有意义的。因此通常一个队列的持久化标志是false，那么它的消息也就不需要持久化处理！

1.3 绑定数据管理模块

描述一下将哪个队列与哪个交换机绑定在一起。管理的数据：

交换机名称队列名称binding_key：绑定密钥–描述了在交换机的主题交换&直接交换的消息发布匹配规则。由数字，字符，_，#，.。

对绑定数据管理操作：

添加绑定解除绑定获取交换机相关的所有绑定信息：方便删除交换机时删除相关绑定信息。当消息发布到交换机，交换机得通过这些信息来将消息发布到指定队列获取队列相关的所有绑定信息：方便删除队列时删除相关的绑定信息获取绑定信息数量 1.4 消息数据管理模块

一条消息中应该包含哪些信息：

消息属性： ID：消息的唯一标识持久化标志：表示是否对该消息进行持久化routing_key：决定了当前消息要发布的队列（消息发布到交换机后，根据绑定队列的binding_key决定是否发布到指定队列）消息主体：消息的内容服务器为了管理所添加的信息储存偏移量：消息一队列为单元储存在文件中这个偏移量，是当前消息相对于文件起始位置的偏移量消息长度：从偏移量位置取出指定长度的消息（解决粘包问题）是否有效标志位：标识当前消息是否已经被删除（删除一条消息，并不会每次直接将后边的数据拷贝到前边，而只是重置了标志位。当一个文件中，有效消息占比不足50%时，且数据量超过2000，则进行垃圾回收，重新整理文件数据储存，将有效数据读取处理，然后重新截断文件，将消息连续的写入文件中！重启系统也只需要重新加载有效信息即可）

消息的管理方式：以队列为单元进行管理（因为消息的所有操作都是以队列为单元的）针对消息的管理操作：

消息链表：保存所有的待推送消息。待确认消息hash：消息推送给客户端，会等待客户端进行消息确认，收到确认后，才会真正删除消息！持久化消息hash：假设消息都会进行持久化储存，操作过程中会存在垃圾回收操作，但是垃圾回收会改变消息存储位置的！所以每次垃圾回收之后都要重新更新消息的属性！持久化的有效消息数量持久化的总消息数量：决定了什么时候进行垃圾回收！

对消息数据的管理操作：

向队列中新增消息获取队首消息：获取消息后，就会从消息从消息链表中删除（不再是待发送消息，而是待确认消息），加入到待确认消息中确认消息：从待确认消息中移除消息，并进行持久化数据的删除！恢复队列历史消息：主要是在构造函数中进行（只有在重启的时候才会进行）垃圾回收（消息持久化子模块完成）：持久化文件中有效消息比例小于50%，且总消息数量超过200进行垃圾处理删除队列相关消息文件：当一个队列被删除了，那他的消息也就没有存在的意义了！

队列消息管理：

初始化队列消息结构移除队列消息结构：在一个队列创建/删除时进行调用向队列新增消息对队列消息进行确认恢复队列历史消息 1.5 虚拟机数据管理模块

虚拟机数据管理是对：交换机+队列+绑定+消息的数据管理的集合需要管理的数据：

交换机数据管理句柄队列数据管理句柄绑定信息管理句柄消息数据管理句柄

需要进行的管理操作

声明/删除交换机：注意在删除交换机的时候要删除相关的绑定信息声明/删除队列：注意在删除队列的时候，要删除相关的绑定信息以及消息数据队列的绑定/解除绑定：注意绑定的时候，交换机与队列是存在的！获取指定队列的信息对指定队列的指定消息进行确认获取交换机相关的所有绑定信息：一条消息要发布给指定交换机的时候，交换机获取所有绑定信息，来确定消息要发布到哪个队列。 2 功能模块 2.1 路由匹配模块

路由匹配模块决定了一条消息是否能够发布到指定的队列中。在每个队列和交换机的绑定信息中都有一个binding_key ：这是队列发布的匹配规则。在每条要发布的消息中都有一个routing_key：这是消息的发布规则。交换机有三种交换类型：直接，广播，主题。

直接交换Direct：binding_key 与routing_key 一致，将消息放入队列中。广播交换Fanout：将消息放入交换机绑定的所有队列中。主题交换Topic：routing_key与多个绑定队列的binding_key有个匹配规则，匹配成功则放入

路由匹配模块本质上来说没有要管理的数据，只有向外提供的路由匹配操作：

提供一个判断routing_key与binding_key是否能够匹配成功的接口判断routing_key是否符合规定：格式约定：只能由数字，字母，_ . 构成判断binding_key是否符合规定：格式约定：只能由数字，字母，_ . # *构成 2.2 消费者管理模块

客户端有两种：发布消息客户端，订阅消息客户端。因此订阅了指定队列消息的客户端才是一个消费者！

消费者数据存在的意义：当指定队列有了消息之后，就需要将消息推送给这个消费者客户端（推送的时候就需要找到这个客户端相关的信息—连接）！

消费者信息：

消费者标识tag订阅队列名称：当当前队列有消息就会推送给当前客户端，以及当客户端收到消息，需要对指定队列的消息进行确认自动确认标志：自动确认—推送消息后直接删除消息不需要额外确认。手动确认—推送消息后，需要等到消息确认回复之后再去删除消息。消费处理回调函数指针：队列有一条消息之后，通过哪个函数进行处理（函数内部其实逻辑是固定的—向指定客户端推送消息）

消费者管理思想：

以队列为单元进行管理，每个消费者订阅的都是指定队列的消息，消费者对消息进行确认也是以队列进行确认。最关键的是：当指定队列中有消息了，必然是获取订阅了这个队列的消费者信息进行消息推送

队列消费者管理结构：

数据信息：消费者链表—保存当前队列的所有消费者信息（RR轮转思想，每次取出下一个消费者进行消息推送，一个消息只需要被一个客户端处理即可）管理操作：新增消费者RR轮转获取一个消费者删除消费者队列消费者数量是否为空

整体的管理操作：

初始化队列消费者结构删除队列消费者结构向直到队列添加消费者获取指定队列消费者删除指定队列消费者 2.3 信道管理模块

信道是网络通信中的一个概念，叫做通信通道。网络通信的时候，必然是通过网络通信连接来完成的，为了能够更加充分的利用资源，因此对通信连接又进行了一次细化，细化出了通信通道，对于用户来说，一个通信通道，就是进行网络通信的载体，而一个真正的通信连接，可以创造出多个通信通道！

每一个信道之间，在用户眼中是相互独立的，而本质的底层它们使用同一个通信连接进行网络通信! 因此，信道是用户眼中的一个通信通道。所以所有的网络通信服务都是由信道提供的！

信道提供的服务操作：

声明/删除交换机声明/删除队列绑定/解绑队列与交换机发布消息/订阅队列消息/取消队列订阅/队列消息确认

信道要管理的数据：

信道关联的虚拟机句柄信道管理的消费者句柄：当信道关闭的时候，所有关联的消费者订阅都要取消，相当于删除所有的相关消费者！工作线程池句柄：一旦信道进行了消息发布操作到指定队列之后，从指定队列获取一个消费者，对这条消息进行消费（也就是将这条消息推送给一个客户端的操作交给线程池来执行！）。并发每个信道都有一个线程池，而是整个服务器有一个线程池，大家所有的信道都是通过同一个线程池进行一步操作的！

对信道的管理操作

创建一个信道关闭一个信道获取指定信道句柄 2.4 连接管理模块

连接就是网络通信连接。在网络通信连接中我们使用muduo库来实现底层通信，muduo库中本身就有Connection连接的概念和对象类。但是在我们的连接中，还有一个上层通道信道的概念（muduo库中的信道是底层管理Socket通信的），因此我们的Connection是对muduo库的Connection的二次封装，形成我们自己所需的连接管理！

管理的数据；

muduo库的通信连接当前连接关联的信道管理句柄

连接提供的操作

创建信道关闭信道

管理的操作

新增连接关闭连接获取指定连接信息 3 服务器模块

这个模块是一个功能整合模块，整合以上所有模块。本质上并不提供实质的功能性操作，重要的是搭建网络通信服务器，实现与客户端网络通信，能够识别客户端请求，并提供客户端请求的处理服务。

管理信息：

虚拟机管理模块句柄：一个服务器有一个虚拟机，其他所有交换机，队列，绑定，消息的操作都是针对这个虚拟机进行的！消费者管理模块句柄：一个服务器有一个消费者管理模块通信连接管理模块与协议处理模块句柄：一个服务器有一套通信模块。工作线程池句柄：一个服务器有一个工作线程池，其他所有的信道操作都是这一个线程池muduo库通信所需元素 4 客户端模块

客户端模块包含几个模块：消费者管理模块，信道管理模块

消费者管理模块需要：

消费者标识订阅的队列名称自动确认标志消息回调函数函数指针：当消费者订阅了某一个队列的消息，这个队列有了这个消息之后，就会将消息推送给这个客户端，这时候收到了消息则使用回调函数进行处理，处理完毕后根据确认标志决定是否进行消息确认！

消费者管理模块管理的操作：增删查

信道管理模块：所有提供的操作与服务端雷同，因为客户端给用户要提供什么服务，服务器就要给客户端通过什么服务！

信道管理模块管理的信息有：

消费者管理模块：每个信道都要有自己的相关消费者线程池句柄：对推送过来的消息进行回调处理，处理过程通过工作线程来进行信道关联的连接。

信道提供的服务：

声明/删除交换机声明/删除队列绑定/解绑队列与交换机发布消息/确认消息订阅队列消息/取消订阅队列消息创建/关闭信道

信道的管理：增删查

连接管理模块：客户端连接的管理，本质上是对客户端TcpClient的二次封装和管理。面对用户来说，不需要有客户端的概念，连接对应用户来说就是客户端，通过连接创建信道，通过信道完成自己的所需服务。因此当前客户端这边的连接，对于用户来说就是一个资源的载体！

管理的资源：工作线程池，连接关联的信道管理句柄管理的操作：