“仿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,也不需要每个信道的消息处理都有自己的线程池。


四、项目模块整体关系图


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

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

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

使用 Scapy 库编写源路由攻击脚本

一、介绍 源路由攻击是一种网络攻击方法,攻击者通过利用IP数据包中的源路由选项来控制数据包的传输路径,从而绕过安全设备或防火墙,直接访问目标系统。源路由功能允许数据包的发送方指定数据包通过的路径,而不是由路由器根据路由…

达梦数据库(六) -------- 数据迁移MYSQL->达梦数据库

前言:确保安装好达梦数据库,达梦数据库(三) -------- 安装和初始化达梦数据库_达梦数据库安装-CSDN博客,需要安装达梦的数据库软件包,直接安装达梦8是包含这个功能的。安装包下载地址如下需要注册: https://www.damen…

在鲲鹏服务器搭建k8s高可用集群分享

高可用架构 本文采用kubeadm方式搭建k8s高可用集群,k8s高可用集群主要是对apiserver、etcd、controller-manager、scheduler做的高可用;高可用形式只要是为: 1. apiserver利用haproxykeepalived做的负载,多apiserver节点同时工作…

UnityLeapMotion流程记录

突然接到一个LeapMotion的项目,回想起上次做LeapMotion还是在几年前,但是当时没有去记录,所以这次就相当于是重新走了一遍流程。很苦恼,赶紧记录下来。防止之后忘记。这次的需求还是比较简单的,用手滑动控制图片序列播…

高速服务区智慧公厕管理系统引导屏UI界面展示

在现代社会,高速服务区作为人们出行途中的重要休憩场所,其各项设施的智能化水平也在不断提升。其中,智慧公厕管理系统的出现,为人们带来了更加便捷、舒适的如厕体验,而引导屏 UI 界面更是这一系统的重要展示窗口。 智慧…

FinalShell 配置SSH密钥登陆

转载请标明出处:http://blog.csdn.net/donkor_/article/details/139355489 文章目录 前言生成密钥服务器配置公钥本地配置私钥存储私钥FinalShell配置 总结 前言 本机FinalShell 配置SSH密钥登陆服务器,这样就不再需要使用密码进行登陆了。由于FinalSh…

hadoop(1)--hdfs部署(亲测可用)

一、准备: 1、三台集群部署,配置hosts #cat /etc/hosts 192.168.46.128 node1 #nameNode dataNode secondaryNameNode 192.168.46.129 node2 #datanode 192.168.46.130 node3 #datanode说明: NameNode: 主节点管理者 DataNode&…

AI+教育:OpenAI推出ChatGPT Edu教育版

大家好,我是木易,一个持续关注AI领域的互联网技术产品经理,国内Top2本科,美国Top10 CS研究生,MBA。我坚信AI是普通人变强的“外挂”,所以创建了“AI信息Gap”这个公众号,专注于分享AI全维度知识…

Java web应用性能分析之【压测工具ab】

常用的性能测试工具有:JMeter、loadRunner、ab;对于开发人员来说用的多的是免费的Jmeter和ab,对于测试来说可能用收费的商业软件loadRunner多。在这里我们就说说ab压测工具,因为ab基本满足web接口测试要求,jmeter后面再…

开发者工具-sources(源代码选项)

一、概要说明 源代码面板从视觉效果上分为三个区域:菜单区、内容区、监听区。 菜单区里面有5个子分类: 网页(Page):指页面源,包含了该页面中所有的文件,即使多个域名下的文件也都会展示出来,包括iframe…

【网络层】IP地址基础 与 子网掩码

文章目录 IP地址基础IP地址概念IP地址分类公网地址和私网地址 子网掩码子网掩码作用默认子网掩码网络地址、主机地址、广播地址 IP地址基础 IP地址概念 IP地址:IP Address 在网络中,通信节点都需要有一个IP地址 IP地址以点分十进制表示,有…

openresty(Nginx) 配置 特殊URL 密码访问 使用htpasswd 配置 Basic_Auth登录认证

1 使用htpasswd 生成密码文件.htpasswd是Apache附带的工具。如果没有可以安装。 #centos 8.5 系统 yum install httpd-tools #Ubuntu 24.04 系统 sudo apt update sudo apt-get install apache2-utils #生成密码文件,用户test sudo htpasswd -c /usr/local/openresty/nginx/…

Python魔法之旅-魔法方法(07)

目录 一、概述 1、定义 2、作用 二、应用场景 1、构造和析构 2、操作符重载 3、字符串和表示 4、容器管理 5、可调用对象 6、上下文管理 7、属性访问和描述符 8、迭代器和生成器 9、数值类型 10、复制和序列化 11、自定义元类行为 12、自定义类行为 13、类型检…

Matlab操作Excel筛选指定数据的对应数据

Matlab中在表格中寻找指定汉字,并返回其所在行数, 将该行数的另一列提取出来。 目录 一、前言 二、直接在命令行输出 三、保存筛选数据excel 一、前言 源数据excel: 指定汉子:买,得到下面数据: 二、直接…

特征工程技巧——OneHot编码

我们以Kaggle比赛里面的一个数据集跟一个公开代码为例去解释我们的OneHot编码。 简单来说,独热编码是一种将类别型变量转换为二进制表示的方法,其中每个类别被表示为一个向量,向量的长度等于类别的数量,其中只有一个元素为1&…

Kafka篇:Kafka搭建、使用、及Flink整合Kafka文档

一、Kafka搭建 1、上传并解压改名 tar -xvf kafka_2.11-1.0.0.tgz mv kafka_2.11-1.0.0 kafka-1.0.0 2、配置环境变量 vim /etc/profile export KAFKA_HOME/usr/local/soft/kafka-1.0.0 export PATH$PATH:$KAFKA_HOME/bin source /etc/profile (使环境变量生效…

05.k8s弹性伸缩

5.k8s弹性伸缩 k8s弹性伸缩,需要附加插件heapster监控 弹性伸缩:随着业务访问量的大小,k8s系统中的pod比较弹性,会自动增加或者减少pod数量; 5.1 安装heapster监控 1:上传并导入镜像,打标签 ls *.tar.gz for n in ls *.tar.gz…

Linux下的Git应用

1、卸载 2、安装 3、创建并初始化 4、配置 (附加删除语句) 5、查看(tree .git/) 6、增加和提交 7、打印日志 8、验证已操作工作

7、css3实现边框不停地跑动效果

效果例图&#xff1a; 1、上html代码&#xff1a; <!DOCTYPE html> <html lang"zh"> <head><meta charset"UTF-8" /><meta name"viewport" content"widthdevice-width, initial-scale1.0" /><meta …

unity打包的WebGL部署到IIS问题

部署之后会出错&#xff0c;我遇到的有以下几种&#xff1b; 进度条卡住不动 明明已经部署到了IIS上&#xff0c;为什么浏览网页的时候还是过不去或者直接报错。 进度条卡住不动的问题其实就是wasm和data的错误。 此时在浏览器上按F12进入开发者模式查看错误&#xff08;下图…