手写消息队列(基于RabbitMQ)

一、什么是消息队列?

提到消息队列是否唤醒了你脑海深处的记忆?回看前面的这篇文章:《Java 多线程系列Ⅳ(单例模式+阻塞式队列+定时器+线程池)》,其中我们在介绍阻塞队列时说过,阻塞队列最大的用途就是实现 生产者消费者模型

我们知道对于生产者消费者模型来说,它具有两个十分亮眼的特点:

  1. 解耦合.
  2. 削峰填谷.

(1)解耦合
在引入生产者消费者模型之前,两台服务器之间通常是直接交互,这种交互模式使得服务器之间的耦合是非常大的。而引入生产者消费者模型之后,两台服务器之间不再进行直接通信,而是借助阻塞队列进行业务处理,起到了解耦的效果。

(2)削峰填谷
在引入生产者消费者模型之前,同样是两台服务器进行直接通信,如果在一个时间点,服务器 A 突然发送一组请求峰值,此刻服务器 B 也会随之感受到峰值,这种情况下很可能造成服务器故障。如果此时使用阻塞队列,A 将收到的请求发给队列,虽然队列中有很多请求,但是服务器 B 仍然和以按照原有的节奏读取请求。

其实正是因为生产者消费者模型具有以上诸多好处,在实际的后端开发中,特别是分布式系统里,跨主机使用生产者消费者模型是非常普遍的需求。因此通常会把阻塞队列单独分离出来,赋予更加丰富的功能,封装成一个独立的服务器程序,这个程序就称为 消息队列

二、需求整理

1、生产者消费者模型核心概念

  1. 生产者 (Producer): 负责将消息发送到消息队列中。

  2. 消费者 (Consumer): 从消息队列中接收和处理消息。。

  3. 中间人 (Broker): 它负责接收发布者发送的消息,并将这些消息存储在队列中,然后将这些消息传递给订阅者。

  4. 发布 (Publish): 生产者将消息投递到中间人的过程。

  5. 订阅 (Subscribe): 消费者在中间人这里注册的过程。只有消费者注册之后,当一个消息发布到消息队列时消息才会被发送给相应的订阅者。

根据以上概念,我们可以大致画出生产者消费者模型概念图:(PS:下面的每个模块均表示服务器)

一个生产者,一个消费者:

N 个生产者,N 个消费者:

2、Broker 设计概要

我们当前的目的是为了实现一个消息队列,其中 Broker 是最核心的部分,它主要负责消息的 存储转发,其中涉及到的核心概念如下:

  1. 虚拟主机 (VirtualHost): 类似于 MySQL 的 “database”,是⼀个逻辑上的集合。在实际的开发中一个 BrokerServer 可能会同时管理多组业务线上的数据,此时可以使用不同的 VirtualHost 进行区分。

  2. 交换机 (Exchange): 生产者把消息先发送到 Broker 的 Exchange 上,再根据不同的规则,把消息转发给不同的 Queue。

  3. 队列 (Queue): 真正用来存储消息的部分,每个消费者决定自己从哪个 Queue 上读取消息(根据订阅的队列)。⼀个 Exchange 可以绑定多个 Queue (可以向多个 Queue 中转发消息),一个 Queue 也可以被多个 Exchange 绑定
    (一个 Queue 中的消息可以来自于多个 Exchange)。

  4. 绑定 (Binding): Exchange 和 Queue 之间的关联关系,Exchange 和 Queue 可以理解成 “多对多” 关系。使用一个关联表就可以把这两个概念联系起来。

  5. 消息 (Message): 具体来说是服务器之间的请求和响应。一个消息,可以视为一个字符串(二进制数据),具体由程序员自定义。

上述概念在 Broker 中的体现如图所示:

补充说明1:数据存储

以上这些概念对应的数据,既需要在内存中存储,也需要在硬盘上存储,以内存为主,硬盘为辅:

  • 内存存储:对于 MQ 来说,能够高效的处理转发数据时非常关键的指标,因此使用内存组织上述数据,能够得到较高的效率。
  • 硬盘存储:主要是为了防止内存中的数据随着进程/主机重启而丢失。

补充说明:2: 交换机类型与转发规则

