RabbitMQ 架构分析

文章目录

  • 前言
  • 一、RabbitMQ架构分析
    • 1、Broker
    • 2、Vhost
    • 3、Producer
    • 4、Messages
    • 5、Connections
    • 6、Channel
    • 7、Exchange
    • 7、Queue
    • 8、Consumer
  • 二、消息路由机制
    • 1、Direct Exchange
    • 2、Topic Exchange
    • 3、Fanout Exchange
    • 4、Headers Exchange
    • 5、notice
      • 5.1、备用交换机(Alternate Exchange)
      • 5.2、Mandatory标志
      • 5.3、Dead Letter Exchange(死信交换机)


前言

RabbitMQ是一款功能强大、利用广泛的消息中间件,通常用于分布式系统和微服务架构中的异步通信。理解其架构有助于正确配置和优化系统,实现高效可靠的消息传递


一、RabbitMQ架构分析

架构图
在这里插入图片描述

1、Broker

我们要使用 RabbitMQ 来收发消息,必须要安装一个 RabbitMQ 的服务,可以安装在 Windows 上面也可以安装在 Linux 上面,默认是 5672 的端口。
这台 RabbitMQ 的服务器我们把它叫做 Broker。

2、Vhost

我们每个需要实现基于 RabbitMQ 的异步通信的系统,都需要在 Broker 上创建自己要用到的交换机、队列和它们的绑定关系。

如果某个业务系统不想跟别人混用一个 Broker,有办法不需要安装多个 RabbitMQ 的服务吗?那就是建立一个新的虚拟主机 VHOST。 VHOST 除了可以提高硬件资源的利用率之外,还可以实现资源的隔离和权限的控制。

它的作用类似于其他编程语言中的 namespace 和 package,不同的 VHOST 中可以有同名的 Exchange 和 Queue,它们是完全独立的。

我们可以为不同的业务系统创建专属于他们自己的 VHOST,然后再为他们创建专属的用户,给用户分配对应的 VHOST 的权限。

我们安装 RabbitMQ 的时候会自带一个默认的 VHOST,名字是“/”。

3、Producer

消息生产者(Producer)是发送消息的应用程序。在RabbitMQ中,生产者将消息发送到交换机(Exchange)而不是直接发给队列。
生产者可以是任何能够与RabbitMQ服务器进行通信并使用AMQP协议发送消息的应用。

4、Messages

消息是由生产者发送给RabbitMQ的内容单元,每条消息由负载(payload)和一些元数据(metadata)组成。负载是消息的实际内容,而元数据包含消息的属性或标识。

5、Connections

生产者和消费者与RabbitMQ服务器之间的TCP长连接,每个连接用于消息的传递和通信。

6、Channel

如果所有的生产者发送消息和消费者接收消息,都直接创建和释放 TCP 长连接的话,对于 Broker 来说肯定会造成很大的性能损耗,也会浪费时间。

在 AMQP 里面引入了 Channel (消息信道)的概念,它是一个虚拟的连接。这样我们就可以在保持的 TCP 长连接里面去创建和释放Channel,大大了减少了资源消耗。

不同的 Channel 是相互隔离的,每个 Channel 都有自己的编号。每个客户端线程独占一个channel。
Channel 是 RabbitMQ 原生 API 里面的最重要的编程接口,也就是说我们定义交换机、队列、绑定关系,发送消息,消费消息,调用的都是Channel 接口上的方法。

7、Exchange

现在我们来思考一个问题,如果要把一条消息发送给多个队列,给多个消费者消费,如果交给生产者来做,当有成千上万个队列的时候,那要发送大量消息,耗费太多资源。

如何更优雅的处理呢?RabbitMQ考虑到了这一点,它设计了一个帮我们路由消息的组件,叫做 Exchange。

不管有多少个队列需要接收消息,我都只需要发送到 Exchange 就OK了,由它帮我来分发。Exchange 是不会存储消息的,它只做一件事情,根据规则分发消息。 分发就需要有规则,并且与队列绑定,然后进行个性话分发。

Exchange 和队列是多对多的绑定关系,也就说,一个交换机的消息一个路由给多个队列,一个队列也可以接收来自多个交换机的消息。

绑定关系建立好之后,生产者发送消息到 Exchange,也会携带一个特殊的标识。 当这个标识跟绑定的标识匹配的时候,消息就会发给一个或者多个符合规则的队列。

创建Exchange可以配置相关的属性

