1.RabbitMQ

1.1 MQ概念

MQ（Message Queue）消息队列，是基础数据结构中“先进先出”的一种数据结构。指把要传输的数据（消息）放在队列中，用队列机制来实现消息传递——生产者产生消息并把消息放入队列，然后由消费者去处理。消费者可以到指定队列拉取消息，或者订阅相应的队列，由MQ服务端给其推送消息。

1.2 队列模型

最初的消息队列就是上面的原始模型，它是一个严格意义上的队列（Queue）。消息按照什么顺序写进去，就按照什么顺序读出来。不过，队列没有 “读” 这个操作，读就是出队，从队头中 “删除” 这个消息。

1.3发布订阅模型

如果需要将一份消息数据分发给多个消费者，并且每个消费者都要求收到全量的消息。很显然，队列模型无法满足这个需求。

一个可行的方案是：为每个消费者创建一个单独的队列，让生产者发送多份。这种做法比较笨，而且同一份数据会被复制多份，也很浪费空间。

为了解决这个问题，就演化出了另外一种消息模型：发布-订阅模型。

在发布-订阅模型中，存放消息的容器变成了 “主题”，订阅者在接收消息之前需要先 “订阅主题”。最终，每个订阅者都可以收到同一个主题的全量消息。

仔细对比下它和 “队列模式” 的异同：生产者就是发布者，队列就是主题，消费者就是订阅者，无本质区别。唯一的不同点在于：一份消息数据是否可以被多次消费。

1.4 MQ应用场景

MQ主要解决应用耦合、异步消息、流量削峰等问题。实现高性能、高可用、可伸缩和最终一致性架构。是大型分布式系统不可缺少的中间件。

1.4.1 异步

从第三方平台中接收数据，数据中包含了很多的图片，将图片保存到美团云耗时比较久，后续还有计算分数等耗时比较久的操作

异步处理后，主流程只需要100ms，其他的都通过异步的方式进行处理

1.4.2 解耦

当系统A中在订单创建后，需要通知B系统和C系统，然后B系统和C系统再做出相应的处理

此时A系统是强依赖B系统和C系统，当B系统需要下线，或者需要重新加入D系统，则需要修改代码：

如此这样反复的添加和删除依赖的系统，使得系统难以维护，此时可以通过MQ来进行解耦

这个时候A系统就与需要关心订单创建事件的系统解耦开，不再关心下游有哪些系统，也不用受下游系统可用性的影响。

1.4.3 削峰

有一个活动页面，平时大概就50qps，但每天有一个时刻11点到13点访问的人比较多，达到了1000qps，但是当前系统的处理能力为100qps。整个活动大部分时间流量都不太高，扩充太多的机器利用率太低，这个时候可以通过mq来进行削峰

1.5 技术选型

消息Broker，目前常见的实现方案就是消息队列（MessageQueue），简称为MQ.

在MQ长期发展过程中，诞生了很多MQ产品，但是有很多MQ产品都已经逐渐被淘汰了。比如早期的ZeroMQ,ActiveMQ等。目前最常用

的MQ产品包括kafka、RabbitMQ和RocketMQ。我们对这几个产品做下简单的比较，重点需要理解他们的适用场景。

几种常见MQ的对比：

追求可用性：Kafka、 RocketMQ 、RabbitMQ

追求可靠性：RabbitMQ、RocketMQ

追求吞吐能力：RocketMQ、Kafka

追求消息低延迟：RabbitMQ、Kafka

好的产品都是在不断演进的，所以对这些产品的理解也需要与时俱进。比如现在还有个MQ产品Pulsar,非常适合于大型企业内部海量的系

统调用，也体现了非常强大的竞争力。

2.RabbitMQ

2.1.环境搭建

2.1.1 介绍

RabbitMQ是流行的开源消息队列系统， 使用Erlang语言开发，是AMQP（高级消息队列协议）的标准实现，官网地址：Messaging that just works — RabbitMQ

RabbitMQ四大核心：

• 生产者
• 消费者
• 队列
• 交换机

AMQP协议是一种二进制协议，它定义了一组规则和标准，以确保消息可以在不同的应用程序和平台之间传递和解释，AMQP协议包含四个核心组件：