上面我们提到,在生产者发送消息时,首先会将消息发送到 Broker 的交换机上,再由交换机根据不同的规则转发到相应的队列中。在 MQ 中支持四种类型的交换机,它们分别是: Direct(直接交换机)、Fanout(扇出交换机)、Topic(主题交换机)、Header(头部交换机)。其中 Header 这种方式比较复杂,也比较少见,当前项目中主要实现了前三种,下面分别对他们进行详细介绍:

前要说明:

  • 以下 bindingKey(绑定键)是在创建队列和交换机绑定关系时指定的关键字。
  • 以下 routingKey(路由键)是生产者发送消息时指定的关键字。

(1)Direct(直接交换机)

  1. 生产者发送消息时,会指定一个"目标队列"的名字(此时的 routingKey 就是 队列的名字,bindingKey 无效)
  2. 交换机收到消息后,查看当前交换机对应的绑定里面是否存在队列名字为routingKey的队列
  3. 如果有,就转发过去(把消息塞进对应的“目标队列”中)
  4. 如果没有,消息直接丢弃

(2)Fanout(扇出交换机)

  1. 生产者无需指定routingKey,直接发送消息到指定交换机
  2. 交换机收到消息后,直接将消息转发给当前交换机已绑定的所有队列中。(此时的 bindingKey 和 routingKey 对扇出交换机无效。)

(3)Topic(主题交换机)

  1. 生产者发送消息时,指定一个 routingKey
  2. 交换机收到消息后,查看当前交换机对应的绑定中是否存在一个 bindingKey 通过一定的规则和 routingKey 相匹配
  3. 如果有,就将消息转发到对应的绑定队列中。
  4. 如果没有,则将消息丢弃。

PS:以上所有概念出自 AMQP 协议:一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。

3、Broker 核心 API

Broker 基于以上概念和功能,需要实现的核心 API 如下:

  1. 创建队列 (queueDeclare)
  2. 销毁队列 (queueDelete)
  3. 创建交换机 (exchangeDeclare)
  4. 销毁交换机 (exchangeDelete)
  5. 创建绑定 (queueBind)
  6. 解除绑定 (queueUnbind)
  7. 发布消息 (basicPublish)
  8. 订阅消息 (basicConsume)
  9. 确认消息 (basicAck)

补充说明:

  1. 从上面可以看出来,在进行创建操作时并没有使用 create,而是使用 declare,这是从语义上说明,这里的创建起到的效果是不存在则创建,存在则啥也不做。
  2. 上述并没有创建一个“消费消息”的API,这是因为当前我们使用的工作模式是 push(推),Broker 会将消息主动推送给订阅的消费者。当然也有 pull(拉) 工作模式,需要消费者主动调用 Broker 的 API 获取消息,当前项目不涉及这种模式。
  3. 在MQ中有两种应答方式,一种是自动应答,这种方式下 Broker 将消息推送给订阅的消费者后就算应答完毕。另一种应答方式是手动应答,上述确认消息(basicAck)起到的效果,是可以让消费者 显式 的告诉 Broker,这个消息我处理完毕了,提高整个系统的可靠性。

4、客户端设计概要

生产者、消费者都是客户端程序,broker 则是作为服务器,通过网络进行通信。此处设定,使用 TCP + 自定义的应用层协议 实现 生产者/消费者 和 BrokerServer 之间的交互工作,这里需要给客户端提供一组 API,让客户端的业务代码来调用,从而通过网络通信的方式远程调用 brokerserver 上的方法。

客户端核心API:

  1. 创建 Connection
  2. 关闭 Connection
  3. 创建 Channel
  4. 关闭 Channel
  5. 创建队列 (queueDeclare)
  6. 销毁队列 (queueDelete)
  7. 创建交换机 (exchangeDeclare)
  8. 销毁交换机 (exchangeDelete)
  9. 创建绑定 (queueBind)
  10. 解除绑定 (queueUnbind)
  11. 发布消息 (basicPublish)
  12. 订阅消息 (basicConsume)
  13. 确认消息 (basicAck)