属性含义
name交换机的名称
type交换机的类型决定了它如何路由消息。主要有以下几种类型。Direct:路由规则是完全匹配路由键。
Fanout:把接收到的消息广播到所有绑定的队列上,不需要匹配键。
Topic:根据路由键的通配符模式进行路由。
Headers:根据消息头属性进行匹配,而不是路由键。
durable是否持久化(重启 rabbitmq 之后, 交换机是否还存在),默认false
autoDelete当所有绑定到这个交换机的队列都不再使用时,交换机会自动删除。默认为false
internal交换机将是内部的,不能被生产者直接发送消息,只能用于交换机到交换机的绑定(即路由)。默认为false
arguments一个字典,可以包含与RabbitMQ插件或未来版本兼容的其他设置。具体参数可能会依赖于具体的交换机类型和业务需求,比如
alternate-exchange:配置一个备用交换机(当消息无法路由时,消息将发送到这个备用交换机)

7、Queue

在 Broker 上有一个对象用来存储消息,在 RabbitMQ 里面这个对象叫做 Queue。
队列也是生产者和消费者的纽带,生产者发送的消息到达队列,在队列中存储,消费者从队列获取消息进行消费。

队列的相关属性

属性含义
name
durable(默认为false)设置队列是否为持久化队列。如果设置为true,即使RabbitMQ服务器重启,队列仍然存在。
exclusive(默认为false)设置队列是否为排他队列。如果设置为true,队列仅限于首次声明它的连接(Connection)使用,并且连接关闭时队列会自动删除。这个参数优先级高于autoDelete
autoDelete(默认为false)设置队列是否在所有消费者断开连接后自动删除。如果设置为true,当不再有任何消费者订阅该队列时,它会被删除。
argumentsx-message-ttl: 每个消息的生存时间(以毫秒为单位)。如果消息在队列中停留超过这个时间则会被删除。
x-expires: 在队列空闲(没有消费者订阅)超过指定时间(以毫秒为单位)后,队列会被删除。
x-max-length: 队列允许的最大消息数。超过限制时,新消息将从队列的头部移除(可选丢弃策略)。
x-max-length-bytes: 队列允许的最大字节数。超过限制时,新消息将从队列的头部移除(可选丢弃策略)。
x-dead-letter-exchange: 设置消息从该队列被移除(由于TTL、长度限制等)后发送到的备用交换机(Dead Letter Exchange)。
x-dead-letter-routing-key: 设置消息重新投递到备用交换机时使用的路由键。
x-max-priority: 队列的最大优先级,如果设置,队列将成为一个优先级队列。
x-queue-mode: 可以设置为 lazy,表明队列应尽可能保存消息到磁盘而不是内存中,以减少RAM的占用。
x-queue-master-locator: 用于集群模式下,控制队列主节点的分布策略。

8、Consumer

消费者消费消息有两种模式。

  • Pull 模式,对应的方法是 basicGet。消息存放在服务端,只有消费者主动获取才能拿到消息。如果每隔一段时间获取一次消息,消息的实时性会降低。但是好处是可以根据自己的消费能力决定获取消息的频率
  • Push 模式,对应的方法是 basicConsume,只要生产者发消息到服务器,就马上推送给消费者,消息保存在客户端,实时性很高,如果消费不过来有可能会造成消息积压。

Spring AMQP 是 push 方式,通过事件机制对队列进行监听,只要有消息到达队列,就会触发消费消息的方法。

RabbitMQ 中 pull 和 push 都有实现。而 kafka 和 RocketMQ 只有 pull。

由于队列有 FIFO 的特性,只有确定前一条消息被消费者接收之后,Broker 才会把这条消息从数据库删除,继续投递下一条消息。
一个消费者是可以监听多个队列的,一个队列也可以被多个消费者监听。 但是在生产环境中,我们一般是建议一个消费者只处理一个队列的消息。
如果需要提升处理消息的能力,可以增加多个消费者。这个时候消息会在多个消费者之间轮询。

二、消息路由机制

RabbitMQ 中一共有四种类型的交换机,Direct、Topic、Fanout、Headers(不常用)。

1、Direct Exchange

特点:

  • 直接交换机根据消息的路由键(Routing Key)与绑定键(Binding Key)进行精确匹配,将消息路由到绑定的队列。

工作原理:

  • 当生产者发送消息时,需要指定路由键。
  • 交换机会查找与该路由键完全匹配的绑定队列,并将消息路由到这些队列。

在这里插入图片描述

例如发送如下消息

# 只有 binding key = rabbit 能收到消息
channel.basicPublish(“MY_DIRECT_EXCHANGE”,”rabbit”,”msg 1”);# 匹配不上,消息丢失
channel.basicPublish(“MY_DIRECT_EXCHANGE”,”rabbit0000”,”msg 1”);