• 消息
• 交换机
• 队列
• 绑定

2.1.2 安装

erlang 地址： https://erlang.org/download/

erlang: 安装 https://blog.csdn.net/wcy1900353090/article/details/121294629

rabbitMQ安装方式：https://blog.csdn.net/laterstage/article/details/131522924

2.1.3 概念

• **publisher**：生产者，也就是发送消息的一方
• **consumer**：消费者，也就是消费消息的一方
• **queue**：队列，存储消息。生产者投递的消息会暂存在消息队列中，等待消费者处理
• **exchange**：交换机，负责消息路由。生产者发送的消息由交换机决定投递到哪个队列。
• **virtual host**：虚拟主机，起到数据隔离的作用。每个虚拟主机相互独立，有各自的exchange、queue

上述这些东西都可以在RabbitMQ的管理控制台来管理

2.1.4 用户管理

创建用户：

分配权限：

分配 virtual hosts：

2.1.5 架构

RabbitMQ对应的架构如图：

​ Broker:接收和分发消息的应用，RabbitMQ Server就是Message Broker

​ Virtual host:Virtual host是一个虚拟主机的概念，一个Broker中可以有多个Virtual host,每个Virtual host都有一套自己的Exchange

和Queue,同一个Virtual host中的Exchange和Queue不能重名，不同的Virtual host中的Exchange和Queue名字可以一样。这样，不同的

用户在访问同一个RabbitMQ Broker时，可以创建自己单独的Virtual host,然后在自己的Virtual host中创建Exchange和Queue,很好地做

到了不同用户之间相互隔离的效果。

​ Connection:publisher/consumer和porker之间的TCP连接

​ Channel:发送消息的通道，如果每一次访问RabbitMQ都建立一个Connection,在消息量大的时候建立TCP Connection的开销将是巨

大的，效率也较低。Channel是在connection内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的channel进

行通讯，AMQP method包含了channel id帮助客户端和message broker识别channel,所以channel之间是完全隔离的。Channel作为

轻量级的Connection极大减少了操作系统建立TCP connection的开销

​ Exchange:message到达broker的第一站，根据分发规则，匹配查询表中的routing key,分发消息到queue中去。常用的类型有：

direct (point-to-point),topic (publish-subscribe)and fanout(multicast)

​ Queue:Queue是一个用来存放消息的队列，生产者发送的消息会被放到Queue中，消费者消费消息时也是从Queue中取走消息。

​ Binding:exchange和queue之间的虚拟连接，binding中可以包含routing key,Binding信息被保存到exchange中的查询表中，用

于message的分发依据

2.2.收发消息

2.2.1.创建host

2.2.2 .创建队列

Queues选项卡，新建一个队列：

再以相同的方式，创建一个队列，密码为queue2，最终队列列表如下：

2.2.3 .交换机

此时，我们向amq.fanout交换机发送一条消息。

怎么回事呢？

发送到交换机的消息，只会路由到与其绑定的队列，因此仅仅创建队列是不够的，我们还需要将其与交换机绑定。

2.2.4.绑定关系

点击Exchanges选项卡，点击amq.fanout交换机，进入交换机详情页，然后点击Bindings菜单，在表单中填写要绑定的队列名称：

相同的方式，将queue2也绑定到改交换机。

再次点击发布：

回到Queue and Streams选项卡就能看到消息了

2.2.5.接受消息

3.选择合适的队列

3.1. Classic经典队列

之前我们一直在使用Classic经典队列。

其实在创建队列时可以看到，我们实际上是可以选择三种队列类型的，Classic经典队列，Quorum仲裁队列，Stream流式队列。

RabbitMQ自3.8.x版本推出了Quorum仲裁队列，3.9.x版本推出了Stream流式队列。这些新的队列类型都是RabbitMQ针对现代新的业务

场景做出的大的改善。最明显的，以往的RabbitMQ版本，如果消息产生大量积累就会严重影响消息收发的性能。而这两种新的队列可以

极大的提升RabbitMQ的消息堆积性能。

其中，Durability有两个选项，Durable和Transient。Durable:表示队列会将消息保存到硬盘，这样消息的安全性更高。但是同时，由于