补充说明:

  1. 和 Broker 服务器 API 相比,客户端程序还提供了如下 4 个 API:创建 Connection、关闭 Connection、创建 Channel、关闭 Channel。
  2. 这里的一个Connection对象代表一个TCP连接。
  3. Channel 是 Connection 内部逻辑上的链接,多个Channel复用同一个TCP连接,一个Connection 中可以包含多个Channel,每个Channel负责客户端中不同的模块,其中传输的数据是互不相干的。
  4. 这样的设定主要是为了能够更好的复用 TCP 连接, 达到长连接的效果, 避免频繁的创建关闭 TCP 连接。
  5. 上述客户端提供的 API 只是给业务代码进行调用,真正的方法执行是交给了BrokerServer。这个过程称为 远程过程调用(Remote Procedure Call,简称RPC)是一种计算机通信协议,它允许程序调用另一个地址空间(通常是不同的计算机)中的过程或函数,而无需程序员显式编写网络代码。通过使用RPC,应用程序可以像调用本地进程一样调用远程服务器上的进程。

5、小结

最后简单总结一下,我们大致需要做的工作,其中涉及到的细节问题,我们后面在进行补充:

  1. 实现生产者、BrokerServer、消费者三个部分
  2. 针对生产者、消费者来说,主要编写客户端和服务器的网络通信部分。
  3. 重点实现BrokerServer,包括内部的基本概念和核心 API。
  4. 数据的持久化存储。

三、具体实现

附上连接:

1、消息队列详细设计与实现思维导图
2、Gitee 完整代码地址

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

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

相关文章

在VSCode创建vue项目,出现“因为在此系统上禁止运行脚本”问题

问题:vue : 无法加载文件 C:\Users\***\***\Roaming\npm\vue.ps1,因为在此系统上禁止运行脚本。有关详细信息,请参阅 ht tps:/go.microsoft.com/fwlink/?LinkID135170 中的 about_Execution_Policies。 所在位置 行:1 字符: 1 解决&#xff…

目标检测标注工具AutoDistill

引言 在快速发展的机器学习领域,有一个方面一直保持不变:繁琐和耗时的数据标注任务。无论是用于图像分类、目标检测还是语义分割,长期以来人工标记的数据集一直是监督学习的基础。 然而,由于一个创新性的工具 AutoDistill&#x…

Linux shell编程学习笔记26:stty(set tty)

之前我们探讨了Linux中的tty,tty命令的主要功能是显示当前使用的终端名称。 如果我们想进一步对tty进行设置,就要用到stty。 stty的功能:显示和修改终端特性(Print or change terminal characteristics)。 1 stty -…

轻量封装WebGPU渲染系统示例<35>- HDR环境数据应用到PBR渲染材质

当前示例源码github地址: https://github.com/vilyLei/voxwebgpu/blob/feature/rendering/src/voxgpu/sample/BasePbrMaterialTest.ts 当前示例运行效果: 微调参数之后的效果: 此示例基于此渲染系统实现,当前示例TypeScript源码如下: export class BasePbrMateri…

Sentinel 熔断规则 (DegradeRule)

Sentinel 是面向分布式、多语言异构化服务架构的流量治理组件,主要以流量为切入点,从流量路由、流量控制、流量整形、熔断降级、系统自适应过载保护、热点流量防护等多个维度来帮助开发者保障微服务的稳定性。 SpringbootDubboNacos 集成 Sentinel&…

【Linux】21、软中断、网络小包、SYN FLOOD 攻击、sar tcpdump

文章目录 一、通俗理解:从“取外卖”看中断二、软中断2.1 网卡收发数据包2.2 查看软中断和内核线程2.3 案例2.3.1 案例:动态库 sleep 导致软中断2.3.2 Nginx 进程的不可中断状态是系统的一种保护机制,可以保证硬件的交互过程不被意外打断。所…

QT自定义信号,信号emit,信号参数注册

qt如何自定义信号 使用signals声明返回值是void在需要发送信号的地方使用 emit 信号名字(参数)进行发送 在需要链接的地方使用connect进行链接 ct进行链接

无需云盘,不限流量实现Zotero跨平台同步:内网穿透+私有WebDAV服务器

