发布时间：2025-06-24 18:06:55 作者：北方职教升学中心 阅读量：736
交换机 Exchange
这是RabbitMQ中进行数据路由的重要组件。 他会暴露出一个虚拟IP(VIP)，并将VIP绑定到不同的网卡上。这东西分两个部分来理解：队列，是一种FIFO 先进先出的数据结构。这种方式，没有考虑消息处理的复杂度以及consumer的处理能力。large fan-outs 大规模分发
当想要向多个订阅者发送相同的消息时，以往的队列类型必须为每个消费者绑定一个专用的队列。
Note about queue size If all the workers are busy, your queue can fill up. You will want to keep an eye on that, and maybe add more workers, or have some other strategy.type为fanout的exchange
这个机制是对上面的一种补充。RabbitMQ中有多种不同类型的交换机来支持不同的路由策略。每个信道都会被分配一个唯一的ID。如果消费者的数量很大，这就会导致性能低下。
另外，死信在转移到死信队列的过程中，是没有经过消息发送者确认的，所以并不能保证消息的安全性。
Haproxy是一个免费开源的负载均衡工具。这也是最近RabbitMQ功能变动最大的地方。关键的message不能因为服务出现问题而被忽略。但是，消息写入硬盘的过程中，是会阻塞队列的。
Producer消息发送给queue，多个Consumer同时往队列上消费消息。他通过Headers来进行消息路由。为了防止Haproxy单点崩溃的问题，可以引入keepalived组件来保证Haproxy的高可用。声明queue
step4、这个并不是RabbitMQ特有的概念，但是却是几乎所有MQ产品通用的一个概念。生产者发送的消息通过Exchange交换机分配到各个不同的Queue队列上，而对于消息消费者来说，通常只需要关注自己感兴趣的队列就可以了。如果一个queue在开始时被声明为durable，那在后面再次声明这个queue时，即使声明为 not durable，那这个queue的结果也还是durable的。而在使用这种模式时，消费者应该尽量的连接上每一个节点，减少消息在集群中的传输。当然，如果配置了死信队列的话，就会进入对应的死信队列。并且这种经典队列不适合积累太多的消息。发布单条消息即发布一条消息就确认一条消息。
死信产生
有以下三种情况，RabbitMQ会将一个正常消息转成死信
- 消息被消费者确认拒绝。死信队列也是一个普通的队列，同样可以在队列上声明消费者，继续对消息进行消费处理。另外也还有很多新的MQ产品体现了更强大的竞争力，比如Pulsar。**而对于应用程序，自始至终只需要访问keepalived暴露出来的VIP，感知不到VIP漂移的过程。但是如果配置了x-dead-letter-routing-key这个参数的话，routingkey就会被替换为配置的这个值。不持久等高级队列更能要求更严格的场景。
3、Replay/Time-travelling 消息回溯
RabbitMQ已有的这些队列类型，在消费者处理完消息后，消息都会从队列中删除，因此，无法重新读取已经消费过的消息。Queue结构天生就具有FIFO的顺序，消息最终都会被分发到不同的Queue当中，然后才被消费者进行消费处理。
后面的Arguments部分，还有非常多的参数，可以点击后面的问号逐步了解。
**Quorum队列更适合于 队列长期存在，并且对容错、而他的弊端也是明显的，集群内部的网络带宽会被这种同步通讯大量的消耗，进而降低整个集群的性能。如果消息需要重新投递，就需要再次入队。最为常用的是经典队列Classic。
经典队列可以选择是否持久化(Durability)以及是否自动删除(Auto delete)两个属性。RocketMQ、然后，就可以通过设置 Delivery limit参数来定制一个毒消息的删除策略。这种队列类型的消息是持久化到磁盘并且具备分布式备份的，更适合于消费者多，读消息非常频繁的场景。
x-dead-letter-exchange: mirror.dlExchange 对应的死信交换机x-dead-letter-routing-key: mirror.messageExchange1.messageQueue1 死信交换机routing-keyx-message-ttl: 3000 消息过期时间durable: true 持久化，这个是必须的。Quorum仲裁队列
仲裁队列，是RabbitMQ从3.8.0版本，引入的一个新的队列类型，整个3.8.X版本，也都是在围绕仲裁队列进行完善和优化。应用程序只需要访问haproxy的服务端口，Haproxy会将请求以负载均衡的方式转发到后端的RabbitMQ服务上。这种集群模式下，集群的各个节点之间只会有相同的元数据，即队列结构，而消息不会进行冗余，只存在一个节点中。
2、
5、从Web管理界面就能看到，在每个虚拟主机中，RabbitMQ都会默认创建几个不同类型的交换机来。
关键处就是type为”fanout” 的exchange,这种类型的exchange只负责往所有已绑定的队列上发送消息。当键值对匹配时，对应的消费者就能接收到消息。channel.basicQos(prefetchCount);表示当前这个consumer可以同时处理几个message。
Quorum是基于Raft一致性协议实现的一种新型的分布式消息队列，他实现了持久化，多备份的FIFO队列，主要就是针对RabbitMQ的镜像模式设计的。
并且，这种集群模式也不支持高可用，即当某一个节点服务挂了后，需要手动重启服务，才能保证这一部分消息能正常消费。
从整体功能上来说，Quorum队列是在Classic经典队列的基础上做减法。Throughput Performance 高吞吐性能
Strem队列的设计以性能为主要目标，对消息传递吞吐量的提升非常明显。通过Federation可以搭建一个单向的数据同步通道。
有了haproxy后，如果某一个RabbitMQ服务崩溃了，Haproxy会将请求往另外一个RabbitMQ服务转发，这样应用程序就不需要做IP切换了。RabbitMQ提供的Sharding插件，就是一个可选的方案。这样服务器在进行消息发送前，会检查这个consumer当前正在处理中的message(message已经发送，但是未收到consumer的basicAck)有几个，如果超过了这个consumer节点的能力值，就不再往这个consumer发布。但是同时，由于需要有更多的IO操作，所以生产和消费消息的性能，相比Transient会比较低。因此，Stream也是在视图解决一个RabbitMQ一直以来，让人诟病的缺点，就是当队列中积累的消息过多时，性能下降会非常明显的问题。但是很显然，懒队列的方式属于治标不治本。
上面的懒队列其实就是针对这个问题的一种解决方案。
receiver: //将消费的目标队列绑定到exchange上。
Routing 基于内容的路由
type为”direct” 的exchange
在上一章 exchange 往所有队列发送消息的基础上，增加一个路由配置，指定exchange如何将不同类别的消息分发到不同的queue上。
关于参数x-dead-letter-routing-key
死信在转移到死信队列时，他的Routing key 也会保存下来。但是发送者发送消息是否成功是没有保证的。channel.basicReject或者channel.basicNack。这时就可以考虑使用RabbitMQ的Federation插件，搭建联邦队列Federation。Kafka对比
特性MQ ActiveMQ RabbitMQ RocketMQ Kafka 生产者消费者模式 支持 支持 支持 支持 发布订阅模式 支持 支持 支持 支持 请求回应模式 支持 支持 不支持 不支持 Api完备性 高 高 高 高 多语言支持 支持 支持 java 支持 单机吞吐量 万级 万级 万级 十万级 消息延迟 无 微秒级 毫秒级 毫秒级 可用性 高（主从） 高（主从） 非常高（分布式） 非常高（分布式） 消息丢失 较低 低 理论上不会丢失 理论上不会丢失 文档的完备性 高 高 较高 高 提供快速入门 有 有 有 有 社区活跃度 高 高 中 高 商业支持 无 无 商业云 商业云 这里的优缺点并不是绝对的，因为每个产品都在不管演进。但是Stream队列的设计目标就是以最小的内存开销高效地存储大量的数据。
这个模式应该是最常用的模式，也是官网讨论比较详细的一种模式，所以官网上也对这种模式做了重点讲述。这种队列都依靠各个Broker自己进行管理，在分布式场景下，管理效率是不太高的。这些读消息应该有保障机制进行标记并及时删除。但是，不管访问哪个服务，如果这个服务崩溃了，虽然RabbitMQ集群不会丢失消息，另一个服务也可以正常使用，但是客户端还是需要主动切换访问的服务地址。而他们改进后的方案，是consumer可以向服务器声明一个prefetchCount，我把他叫做预处理能力值。keepalived防止haproxy单点崩溃
Haproxy保证了RabbitMQ的服务高可用，防止RabbitMQ服务单点崩溃对应用程序的影响。引入keepalived后，可以将VIP先绑定在已有的Haproxy服务上，然后引入一个从Haproxy作为一个备份**。Exclusivity表示独占队列，即表示队列只能由声明该队列的Connection连接来进行使用，包括队列创建、
2、单词之间用,隔开，* 代表一个具体的单词。而这个模块就是通过给发送者提供一些确认机制，来保证这个消息发送的过程是成功的。从而给整个集群提供高可用能力。比如内部系统之间的服务调用。删除、这个headers是一个键值对，发送者可以在发送的时候定义一些键值对，接受者也可以在绑定时定义自己的键值对。
Auto delete属性如果选择为是，那队列将在至少一个消费者已经连接，然后所有的消费者都断开连接后删除自己。
懒队列的设计目标是为了支持非常长的队列(数百万级别)。
RabbitMQ为了减少性能开销，也会在一个Connection中建立多个Channel，这样便于客户端进行多线程连接，这些连接会复用同一个Connection的TCP通道，所以在实际业务中，对于Connection和Channel的分配也需要根据实际情况进行考量。我们可以回顾下，发送者发送消息的基础API：Producer.basicPublish方法是没有返回值的，也就是说，一次发送消息是否成功，应用是不知道的，这在业务上就容易造成消息丢失。
4、
3、Large logs 大日志
RabbitMQ一直以来有一个让人诟病的地方，就是当队列中积累的消息过多时，性能下降会非常明显。而Stream队列允许任意数量的消费者使用同一个队列的消息，从而消除绑定多个队列的需求。
- 首先。这也就是 Poison Message(毒消息)
- 其次，官方特意提到的message的持久性。
2、如果队列中积累的消息太多了，会严重影响客户端生产消息以及消费消息的性能。所谓毒消息是指消息一直不能被消费者正常消费(可能是由于消费者失败或者消费逻辑有问题等)，就会导致消息不断的重新入队，这样这些消息就成为了毒消息。将消息以队列的形式存储起来，并且在不同的应用程序之间进行传递，这就成了MessageQueue。 另外，如果消息没有做持久化，重启就消息就会丢失。这意味着在正常操作的大多数情况下，RAM中要保存的消息要少得多。
2、如上图，就是把producer发送的消息，交由exchange同时发送到两个queue里，然后由不同的Consumer去进行消费。这种方式可以保证消息在集群内部不会丢失。消息以FIFO先进先出的方式存入队列。在单机环境中，拥有比较高的消息可靠性。收发消息等，并且独占队列会在声明该队列的Connection断开后自动删除。
MQ：MessageQueue，消息队列。例如，某大型企业，可能在北京机房和长沙机房分别搭建RabbitMQ服务，然后希望长沙机房需要同步北京机房的消息，这样可以让长沙的消费者服务可以直接连接长沙本地的RabbitMQ，而不用费尽周折去连接北京机房的RabbitMQ服务。
其中有个特例就是Poison Message handling(处理有毒的消息)。发送批量消息之前单条确认的机制会对系统的吞吐量造成很大的影响，所以稍微中和一点的方式就是发送一批消息后，再一起确认。比如Kafka现在基本可以做到数据不丢失。类似的工具还有很多，比如F5，nginx等。
3、需要在搭建了普通集群之后再补充搭建。这种队列类似于RocketMQ当中的DLedger集群。
RabbitMQ考虑了两种集群模式：
默认的普通集群模式：
这种模式使用Erlang语言天生具备的集群方式搭建。比如Non-durable queues表示是非持久化的内存队列。
type为"topic" 的exchange
这个模式也就在上一个模式的基础上，对routingKey进行了模糊匹配。虚拟主机 virtual host
RabbitMQ出于服务器复用的想法，可以在一个RabbitMQ集群中划分出多个虚拟主机，每一个虚拟主机都有全套的基础服务组件，可以针对每个虚拟主机进行权限以及数据分配。通常情况下，第三种异步确认机制的性能是最好的。仲裁队列相比Classic经典队列，在分布式环境下对消息的可靠性保障更高。
对于RabbitMQ同样，针对单个队列，如何增加吞吐量呢？ 消费者并不能对消息增加消费并发度，所以，RabbitMQ的集群机制并不能增加单个队列的吞吐量。这种模式下，队列数量最好不要过多。因为如果其中有个节点服务宕机了，那这个节点上的数据就无法消费了，需要等到这个节点服务恢复后才能消费，而这时，消费者端已经消费过的消息就有可能给不了服务端正确应答，服务起来后，就会再次消费这些消息，造成这部分消息重复消费。如果Haproxy服务崩溃了，整个应用程序就完全无法访问RabbitMQ了。信道 Channel
一旦客户端与RabbitMQ建立了连接，就会分配一个AMQP信道 Channel。其本质区别在于，这种模式会在镜像节点中间主动进行消息同步，而不是在客户端拉取消息时临时同步。
发送者确认模式默认是不开启的，所以如果需要开启发送者确认模式，需要手动在channel中进行声明。数据安全方面的要求比低延迟、消息：在不同应用程序之间传递的数据。当然，这是以增加磁盘IO为代价的。
这种模式的消息可靠性更高，因为每个节点上都存着全量的消息。消费者把requeue参数设置为true(false)，并且在消费后，向RabbitMQ返回拒绝。Quorum队列会持续跟踪消息的失败投递尝试次数，并记录在"x-delivery-count"这样一个头部参数中。 Durable表示队列会将消息保存到硬盘，这样消息的安全性更高。Headers类型的Exchange就是一种忽略routingKey的路由方式。
但是，这其中也要注意一个很常见的BUG，就是如果所有的consumer都忘记调用basicAck()了，就会造成message被不停的分发，也就造成不断的消耗系统资源。RabbitMQ的Stream队列就是模拟Kafka的实现机制，消息吞吐量也提升了非常多。而Stream队列允许用户在日志的任何一个连接点开始重新读取数据。
默认情况下，RabbitMQ接收到消息时，会保存到内存以便使用，同时把消息写到硬盘。而接下来，死信交换机就可以像普通交换机一样，通过RoutingKey将消息转发到对应的死信队列中。**例如 电商系统的订单，引入MQ后，处理速度可以慢一点，但是订单不能丢失。
1、消费时，如果消费的不是存有数据的节点， RabbitMQ会临时在节点之间进行数据传输，将消息从存有数据的节点传输到消费的节点。使用Stream队列可以比较轻松的在队列中积累百万级别的消息。简单理解就是quorum队列中的消息需要有集群中多半节点同意确认后，才会写入到队列中。简单理解，他是RabbitMQ对于未能正常消费的消息进行的一种补救机制。
keepalived是一个搭建高可用服务的常见工具。完成以后关闭连接，释放资源
Consumer主要有两种消费方式：
被动消费模式
Consumer等待rabbitMQ 服务器将message推送过来再消费。
在这里，x-dead-letter-exchange指定一个交换机作为死信交换机，然后x-dead-letter-routing-key指定交换机的RoutingKey。他们的部署也不麻烦，就是下载+配置+运行即可。这时要如何进行数据同步呢？搭建一个跨度这么大的内部子网显然就不太划算。
懒队列会尝试尽可能早的把消息写到硬盘中。还要注意，官方特意提到，所有的queue是不能被多次定义的。
为了防止这种情况情况，可以在RabbitMQ之前部署一个Haproxy，这是一个TCP负载均衡工具。
ActiveMQ、官方文档中表示，未来会使用Quorum仲裁队列代替传统Classic队列。
这是RabbitMQ最基础也是最常用的一种工作机制。
RabbitMQ的消息可靠性是非常高的，但是他以往的机制都是保证消息发送到了MQ之后，可以推送到消费者消费，不会丢失消息。
2、Queue不需要Exchange也可以独立工作，只不过通常在业务场景中，会增加Exchange实现更复杂的消息分配策略。
producer: //只负责往exchange里发消息，后面的事情不管。队列 Queue
Queue是实际保存数据的最小单位。但是如何在消费者的处理能力有限的前提下提升消费者的消费速度呢？RabbitMQ提供的Sharding插件，就提供了一种思路。当消息的重复投递次数超过了Delivery limit参数阈值时，RabbitMQ就会删除这些毒消息。
另一种是主动消费模式
Comsumer主动到rabbitMQ服务器上去拉取messge进行消费。
这是RabbitMQ最为经典的队列类型。消息发送到RabbitMQ中后，会首先进入一个交换机，然后由交换机负责将数据转发到不同的队列中。
4、
镜像模式：
这种模式是在普通集群模式基础上的一种增强方案，这也就是RabbitMQ的官方HA高可用方案。Haproxy反向代理
有了镜像集群之后，客户端应用就可以访问RabbitMQ集群中任意的一个节点了。
step1、
在企业中有很多大型的分布式场景，在这些业务场景下，希望服务也能够同样进行分布式部署。
Haproxy+Keeperalived的组合是分布式场景中经常用到的一种高可用方案。这里，RabbitMQ默认是采用的fair dispatch，也叫round-robin模式，就是把消息轮询，在所有consumer中轮流发送。producer只负责发送消息，至于消息进入哪个queue，由exchange来分配。
在RabbitMQ中，经典队列是一种非常传统的队列结构。真正要提升RabbitMQ单队列的吞吐量，还是要从数据也就是消息入手，只有将数据真正的分开存储才行。也可以理解为是客户端与RabbitMQ实际进行数据交互的通道，我们后续的大多数的数据操作都是在信道 Channel 这个层面展开的。同时，Quorum是以牺牲很多高级队列特性为代价，来进一步保证消息在分布式环境下的高可靠。不同虚拟主机之间是完全隔离的，如果不考虑资源分配的情况，一个虚拟主机就可以当成一个独立的RabbitMQ服务使用。
Lazy Queue懒队列机制提升了消息在RabbitMQ中堆积的能力，但是最终，消息还是需要消费者处理消化。
对于死信队列，在RabbitMQ中主要涉及到几个参数。
并且在集群内部有一个算法会选举产生master和slave，当一个master挂了后，也会自动选出一个来。
在官网的示例中，重点解释了三种策略：
1、当然，这并不是我们的重点，只要了解即可。
死信队列是RabbitMQ中非常重要的一个特性。
step7、
其中，Durability有两个选项，Durable和Transient。如果你熟悉Kafka或者RocketMQ，会对这种日志记录消息的方式非常熟悉。
Stream队列是RabbitMQ自3.9.0版本开始引入的一种新的数据队列类型。一般是启一个一直挂起的线程来等待。
1、首先创建连接，获取Channel
step2、未来如果需要搭建集群，就需要通过这些Broker来构建。RabbitMQ虽然针对写入硬盘速度做了很多算法优化，但是在长队列中，依然表现不是很理想，所以就有了懒队列的出现。这样就保证了Haproxy服务的高可用性。另一种是any，表示只要满足其中一个键值就可以了。
对队列进行配置时，只有Classic经典队列和Quorum仲裁队列才能配置死信队列，而目前Stream流式队列，并不支持配置死信队列。
整体上来说，RabbitMQ的Stream队列，其实有很多地方借鉴了其他MQ产品的优点，在保证消息可靠性的基础上，着力提高队列的消息吞吐量以及消息转发性能。异步确认消息实现的方式也比较简单，Producer在channel中注册监听器来对消息进行确认。
交换机多用来与生产者打交道。RabbitMQ、
官网示例中的集中路由策略， direct,fanout,topic等这些Exchange，都是以routingkey为关键字来进行消息路由的，但是这些Exchange有一个普遍的局限就是都是只支持一个字符串的形式，而不支持其他形式。因此，经典队列主要用在数据量比较小，并且生产消息和消费消息的速度比较稳定的业务场景。RabbitMQ以往更专注于企业级的内部使用，但是从这些队列功能可以看到，Rabbitmq也在向更复杂的互联网环境靠拢，未来对于RabbitMQ的了解，也需要随着版本推进，不断更新。这样即可以提高数据的安全性，也能够提升消息读取的性能。
很显然，这种集群模式的消息可靠性不是很高。声明Exchange-可选
step3、声明Exchange与Queue的绑定关系-可选
step5、这种队列提供了RabbitMQ已有的其他队列类型不太好实现的四个特点：
1、连接 Connection
客户端与RabbitMQ进行交互，首先就需要建立一个TCP连接，这个连接就是Connection。RabbitMQ 3.8.X版本添加了Quorum队列，3.9.X又添加了Stream队列。此时，对于RabbitMQ来说，整个集群的服务是稳定的。队列可能会因为一些原因变得非常长-也就是数据堆积。匹配的方式有两种，一种是all，表示需要所有的键值对都满足才行。 当主Haproxy服务出现异常后，keepalived可以将虚拟IP转为绑定到从Haproxy服务的网卡上，这个过程称为VIP漂移。
谈到Sharding，你是不是就想到了分库分表？对于数据库的分库分表，分库可以减少数据库的IO性能压力，而真正要解决单表数据太大的问题，就需要分表。服务主机Broker
一个搭建RabbitMQ Server的服务器称为Broker。下方有几个属性也都是来定义日志文件的大小以及保存时间。三种确认机制的区别这三种确认机制都能够提升Producer发送消息的安全性。
工作任务模式，领导部署一个任务，由下面的一个员工来处理。
RabbitMQ从3.6.0版本开始，就引入了懒队列(Lazy Queue)的概念。
4、消息被Consumer从队列中取出后就会从队列中删除。也就是把producer与Consumer进行进一步的解耦。他会真正将一个队列中的消息分散存储到不同的节点上，并提供多个节点的负载均衡策略实现对等的读与写功能。Producer根据应用场景发送消息到queue
step6、
Stream队列的核心是以append-only只添加的日志来记录消息，整体来说，就是消息将以append-only的方式持久化到日志文件中，然后通过调整每个消费者的消费进度offset，来实现消息的多次分发。既然是通道，那就需要尽量注意在停止使用时要关闭，释放资源。Consumer端的autoAck字段设置的是false,这表示consumer在接收到消息后不会自动反馈服务器已消费了message，而要改在对message处理完成了之后，再调用channel.basicAck来通知服务器已经消费了该message.这样即使Consumer在执行message过程中出问题了，也不会造成message被忽略，因为没有ack的message会被服务器重新进行投递。但是同时又带来了Haproxy的单点崩溃问题。从官网的封面就能看到，现在RabbitMQ主推的是Quorum队列。
所以这种集群模式只适合一些对消息安全性不是很高的场景。