需要有更多的IO操作，所以生产和消费消息的性能，相比Transient会比较低。

Auto delete属性如果选择为是，那队列将在至少一个消费者已经连接，然后所有的消费者都断开连接后删除自己。

后面的Arguments部分，还有非常多的参数，可以点击后面的问号逐步了解。

在RabbitMQ中，经典队列是一种非常传统的队列结构。消息以FIFO先进先出的方式存入队列。消息被Consumer从队列中取出后就会从

队列中删除。如果消息需要重新投递，就需要再次入队。

关于Classic如何持久化数据，RabbitMQ目前提供了两个实现版本。其中Version1就是将数据文件整体写入和读取，这种实现方式比较简

单，但是如果消息有积压，对服务端的压力就会比较大。另一种方式是只读取一部分索引，数据会在需要的时候再加载到内存当中。这也

就是之前版本当中提到的懒对列。这种方式在消息积压时，性能影响就会相对小一点。

这种队列都依靠各个Broker自己进行管理，在分布式场景下，管理效率是不太高的。并且这种经典队列不适合积累太多的消息。如果队列

中积累的消息太多了，会严重影响客户端生产消息以及消费消息的性能。因此，经典队列主要用在数据量比较小，并且生产消息和消费消

息的速度比较稳定的业务场景。比如内部系统之间的服务调用。

3.2 Quorum仲裁队列

仲裁队列，是RabbitMQ从3.8.0版本，引入的一个新的队列类型，也是目前官方比较推荐的一种对列类型。仲裁队列相比Classic经典队

列，在分布式环境下对消息的可靠性保障更高。官方文档中表示，未来会使用Quorum仲裁队列代替传统Classic队列。

在数据安全性方面，Quorum对列主要针对网络分区、通信失败等复杂网络情况下，可以提升数据的安全性。通常建议配合Publisher

Confirms机制使用。RabbitMQ能够保证经生产者确认过的消息，在集群内时安全的。但是，对于未经生产者确认的消息，RabbitMQ并

不能保证消息安全。

Quorum队列更适合于队列长期存在，并且对容错、数据安全方面的要求比低延迟、不持久等高级队列更能要求更严格的场景。例如电商

系统的订单，引入MQ后，处理速度可以慢一点，但是订单不能丢失。

也对应以下一些不适合使用的场景：

1、一些临时使用的队列：比如transientl 临时队列，exclusive独占队列，或者经常会修改和删除的队列。

2、对消息低延迟要求高：一致性算法会影响消息的延迟。

3、对数据安全性要求不高：Quorum队列需要消费者手动通知或者生产者手动确认。

4、队列消息积压严重：如果队列中的消息很大，或者积压的消息很多，就不要使用Quorum队列。Quorum队列当前会将所有消息始终

保存在内存中，直到达到内存使用极限。这种情况下，stream流式对列是一种比较好的选择。

3.3 Stream流式队列

Stream队列是RabbitMQ自3.9.0版本开始引入的一种新的数据队列类型。这种队列类型的消息是持久化到磁盘并且具备分布式备份的，

更适合于消费者多，读消息非常频繁的场景。

Stream队列的核心是以append-only只添加的日志来记录消息，整体来说，就是消息将以append-only的方式持久化到日志文件中，然后

通过调整每个消费者的消费进度offset,来实现消息的多次分发。

Stream不支持死信交换机，不支持处理毒消息。

这种队列提供了RabbitMQ已有的其他队列类型不太好实现的四个特点：

1、large fan-outs大规模分发

当想要向多个订阅者发送相同的消息时，以往的队列类型必须为每个消费者绑定一个专用的队列。如果消费者的数量很大，这就会导致性

能低下。而Stream队列允许任意数量的消费者使用同一个队列的消息，从而消除绑定多个队列的需求。

2、Replay/Time=travelling消息回溯

RabbitMQ已有的这些队列类型，在消费者处理完消息后，消息都会从队列中删除，因此，无法重新读取已经消费过的消息。

而Stream队列允许用户在日志的任何一个连接点开始重新读取数据。

3、Throughput Performance高吞吐性能

Stream队列的设计以性能为主要目标，对消息传递吞吐量的提升非常明显。

4、Large logs大日志