应用场景:

  • 适用于需要精确匹配路由键的场景,比如日志系统中不同级别的日志(“info”, “error”)可以精确路由到不同的队列进行处理。

2、Topic Exchange

特点:

  • 主题交换机通过模式匹配路由键和绑定键来路由消息,支持部分匹配和通配符。

工作原理:

  • 生产者发送消息时指定带有模式的路由键,路由键可以包含点号(.)分隔的多个单词,例如 a.bc.def 是 3 个单词。
  • 绑定键可以包含通配符:"*“匹配一个单词,”#"匹配零个或多个单词,可以当前缀也可以当后缀。

在这里插入图片描述

分析:

  • rabbit.#:支持路由键以 rabbit 开头的消息路由,后面可以有单词,也可以没有
  • #.rocket:支持路由键以 rocket 结尾,前面可以有也可以没有单词的消息路由
  • kafka.*:支持路由键以 kafka 开头,并且后面是一个单词的消息路由

例如发送如下消息

# 只有 binding key = rabbit.# 能收到消息
channel.basicPublish("MY_TOPIC_EXCHANGE","rabbit.abc.jvm","msg 2");# 只有 binding key = kafka.* 能收到消息
channel.basicPublish("MY_TOPIC_EXCHANGE","kafka.jvm","msg 2");# 只有 binding key = #.rocket 能收到消息
channel.basicPublish("MY_TOPIC_EXCHANGE","kafka.abc.rocket","msg 2");# binding key = rabbit.# , binding key = #.rocket 能收到消息
channel.basicPublish("MY_TOPIC_EXCHANGE","rabbit.dad.rocket","msg 2");# binding key = #.rocket , binding key = kafka.*  能收到消息
channel.basicPublish("MY_TOPIC_EXCHANGE","kafka.rocket","msg 2");

应用场景:

  • 适用于多层次、复杂的路由场景,例如新闻系统可以根据新闻类别和地区进行路由。

3、Fanout Exchange

特点:

  • 广播交换机会将接收到的每条消息路由到所有绑定的队列,而不考虑路由键。

工作原理:

  • 生产者发送消息时无需指定路由键。
  • 消息会被广播到所有与交换机绑定的队列中。

在这里插入图片描述

例如发送如下消息

# 三个队列都会收到 msg 4
channel.basicPublish("MY_FANOUT_EXCHANGE", "", "msg 4");

应用场景:

  • 适用于广播场景,将一条消息分发给多个消费者,常用于发布-订阅模式。比如广告、日志等

4、Headers Exchange

特点:

  • 头交换机不是通过路由键,而是通过消息头(Headers)中的属性进行匹配路由。

工作原理:

  • 绑定包含一组键值对,消息头也包含键值对。
  • 交换机将消息头与绑定键值对进行匹配来路由消息,可选择完全匹配或者部分匹配。

应用场景:

  • 适用于消息头包含丰富信息,且需要基于多属性进行路由的场景。例如,需要根据多种复杂条件路由的应用。

5、notice

Topic Exchange 和 Direct Exchange 匹配不上都可能导致消息丢失。
可以按照如下方式处理

5.1、备用交换机(Alternate Exchange)

备用交换机是RabbitMQ提供的一种机制,用来处理没有匹配队列的消息。你可以为一个交换机配置一个备用交换机,当消息未能被路由到任何队列时,这些消息会被发送到备用交换机进行处理。

配置备用交换机步骤:

  • 创建一个备用交换机。
  • 在主要交换机上设置该备用交换机。

示例:

  • 假设我们有一个主交换机名为primary_exchange和一个备用交换机名为alternate_exchange,你可以用以下方式设置:
// 当消息在primary_exchange中没有匹配的队列时,消息将被路由到alternate_exchange。 
Map<String, Object> args = new HashMap<String, Object>();
args.put("alternate-exchange", "alternate_exchange");
channel.exchangeDeclare("primary_exchange", "direct", true, false, args);

5.2、Mandatory标志

Mandatory标志是RabbitMQ的消息属性之一。在生产者发送消息时,如果设置了mandatory标志为true,而消息未能找到匹配的队列,那么消息不会被丢弃,Broker会将消息返回给生产者。

如何使用:

  • 在生产者发送消息时,可以配置mandatory标志。
String message = "Hello World!";
channel.basicPublish(exchangeName, routingKey, true, null, message.getBytes());

