“仿RabbitMQ实现消息队列”---整体架构与模块说明

顾得泉:个人主页

个人专栏:《Linux操作系统》 《C++从入门到精通》  《LeedCode刷题》

键盘敲烂,年薪百万!


一、概念性框架理解

我们主要实现的内容:

       1.Broker服务器:消息队列服务器(服务端

       2.消息发布客户端:向服务器发布消息

       3.消息订阅客户端:从服务器订阅消息

    broker服务器是我们最核心的部分,负责消息的存储和转发。

       而我们使用的AMQP(Advanced Message Queuing Protocol-高级消息队列协议,其中一个提供统一消息服务的应用层标准高级消息队列协议,为面向消息的中间件设计,使得遵从该规范的客户端应用和消息中间件服务器的全功能互操作成为可能)模型中,也就是消息中间件服务器Broker中,又存在以下概念:

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


二、服务端模块概要设计

一、交换机数据管理模块

1.要管理的数据:描述了一个交换机应该有什么数据

  1.交换机名称:唯一标识
  2.交换机类型:决定了消息的转发方式
       每个队列绑定中有个binding_key,每条消息中有个routing_key

       1.直接交换: binding_key与routing_key相同,则将消息放入队列

       2.广播交换:将消息放入交换机绑定的所有队列中
       3.主题交换: routing_key与多个绑定队列的binding_key有匹配规则,匹配成功了则放入

  3.持久化标志:决定了当前交换机信息是否需要持久化存储
  4.自动删除标志:指的是关联了当前交换机的所有客户端都退出了,是否要自动删除交换机

  5.交换机的其他参数:当前未使用。

2.对交换机的管理操作:

    1.创建交换机:本质上需要的是声明-------强断言的思想-有就OK,没有则创建的意思
    ⒉删除交换机:注意事项--每个交换机都会绑定一个或多个队列(意味着会有一个或多个绑定信息),因此删除交换机需要删除相关绑定信息

    3.获取指定名称交换机
    4.获取当前交换机数量

二、队列数据管理模块

1.要管理的数据:

    1.队列名称:唯一的标识
    2.持久化存储标志:决定了是否将队列信息持久化存储起来,决定了重启后,这个队列还是否存在

    3.是否独占标志:独占就指的是,只有当前客户端自己能够订阅队列消息
    4.自动删除标志:当订阅了当前队列的所有客户端退出后,是否删除队列((暂不考虑)。

    5.其他参数:(暂不考虑)

2.提供的管理操作(还就是增删查三个操作)

    1.创建队列
    2.删除队列
    3.获取指定队列信息

    4.获取队列数量
    5.获取所有队列名称

       当系统重启后,需要重新加载数据,加载历史消息(消息以队列为单元存储在文件中)
而加载消息需要知道队列名称,因为后边消息存储的时候,存储文件以队列名称进行的取名

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

三、绑定数据管理模块

管理的数据:

    1.交换机名称

    2.队列名称
    3. binding_key:
绑定密钥--描述了在交换机的主题交换&直接交换的消息发布匹配规则
       由数字,字符,_,#,.,*组成:

           binding_key: news.music.#      routing_key: news.sport.football

管理的操作:

    1.添加绑定

    2.解除绑定
    3.获取交换机相关的所有绑定信息:

       1.删除交换机的时候,要删除相关绑定信息
       2.当消息发布到交换机,交换机得通过这些信息来将消息发布到指定队列

    4.获取队列相关的所有绑定信息:
       删除队列的时候,要删除相关的绑定信息

    5.获取绑定信息数量

四、消息数据管理模块

1.消息信息:

消息属性:

    ID:消息的唯一标识
    持久化标志:表示是否对消息进行持久化(还取决于队列的持久化标志)
    routing_key:决定了当前消息要发布的队列(消息发布到交换机后,根据绑定队列的binding_key决定是否发布到指定队列)

消息主体:消息内容

    --以下是服务端为了管理所添加的信息
    存储偏移量:消息以队列为单元存储在文件中,这个偏移量,是当前消息相对于文件起始位置的偏移量

    消息长度:从偏移量位置取出指定长度的消息(解决粘包问题)
    是否有效标志∶标识当前消息是否已经被删除

       删除一条消息,并不会每次直接将后边的数据拷贝到前边,而只是重置了标志,当一个文件中,有效消息占据总消息比例不到50%,且数据量超过2000,则进行垃圾回收,重新整理文件数据存储*当系统重启,也只需要重新加载有效消息即可(相当于进行了一次垃圾回收)

2.消息的管理

管理方式:以队列为单元进行管理(因为消息的所有操作都是以队列为单元的)

管理数据:

    1.消息链表:保存所有的待推送消息
    2.待确认消息hash:消息推送给客户端后,会等待客户端进行消息确认,收到确认后,才会真正删除消息
    3.持久化消息hash:假设消息都会进行持久化存储,操作过程中会存在垃圾回收操作,但是垃圾回收会改变消息的存储位置。但是内存中的消息也会存储消息的实际存储位置,垃圾回收后就不一致了,因此每次垃圾回收后,都需要用新的位置,去更新持久化消息的信息。垃圾回收:将有效消息读取出来,然后重新截断文件,将消息连续写入文件中(文件中都是有效消息)

    4.持久化的有效消息数量
    5.持久化的总的消息数量:
决定了什么时候进行垃圾回收。

管理操作:

    1.向队列新增消息
    2.获取队首消息:
获取消息后,就会将消息从待推送消息链表删除(不再是待发送消息,而是待确认消息),加入到待确认消息中

    3.对消息进行确认:从待确认消息中移除消息,并进行持久化数据的删除
    4.恢复队列历史消息:主要是在构造函数中进行(只有在重启的时候才会进行)
    5.垃圾回收(消息持久化子模块完成)∶持久化文件中有效消息比例小于50%,且总消息数量超过200进行垃圾回收

    6.删除队列相关消息文件:当一个队列被删除了,那它的消息也就没有存在的意义了。

3.队列消息管理

    1.初始化队列消息结构
    2.移除队列消息结构:
在一个队列创建/删除的时候调用

    3.向队列新增消息
    4.对队列消息进行确认

    5.恢复队列历史消息

五、虚拟机数据管理模块

       对交换机+队列+绑定+消息数据管理的整合

要管理的数据:

    1.交换机数据管理句柄

    2.队列数据管理句柄

    3.绑定信息数据管理句柄

    4.消息数据管理句柄

要管理的操作:

    1.声明/删除交换机:注意---在删除交换机的时候要删除相关的绑定信息
    2.声明/删除队列:注意--在删除队列的时候,要删除相关的绑定信息以及消息数据

    3.队列的绑定/解除绑定:注意--绑定的时候,必须交换机和队列是存在的
    4.获取指定队列的消息
    5.对指定队列的指定消息进行确认
    6.获取交换机相关的所有绑定信息:
一条消息要发布给指定交换机的时候,交换机获取所有绑定信息,来确定消息要发布到哪个队列。

六、交换路由模块

       决定了一条消息是否能够发布到指定的队列

    在每个队列跟交换机的绑定信息中,都有一个binding_key:这是队列发布的匹配规则

    在每条要发布的消息中,都有一个routing_key:是消息的发布规则

    交换机有三种交换类型:直接,广播,主题

       广播:直接将消息发布给交换机的所有绑定队列

       直接: routing_key与binding_key完全一致则匹配成功

       主题: binding_key中是匹配规则news.music.#,routing_key是消息规则news.music.pop,匹配成功才能发布

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

    1.提供一个判断routing_key与binding_key是否能够匹配成功的接口
    2.判断routing_key是否符合规定:
       格式约定:只能由数字,字母,_﹒构成

    3.判断binding_key是否符合规定:
       格式约定∶只能由数字,字母,_.#*构成

七、消费者管理模块

    客户端有两种:发布消息,订阅消息
    因此订阅了指定队列消息的客户端才是一个消费者。
    消费者数据存在的意义:当指定队列有了消息以后,就需要将消息推送给这个消费者客户端(推送的时候就需要找到这个客户端相关的信息--连接)

消费者信息:

1.消费者标识--tag
⒉订阅队列名称:当当前队列有消息就会推送给这个客户端,以及当客户端收到消息,需要对指定队列的消息进行确认
3.自动确认标志:自动确认---推送消息后,直接删除消息不需要额外确认,手动确认---推送消息后,需要等到收到确认回复再去删除消息4.消费处理回调函数指针:队列有一条消息后,通过哪个函数进行处理(函数内部其实逻辑固定---向指定客户端推送消息)

消费者管理:

管理思想:以队列为单元进行管理

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

队列消费者管理结构:

    数据信息:消费者链表-…保存当前队列的所有消费者信息(RR轮转每次取出下一个消费者

进行消息推送--一条消息只需要被一个客户端处理即可)

    管理操作:

       1.新增消费者 2.RR轮转获取一个消费者 3.删除消费者 4.队列消费者数量 5.是否为空

管理操作:

    1.初始化队列消费者结构

    2.删除队列消费者结构

    3.向指定队列添加消费者

    4.获取指定队列消费者
    5.删除指定队列消费者

八、信道管理模块

       信道管理: Channel

    信道是网络通信中的一个概念,叫做通信通道。
    网络通信的时候,必然都是通过网络通信连接来完成的,为了能够更加充分的利用资源,因此对通信连接又进行了进一步的细化,细化出了通信通道。

    对于用户来说,一个通信通道,就是进行网络通信的载体,而一个真正的通信连接,可以创建出多个通信通道
    每一个信道之间,在用户的眼中是相互独立的,而在本质的底层它们使用同一个通信连接进行网络通信。
    因此,因为信道是用户眼中的一个通信通道,所以所有的网络通信服务都是由信道提供的。

信道提供的服务操作:

1.声明/删除交换机

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

信道要管理的数据:

    0.信道ID
    1.信道关联的虚拟机句柄

    2.信道关联的消费者句柄:当信道关闭的时候,所有关联的消费者订阅都要取消,相当于删除所有的相关消费者。

    3.工作线程池句柄:信道进行了消息发布到指定队列操作之后;从指定队列获取一个消费者,对这条消息进行消费,也就是将这条消息推送给一个客户端的操作交给线程池执行。并非每个信道都有一个线程池,而是整个服务器有一个线程池,大家所有的信道都是通过同一个线程池进行异步操作而已

信道的管理:

1.创建一个信道   2.关闭一个信道   3.获取指定信道句柄

九、连接管理模块

       概念:网络通信连接

    在网络通信模块中,我们使用muduo库来实现底层通信,muduo库中本身就有Connection连接的概念和对象类。但是我们的连接中,还有一个上层通信信道的概念,这个概念在muduo库中是没有的。
    因此,我们需要在用户的层面,对这个muduo库中的Connection连接进行二次封装。形成我们自己所需的连接管理。

管理数据:

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

连接提供的操作:

1.创建信道   2.关闭信道

管理的操作:

1.新增连接   2.关闭连接   3.获取指定连接信息


三、客户端模块概要设计

一、消费者管理模块

    1.消费者标识

    2.订阅的队列名称

    3.自动确认标志

    4.消息回调处理函数指针

       当当前消费者订阅了某一个队列的消息,这个队列有了消息后,就会将消息推送给这个客户端,这时候收到了消息则使用回调函数进行处理,处理完毕后根据确认标志决定是否进行消息确认。

       管理操作:增删查

二、信道管理模块

       所有提供的操作与服务端雷同,因为客户端给用户要提供什么服务,服务器就要给客户端提供什么服务。

管理信息:

    0.信道ID
    1.消费者管理句柄:
每个信道都有自己相关的消费者
    2.线程池句柄:对推送过来的消息进行回调处理,处理过程通过工作线程来进行

    3.信道关联的连接

信道提供的服务:

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

信道的管理:信道的增删查

三、连接管理模块

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

管理操作:

    1.连接服务器        2.创建信道        

    3.关闭信道        4.关闭连接

管理的资源:工作线程池,连接关联的信道管理句柄

四、异步工作池模块

    1.TcpClient模块需要一个EventLoopThread模块进行IO事件监控。

    2.收到推送消息后,需要对推送过来的消息进行处理,因此需要一个线程池来帮助我们完成消息处理的过程。

       将异步工作线程模块单独拎出来,原因是多个连接用一个EventLoopThread进行I0事件监控就够了,以及所有的推送消息处理也只需要有一个线程池就够了。

       并不需要每个连接都有一个EventLoop,也不需要每个信道的消息处理都有自己的线程池。


四、项目模块整体关系图


结语:关于项目本次的分享到这里就结束了,如果大家有什么问题,欢迎大家在评论区留言~~~ 

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/web/20012.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

连通块中点的数量-java

本次我们通过连通块中点的数量来加深我们对并查集的基本操作和原理,并且知道如何在并查集中添加附属信息。 目录 前言☀ 一、连通块中点的数量☀ 二、算法思路☀ 1.无向图🌙 2.在a b之间连一条边,a b可能相等🌙 3.询问a和b是否在一…

Apache POI对Excel进行读写操作

1、什么是Apache POI ​ Apache POI 是一个处理Miscrosoft Office各种文件格式的开源项目。简单来说就是,我们可以使用 POI 在 Java 程序中对Miscrosoft Office各种文件进行读写操作。一般情况下,POI 都是用于操作 Excel 文件。 Apache POI 的应用场景&…

【为什么 Google Chrome 打开网页有时极慢?尤其是国内网站,如知网等】

要通过知网搜一点资料,发现怎么都打不开。而且B站,知乎这些速度也变慢了!已经检查过确定不是网络的问题。 清空了记录,清空了已接受Cookie,清空了缓存内容……没用!!! 不断搜索&am…

注意力机制详解

引言 在阅读一篇文章时,我们的大脑并不平等地处理每一个字词,而是根据上下文自动筛选出核心信息进行深入理解。注意力机制正是借鉴了这一生物学灵感,使得机器学习模型能够动态地分配其“注意力”资源,针对不同的输入部分赋予不同…

大数据信用报告分析和评估有什么意义

大数据信用这个词在现在已经是很常见的了,只要是申贷的朋友对它就不陌生,在明面上的信用资质刚刚满足审核要求,但又要把控风险的时候,这个时候大数据信用就会作为风控机构交叉核查的重要依据。那你知道大数据信用报告分析和评估有…

代码随想录——二叉搜索树的最小绝对差(Leetcode530)

题目链接 层序遍历 /*** Definition for a binary tree node.* public class TreeNode {* int val;* TreeNode left;* TreeNode right;* TreeNode() {}* TreeNode(int val) { this.val val; }* TreeNode(int val, TreeNode left, TreeNode right) …

if constexpr实现条件编译

#include <iostream>// 利用if constexpr实现了条件编译 template<typename T1, typename T2> void test_func() {if constexpr (std::is_same_v<T1, T2>) {std::cout << "hit stage\n";} else {std::cout << "miss\n";} }i…

Microsoft的Copilot现已登陆Telegram

每周跟踪AI热点新闻动向和震撼发展 想要探索生成式人工智能的前沿进展吗&#xff1f;订阅我们的简报&#xff0c;深入解析最新的技术突破、实际应用案例和未来的趋势。与全球数同行一同&#xff0c;从行业内部的深度分析和实用指南中受益。不要错过这个机会&#xff0c;成为AI领…

数据结构复习指导之插入排序

文章目录 排序 考纲内容 知识框架 复习提示 1.排序的基本概念 1.1排序的定义 2.插入排序 2.1直接插入排序 2.2折半插入排序 2.3希尔排序 知识回顾 排序 考纲内容 &#xff08;一&#xff09;排序的基本概念 &#xff08;二&#xff09;插入排序 直接插…

内网不能访问域名怎么办?

在网络应用中&#xff0c;我们常常遇到内网不能访问域名的问题。这是由于内网环境限制导致的&#xff0c;内网无法直接连接到公网&#xff0c;因而无法访问互联网上的域名。我们可以利用一些特殊技术和工具来解决这个问题。 天联组网技术的应用 天联组网是一种非常受欢迎的解决…

NetApp财季报告亮点:全闪存阵列需求强劲,云计算收入增长放缓但AI领域前景乐观

在最新的财季报告中&#xff0c;NetApp的收入因全闪存阵列的强劲需求而显著增长。截至2024年4月26日的2024财年第四季度&#xff0c;NetApp的收入连续第三个季度上升&#xff0c;达到了16.7亿美元&#xff0c;较前一年同期增长6%&#xff0c;超出公司指导中值。净利润为2.91亿美…

前端开发:$nextTick()的使用及原理

目录 前言 $nextTick()的概念 $nextTick()的用法和原理 1、$nextTick()用法 2、$nextTick()原理 $nextTick()的具体使用示例 拓展&#xff1a;面试中考察$nextTick()的底层原理 最后 前言 在前端开发中&#xff0c;涉及到JS原生的使用原理是非常重要的知识点&#xff0…

使用pytorch搭建textCNN、BERT、transformer进行文本分类

首先展示数据处理后的类型&#xff1a; 第一列为文本&#xff0c;第二类为标注的标签&#xff0c;数据保存在xlsx的表格中&#xff0c;分为训练集和验证集。 textCNN 直接上整个工程代码&#xff1a; import pandas as pd import numpy as np import torch from torch.util…

c++函数基础总结

在给出的代码片段中&#xff0c;我们看到两部分内容&#xff1a;一个类定义和一个全局函数声明。让我们逐一分析它们&#xff1a; 类定义&#xff1a; cpp复制代码 class { public: void a(); }; 这个类定义是不完整的&#xff0c;因为它没有类名。但为了说明&#xff0c;我…

Linux开发工具(个人使用)

Linux开发工具 1.Linux yum软件包管理器1.1Linux安装程序有三种方式1.2注意事项1.3如何查看&#xff0c;安装&#xff0c;卸载软件包1.3.1查看软件包1.3.2安装软件包1.3.3卸载软件 2.Linux vim编辑器2.1vim的基本操作2.2vim正常模式命令集2.3vim底行模式命令集2.4vim配置 3.Lin…

如何设置eclipse中web.xml 文件的地址

新学了一个项目 &#xff0c;项目结构与平常自己构建的web项目不同 &#xff0c;用eclipse打开之后&#xff0c;eclipse竟然自己创建了一个web.xml 而项目里面原本的web.xml 文件eclipse没有识别出来&#xff0c;导致后来浏览器访问任何路径都报错404 一、修改项目中web.xml的…

Centos7.9环境下安装Keepalived(亲测版)

目录 一、在线安装 二、离线安装 (1)、 下载 (2)、安装依赖包 (3)、解压文件 (4)、编译 (4.1)、进入 keepalived-2.2.8 目录中 (4.2)、安装Keepalived (5)、配置文件修改 (6)、启动 (7)、检查启动状态 (8)、 设置开机自启 (9)、配置从节点 (10)、启动从节点keepalived…

vue3中实现鼠标点击后出现点击特效

一、效果展示 图片下方为效果体验地址 缓若江海凝清光 二、代码 js中&#xff1a; <script setup lang"ts"> window.addEventListener("click", (e: MouseEvent) > {const pointer document.createElement("div");pointer.classLi…

数模混合芯片之可靠性设计

一、可靠性设计目的 数模混合芯片设计之所以需要可靠性设计&#xff0c;主要原因有以下几点&#xff1a; 工艺与环境影响&#xff1a; 半导体制造工艺存在着不可避免的随机和系统性偏差&#xff0c;这可能导致芯片内部的模拟电路和数字电路参数发生变化&#xff0c;影响性能…

CobaltStrike基本渗透

目录 CobaltStrike简介 主要功能&#xff1a; 使用注意&#xff1a; 在使用CobaltStrike进行渗透测试时&#xff0c;务必遵守法律法规&#xff0c;并获得合法授权。 CobaltStrike安装 前提 安装 服务端安装 windows安装 CS基本使用 监听器配置 一些基本的攻击…