🔥博客主页: 小羊失眠啦. 🎥系列专栏:《C语言》 《数据结构》 《Linux》《Cpolar》 ❤️感谢大家点赞👍收藏⭐评论✍️ 无需云盘,不限流量实现Zotero跨平台同步:内网穿透私有WebDAV服务器 文章目…

系列十一、你平时工作用过的JVM常用基本配置参数有哪些?

一、常用参数 1.1、-Xms 功能:初始内存大小,默认为物理内存的1/64,等价于 -XX:InitialHeapSize 1.2、-Xmx 功能:最大分配内存,默认为物理内存的1/4,等价于 -XX:MaxHeapSize 1.3、-Xss 功能:设置…

RoCE、IB和TCP等网络的基本知识及差异对比

目前有三种RDMA网络,分别是Infiniband、RoCE(RDMA over Converged Ethernet)、iWARP。 其中,Infiniband是一种专为RDMA设计的网络,从硬件级别保证可靠传输 ,技术先进,但是成本高昂。 而RoCE 和 iWARP都是基于以太网的…

9.docker镜像Tag为none的原因

1.现象 使用docker images命令查看镜像列表,会发现存在许多标签为none的镜像: 2. 原因 docker镜像标签为none的原因如下: (1)构建或重新拉取同名同Tag的新镜像:构建或重新拉取同名同Tag的新镜像后&…

什么是单域名SSL安全证书?

单域名证书是什么? 单域名证书是指只包含一个具体域名的SSL/TLS证书,它可以用于保护单个主机名的HTTPS通信。例如,如果您有一个网站http://www.example.com,则单域名证书将仅为该域名颁发。 这种证书在保护单个域的安全方面很有…

open3d ICP 配准

文章目录 Three common registration techniquesPoint-to-point techniquePoint-to-plane registration ICP registrationHelper visualization functionInputGlobal registrationExtract geometric featureInputRANSAC Point-to-point ICPPoint-to-plane ICP References Three…

【Java 进阶篇】Ajax 入门:打开前端异步交互的大门

欢迎来到前端异步交互的世界!在这篇博客中,我们将深入探讨 Ajax(Asynchronous JavaScript and XML),这是一项能够让你的网页在不刷新的情况下与服务器进行数据交互的技术。无论你是刚刚踏入前端开发的小白,…

pycharm2023 实现鼠标点击某行,调试时代码运行至相应行

按下图取消 Breakpoints Over Line Numbers即可,然后调试时点击某行,代码就会运行至某行

【鸿蒙应用ArkTS开发系列】- 云开发入门简介

目录 概述开发流程工程概览工程模板工程结构 工程创建与配置 概述 HarmonyOS云开发是DevEco Studio新推出的功能,可以让您在一个项目工程中,使用一种语言完成端侧和云侧功能的开发。 基于AppGallery Connect Serverless构建的云侧能力,开发…

windows Oracle Database 19c 卸载教程

目录 打开任务管理器 停止数据库服务 Universal Installer 卸载Oracle数据库程序 使用Oracle Installer卸载 删除注册表项 重新启动系统 打开任务管理器 ctrlShiftEsc可以快速打开任务管理器,找到oracle所有服务然后停止。 停止数据库服务 在开始卸载之前&a…

C++数据结构:并查集

目录 一. 并查集的概念 二. 并查集的模拟实现 2.1 并查集类的声明 2.2 并查集的实现 三. 路径压缩 四. 总结 一. 并查集的概念 在生活中,我们经常需要对某一些事物进行归类处理,即:将N个不同的元素划分为几个互不相交的集合。在初始状态…

镭速,克服UDP传输缺点的百倍提速传输软件工具

在网络传输中,我们经常会面临这样的困难:文件太大,传输速度太慢,浪费时间和流量;文件太小,传输速度太快,容易出现丢包和乱序,损害数据的完整性和正确性。这些困难的根本在于传输层协…

单图像3D重建AI算法综述【2023】

计算机视觉是人工智能的一个快速发展的领域,特别是在 3D 领域。 本概述将考虑一个应用任务:2D 和 3D 环境之间的转换。 在线工具推荐: Three.js AI纹理开发包 - YOLO合成数据生成器 - GLTF/GLB在线编辑 - 3D模型格式在线转换 - 可编程3D场景编…