Callback机制可以用来处理返回的未路由消息。需要设置ReturnCallback:

channel.addReturnListener(new ReturnCallback() {public void handle (Return r){
// 处理未路由的消息System.out.println("Returned message: " + new String(r.getBody()));}
});

5.3、Dead Letter Exchange(死信交换机)

未被正确路由的消息有时会被重定向到称为死信交换机(DLX)的特殊交换机。当一个队列的消息变为死信时,这些消息可以被路由到一个死信交换机上。

一些情况会导致消息变为死信:

  • 消息被拒绝(basic.reject 或 basic.nack)并且 requeue 属性设置为 false。
  • 消息在队列中存活时间超过TTL(Time to Live)。
  • 队列的消息数超过了最大长度(队列溢出)。

你可以配置队列使用死信交换机:

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-dead-letter-exchange", "dlx_exchange");
channel.queueDeclare("queue_name", true, false, false, args);

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

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

相关文章

九、CSS工程化方案

一、PostCSS介绍 二、PostCSS插件的使用 项目安装 - npm install postcss-cli 全局安装 - npm install postcss-cli -g postcss-cli地址&#xff1a;GitHub - postcss/postcss-cli: CLI for postcss postcss地址&#xff1a;GitHub - postcss/postcss: Transforming styles…

SpringBoot开发(二)Spring Boot项目构建、Bootstrap基础知识

1. Spring Boot项目构建 1.1. 简介 基于官方网站https://start.spring.io进行项目的创建. 1.1.1. 简介 Spring Boot是基于Spring4框架开发的全新框架&#xff0c;设计目的是简化搭建及开发过程&#xff0c;并不是对Spring功能上的增强&#xff0c;而是提供了一种快速使用Spr…

GESP2024年3月认证C++六级( 第三部分编程题(2)好斗的牛)

参考程序&#xff08;暴力枚举&#xff09; #include <iostream> #include <vector> #include <algorithm> using namespace std; int N; vector<int> a, b; int ans 1e9; int main() {cin >> N;a.resize(N);b.resize(N);for (int i 0; i &l…

SpringBoot统一数据返回格式 统一异常处理

统一数据返回格式 & 统一异常处理 1. 统一数据返回格式1.1 快速入门1.2 存在问题1.3 案列代码修改1.4 优点 2. 统一异常处理 1. 统一数据返回格式 强制登录案例中,我们共做了两部分⼯作 通过Session来判断⽤⼾是否登录对后端返回数据进⾏封装,告知前端处理的结果 回顾 后…

Elasticsearch+kibana安装(简单易上手)

下载ES( Download Elasticsearch | Elastic ) 将ES安装包解压缩 解压后目录如下: 修改ES服务端口&#xff08;可以不修改&#xff09; 启动ES 记住这些内容 验证ES是否启动成功 下载kibana( Download Kibana Free | Get Started Now | Elastic ) 解压后的kibana目…

十年筑梦,再创鲸彩!庆祝和鲸科技十周年

2025 年 1 月 16 日&#xff0c;“十年筑梦&#xff0c;再创鲸彩” 2025 和鲸科技十周年庆暨 2024 年终表彰大会圆满落幕。 十年征程&#xff0c;和鲸科技遨游于科技蓝海&#xff0c;破浪前行&#xff0c;无惧风雨。期间所取得的每一项成就&#xff0c;都凝聚着全体成员的智慧结…

【Uniapp-Vue3】动态设置页面导航条的样式

1. 动态修改导航条标题 uni.setNavigationBarTitle({ title:"标题名称" }) 点击修改以后顶部导航栏的标题会从“主页”变为“动态标题” 2. 动态修改导航条颜色 uni.setNavigationBarColor({ backgroundColor:"颜色" }) 3. 动态添加导航加载动画 // 添加加…

openlayer getLayerById 根据id获取layer图层

背景&#xff1a; 在项目中使用getLayerById获取图层&#xff0c;这个getLayerById()方法不是openlayer官方文档自带的&#xff0c;而是自己封装的一个方法&#xff0c;这个封装的方法的思路是&#xff1a;遍历所有的layer&#xff0c;根据唯一标识【可能是id&#xff0c;也可能…

Unity入门2 背景叠层 瓦片规则

切割场景 瓦片调色盘 放在Assets里面新建瓦片地图,palettes tile 瓦片 palettes调色板 上下窗口是分开的 拖进这个格子窗 瓦片太碎&#xff0c;要封装 装好之后&#xff0c;只是把瓦片放上去了&#xff0c;但是还没有画布&#xff0c;显示是这样的 no valid target 新建“…

Kafka 日志存储 — 日志清理

Kafka 提供两种日志清理策略&#xff1a;日志清理(Log Delete)与日志压缩(Log Compaction)。 1 日志清理 通过broker端参数log.cleanup.policy来设置日志清理策略&#xff0c;默认值为“delete”。如果要采用日志压缩的清理策略&#xff0c;则设置为“compact”。可以同时支持…

Semantic Kernel - Kernel理解

目录 一、关于Kernel 二、案例实战 三、运行截图 一、关于Kernel 微软的 Semantic Kernel 项目中,Semantic Kernel 是一个工具框架,旨在使得开发人员能够更容易地将大语言模型(如GPT)集成到不同的应用中。它通过提供一组接口、任务模板和集成模块,使开发者能够轻松地设计…

svn: E000111: Error running context: Connection refused

1、具体报错&#xff1a; 看起来是window主机的子系统ubuntu svn客户端无法访问到window主机的svn的服务端。 2、问题&#xff1a; window主机安装子系统ubuntu&#xff0c;ubuntu是可以直接访问外网&#xff0c;但是ubuntu是不能访问window主机的服务&#xff0c;比如svn的se…

【时时三省】(C语言基础)对比一组函数

山不在高&#xff0c;有仙则名。水不在深&#xff0c;有龙则灵。 ----CSDN 时时三省 对比一组函数 比如对比一下scanf fscanf sscanf和printf fprintf sprintf scanf 针对标准输入的格式化的输入语句 其实它针对的是stdin fscanf 针对所有输入流的格式化的输入语句 它是针对s…

(MySQL)头歌数据库作业答案

1.数据库和表的基本操作 1.1数据库和表的基本操作&#xff08;一&#xff09; 第1关&#xff1a;查看表结构与修改表名 本关任务&#xff1a;修改表名&#xff0c;并能顺利查询到修改后表的结构。 USE Company;#请在此处添加实现代码 ########## Begin ###################…

计算机网络 (57)改进“尽最大努力交付”的服务

前言 计算机网络中的“尽最大努力交付”服务是网络层的一种数据传输方式。这种服务的特点是网络层只负责尽力将数据报从源端传输到目的端&#xff0c;而不保证数据传输的可靠性。 一、标记与分类 为数据分组打上标记&#xff1a; 给不同性质的分组打上不同的标记&#x…

Direct Preference Optimization (DPO): 一种无需强化学习的语言模型偏好优化方法

论文地址&#xff1a;https://arxiv.org/pdf/2305.18290 1. 背景与挑战 近年来&#xff0c;大规模无监督语言模型&#xff08;LM&#xff09;在知识获取和推理能力方面取得了显著进展&#xff0c;但如何精确控制其行为仍是一个难题。 现有的方法通常通过**强化学习从人类反馈&…

ppp综合实验

IP地址 r1 r2 r3 r4 hdlc封装 pap认证 r2 r3 chap认证 r2 r4 MGRE 主认证 [r1]int Tunnel 0/0/0 [r1-Tunnel0/0/0]ip add 192.168.4.1 24 [r1-Tunnel0/0/0]tunnel-protocol gre p2mp [r1-Tunnel0/0/0]source 12.1.1.1 [r1-Tunnel0/0/0]nhrp entry multicast dynamic [r1-Tu…

高效流式大语言模型(StreamingLLM)——基于“注意力汇聚点”的突破性研究

论文地址&#xff1a;https://arxiv.org/pdf/2309.17453 github地址&#xff1a;https://github.com/mit-han-lab/streaming-llm 1. 研究背景与挑战 随着大语言模型&#xff08;LLMs&#xff09;在对话系统、文档摘要、代码补全和问答等领域的广泛应用&#xff0c;如何高效且准…

学习数据结构(2)空间复杂度+顺序表

1.空间复杂度 &#xff08;1&#xff09;概念 空间复杂度也是一个数学表达式&#xff0c;表示一个算法在运行过程中根据算法的需要额外临时开辟的空间。 空间复杂度不是指程序占用了多少bytes的空间&#xff0c;因为常规情况每个对象大小差异不会很大&#xff0c;所以空间复杂…

【测试】UI自动化测试

长期更新&#xff0c;建议关注收藏点赞&#xff01; 目录 概论WEB环境搭建Selenium APPAppium 概论 使用工具和代码执行用例。 什么样的项目需要自动化&#xff1f; 需要回归测试、自动化的功能模块需求变更不频繁、项目周期长&#xff08;功能测试时长&#xff1a;UI自动化测…