RabbitMQ一直以来有一个让人诟病的地方，就是当队列中积累的消息过多时，性能下降会非常明显。但是Stream队列的设计目标就是以

最小的内存开销高效地存储大量的数据。使用Stream队列可以比较轻松的在队列中积累百万级别的消息。

整体上来说，RabbitMQ的Streaml队列，其实有很多地方借鉴了其他MQ产品的优点，在保证消息可靠性的基础上，着力提高队列的消息

吞吐量以及消息转发性能。因此，Stream也是在试图解决一个RabbitMQ一直以来，让人诟病的缺点，就是当队列中积累的消息过多时，

性能下降会非常明显的问题。RabbitMQ以往更专注于企业级的内部使用，但是从这些队列功能可以看到，RabbitMQ也在向更复杂的互

联网环境靠拢，未来对于RabbitMQ的了解，也需要随着版本推进，不断更新。

3.4 总结

在企业中，目前用的最多的还是Classic经典队列。而从RabbitMQ的官网就能看出，RabbitMQ目前主推的是Quorum队列，甚至有传言

未来会用Quorum队列全面替代Classic经典队列。至于Stream队列，虽然已经经历了几个版本的完善修复，但是目前还是不太稳定，企

业用得还比较少

4. 死信队列

死信队列是RabbitMQ中非常重要的一个特性。简单理解，他是RabbitMQ对于未能正常消费的消息进行的一种补救机制。死信队列也是

一个普通的队列，同样可以在队列上声明消费者，继续对消息进行消费处理。

对于死信队列，在RabbitMQ中主要涉及到几个参数。

在这里，X-dead-letter-exchange:指定一个交换机作为死信交换机，然后x-dead-letter-routing-key指定交换机的RoutingKey。而接下

来，死信交换机就可以像普通交换机一样，通过RoutingKey:将消息转发到对应的死信队列中。

4.1 何时会产生死信

有以下三种情况，RabbitMQ会将一个正常消息转成死信

• 消息被消费者确认拒绝。消费者把requeue参数设置为true(false),并且在消费后，向RabbitMQ返回拒绝。

  channel.basicReject或者channel.basicNack。

• 消息达到预设的TTL时限还一直没有被消费。

• 消息由于队列已经达到最长长度限制而被丢掉

TTL即最长存活时间Time-To-Live。消息在队列中保存时间超过这个TTL,即会被认为死亡。死亡的消息会被丢入死信队列，如果没有配置

死信队列的话，RabbitMQ会保证死了的消息不会再次被投递，并且在未来版本中，会主动删除掉这些死掉的消息。

设置TTL有两种方式，一是通过配置策略指定，另一种是给队列单独声明TTL

策略配置方式-Wb管理平台配置或者使用指令配置60000为毫秒单位

4.2 关于参数x-dead-letter-routing-key

死信在转移到死信队列时，他的Routing key也会保存下来。但是如果配置了x-dead-letter-routing-key这个参数的话，routingkey就会被

替换为配置的这个值。

另外，死信在转移到死信队列的过程中，是没有经过消息发送者确认的，所以并不能保证消息的安全性。

4.3 如何确定一个消息是不是死信

消息被作为死信转移到死信队列后，会在Header当中增加一些消息。在官网的详细介绍中，可以看到很多内容，比如时间、原因

(rejected,expired,maxlen)、队列等。然后headerr中还会加上第一次成为死信的三个属性，并且这三个属性在以后的传递过程中都不会

更改。

• x-first-death-reason
• x-first-death-queue
• x-first-death-exchange

4.5 基于死信队列实现延迟队列

其实从前面的配置过程能够看到，所谓死信交换机或者死信队列，不过是在交换机或者队列之间建立一种死信对应关系，而死信队列可以

像正常队列一样被消费。他与普通队列一样具有FFO的特性。对死信队列的消费逻辑通常是对这些失效消息进行一些业务上的补偿。

RabbitMQ中，是不存在延迟队列的功能的，而通常如果要用到延迟队列，就会采用TTL+死信队列的方式来处理。

RabbitMQ提供了一个rabbitmq_delayed_message_.exchange:插件，可以实现延迟队列的功能，但是并没有集成到官方的发布包当中，

需要单独去下载。