2023春招面试：消息中间件面试题整理

Posted 2023-01-14 编程指南针

RabbitMQ如何确保消息发送 ？ 消息接收？

• 开启生产者确认机制，确保生产者的消息能到达队列（config机制保证消息正确到达交换机、return机制保证消息正确到达队列）
• 开启持久化功能，确保消息未消费前在队列中不会丢失
• 开启消费者确认机制为auto，由spring确认消息处理成功后完成ack
• 开启消费者失败重试机制，并设置MessageRecoverer，多次重试失败后将消息投递到异常交换机，交由人工处理

RabbitMQ中什么样的消息会成为死信？

• 消费者使用basic.reject或 basic.nack声明消费失败，并且消息的requeue参数设置为false
• 消息是一个过期消息，超时无人消费
• 要投递的队列消息满了，无法投递

RabbitMQ如何实现延迟队列？项目中哪里用到了

• 死信交换机 + TTL机制
• 使用延迟交换机插件（需要安装）

延迟发布文章功能使用了延迟队列

RabbitMQ消息消费失败，且达到最大重试次数会怎么样？

默认情况下，消息会丢弃，我们可以通过MessageRecovery接口三个不同的实现来处理

• RejectAndDontRequeueRecoverer：重试耗尽后，直接reject，丢弃消息。默认就是这种方式
• ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队
• RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机

比较优雅的一种处理方案是RepublishMessageRecoverer，失败后将消息投递到一个指定的，专门存放异常消息的队列，后续由人工集中处理。

RabbitMQ如何解决消息积压问题（答案有缺陷）？

当生产者发送消息的速度超过了消费者处理消息的速度，就会导致队列中的消息堆积，直到队列存储消息达到上限。之后发送的消息就会成为死信，可能会被丢弃，这就是消息堆积问题。

解决消息堆积有两种思路：

• 增加更多消费者，提高消费速度。也就是我们之前说的work queue模式
• 采用惰性队列，接收到消息后直接存入磁盘而非内存，扩大队列容积，提高堆积上限

RabbitMQ集群有几种形式？

1、普通集群，或者叫标准集群（classic cluster）

特点：节点之间共享交换机，队列等元信息，不包含队列中的数据，队列所在节点宕机，队列中数据丢失，提高了消息处理的速度，没有提高可用性。

2、镜像集群：本质是主从模式

特点：节点之间共享所有数据，包括队列中的数据，主节点宕机后，镜像节点会替代成主节点，消息处理速度和可用性都提高了，但是主从节点之间数据同步不是强一致性。

3、仲裁队列：仲裁队列是3.8版本以后才有的新功能，用来替代镜像队列，主从同步基于Raft协议，强一致

MQ消息重复消费怎么办？

消息什么时候会重复消费？

1、消费者采用自动应答模式，提交应答请求后，由于网络问题，服务器没有收到，消息会重新入队列，再次投递

2、手动应答模式下，有与网络问题服务器没有收到消息，或者消费者收到消息后没有来得及处理宕机了

解决办法

幂等

1、消费者记录消费日志，消费消息时，如果该消息已经消费则不在处理

2、消费者视具体情况从业务上幂等，比如删除动作和redis的set动作是天生幂等不用处理，新增时先判断数据库是否已经保存过，如果保存过则不在处理，等等。

RabbitMQ 和 Emq X有什么区别

RabbitMQ 和 Emq 都是基于 Erlang 语言开发的， Emq x是遵循MQTT 协议，RabbitMQ是遵循AMQP协议，也可以通过插件使其支持MQTT协议

Rabbitmq特点如下：

可靠性∶RabbitMQ使用一些机制来保证可靠性，如持久化、传输确认及发布确认等。

灵活的路由∶在消息进入队列之前，通过交换器来路由消息。对于典型的路由功能，RabbitMQ已经提供了一些内置的交换器来实现。针对更复杂的路由功能，可以将多个交换器绑定在一起，也可以通过插件机制来实现自己的交换器。

扩展性∶多个RabbitMQ节点可以组成一个集群，也可以根据实际业务情况动态地扩展集群中节点。

高可用性∶ 队列可以在集群中的机器上设置镜像，使得在部分节点出现问题的情况下队列仍然可用。

多种协议∶ RabbitMQ除了原生支持AMQP协议，还支持STOMP、MQTT等多种消息中间件协议

多语言客户端∶ RabbitMQ几乎支持所有常用语言，比如 Java、Python、Ruby、php、C#、javascript 等。

管理界面∶RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息、集群中的节点等。

插件机制∶RabiMQ提供了许多插件，以实现从多方面进行扩展，当然也可以编写自己的插件

EMQ X 是开源百万级分布式 MQTT 消息服务器（MQTT Messaging Broker），用于支持各种接入标准

MQTT 协议的设备，实现从设备端到服务器端的消息传递，以及从服务器端到设备端的设备控制消息转

发。从而实现物联网设备的数据采集，和对设备的操作和控制。

EMQ X 主要有以下的特点：

EMQ X 支持丰富的物联网协议，包括 MQTT、MQTT-SN、CoAP、 LwM2M、LoRaWAN 和WebSocket 等；

优化的架构设计，支持超大规模的设备连接。企业版单机能支持百万的 MQTT 连接；集群能支持千万级别的 MQTT 连接；

易于安装和使用；

扩展性好，灵活的扩展性，支持企业的一些定制场景；

中国本地的技术支持服务，通过微信、QQ等线上渠道快速响应客户需求；

基于 Apache 2.0 协议许可，完全开源。EMQ X 的代码都放在 Github 中，用户可以查看所有源代码。

单机支持百万连接，集群支持千万级连接；毫秒级消息转发。EMQ X 中应用了多种技术以实现上述功能，

利用 Erlang/OTP 平台的软实时、高并发和容错（电信领域久经考验的语言）

全异步架构 连接、会话、路由、集群的分层设计 消息平面和控制平面的分离等

扩展模块和插件，EMQ X 提供了灵活的扩展机制，可以实现私有协议、认证鉴权、数据持久化、桥接转发和管理控制台等的扩展

桥接：EMQ X 可以跟别的消息系统进行对接，比如 EMQ X Enterprise 版本中可以支持将消息转发

到 Kafka、RabbitMQ 或者别的 EMQ 节点等

共享订阅：共享订阅支持通过负载均衡的方式在多个订阅者之间来分发 MQTT 消息。比如针对物联

网等数据采集场景，会有比较多的设备在发送数据，通过共享订阅的方式可以在订阅端设置多个订

阅者来实现这几个订阅者之间的工作负载均衡

MQTT 是一个非常轻量级的订阅和发布协议，现在已经是物联网领域广泛使用的传输协议。

AMQP，即Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计

EMQ如何实现延迟消费？

EMQ X 的延迟发布功能可以实现按照用户配置的时间间隔延迟发布 PUBLISH 报文的功能。当客户端使

用特殊主题前缀 $delayed/DelayInteval 发布消息到 EMQ X 时，将触发延迟发布功能。延迟发布

的功能是针对消息发布者而言的，订阅方只需要按照正常的主题订阅即可。

延迟发布主题的具体格式如下：

$delayed/DelayInterval/TopicName

$delayed : 使用 $delayed 作为主题前缀的消息都将被视为需要延迟发布的消息。延迟间隔由下一主题层级中的内容决定。

DelayInterval : 指定该 MQTT 消息延迟发布的时间间隔，单位是秒，允许的最大间隔是4294967 秒。如果 DelayInterval 无法被解析为一个整型数字，EMQ X 将丢弃该消息，客户端不会收到任何信息。

TopicName : MQTT 消息的主题名称

发布完消息后，10秒之后消费者才能消费消息

注意：$delayed写在生产方

EMQ X中如何实现共享订阅？

在emq中如果多个消费者同时订阅同一个主体，默认广播模式，所有客户端都会消费消息

另外emq中提供了两种共享订阅方式

不带群组共享订阅

以 $queue/ 为前缀的共享订阅是不带群组的共享订阅，多个消费者同时订阅，只能有一个消费者消费消息

格式为

$queue/主题名称

带群组共享订阅

以 $share/<group-name> 为前缀的共享订阅是带群组的共享订阅，多个客户端订阅相同主题，如果是同一个组，只能有一个客户端消费消息，如果是不同组则都可以消费消息

格式为

$queue/组名/主题名称

注意：无论是$queue 还是$share 都是写在消费方

消息积压怎么解决？

一、前言

在使用消息队列遇到的问题中，消息积压这个问题，应该是最常遇到的问题了，并且，这个问题 还不太好解决。

我们都知道，消息积压的直接原因，一定是系统中的某个部分出现了性能问题，来不及处理上游发送的消 息，才会导致消息积压。

所以，我们先来分析下，在使用消息队列时，如何来优化代码的性能，避免出现消息积压。然后再来看看， 如果你的线上系统出现了消息积压，该如何进行紧急处理，最大程度地避免消息积压对业务的影响。

二、优化性能来避免消息积压

在使用消息队列的系统中，对于性能的优化，主要体现在生产者和消费者这一收一发两部分的业务逻辑中。 对于消息队列本身的性能，你作为使用者，不需要太关注。为什么这么说呢?

主要原因是，对于绝大多数使用消息队列的业务来说，消息队列本身的处理能力要远大于业务系统的处理能 力。主流消息队列的单个节点，消息收发的性能可以达到每秒钟处理几万至几十万条消息的水平，还可以通 过水平扩展Broker的实例数成倍地提升处理能力。

而一般的业务系统需要处理的业务逻辑远比消息队列要复杂，单个节点每秒钟可以处理几百到几千次请求， 已经可以算是性能非常好的了。所以，对于消息队列的性能优化，我们更关注的是，在消息的收发两端，我 们的业务代码怎么和消息队列配合，达到一个最佳的性能。

1. 发送端性能优化

发送端业务代码的处理性能，实际上和消息队列的关系不大，因为一般发送端都是先执行自己的业务逻辑， 最后再发送消息。如果说，你的代码发送消息的性能上不去，你需要优先检查一下，是不是发消息之前的业 务逻辑耗时太多导致的。

对于发送消息的业务逻辑，只需要注意设置合适的并发和批量大小，就可以达到很好的发送性能。为什么这 么说呢?

之前有讲过Producer发送消息的过程，Producer发消息给Broker，Broker收到消息后返回确认 响应，这是一次完整的交互。假设这一次交互的平均时延是1ms，我们把这1ms的时间分解开，它包括了下 面这些步骤的耗时:

发送端准备数据、序列化消息、构造请求等逻辑的时间，也就是发送端在发送网络请求之前的耗时;

发送消息和返回响应在网络传输中的耗时;

Broker处理消息的时延。

如果是单线程发送，每次只发送1条消息，那么每秒只能发送 1000ms / 1ms * 1条/ms = 1000条 消息，这种 情况下并不能发挥出消息队列的全部实力。

无论是增加每次发送消息的批量大小，还是增加并发，都能成倍地提升发送性能。至于到底是选择批量发送还是增加并发，主要取决于发送端程序的业务性质。简单来说，只要能够满足你的性能要求，怎么实现方便 就怎么实现。

比如说，你的消息发送端是一个微服务，主要接受RPC请求处理在线业务。很自然的，微服务在处理每次请 求的时候，就在当前线程直接发送消息就可以了，因为所有RPC框架都是多线程支持多并发的，自然也就实 现了并行发送消息。并且在线业务比较在意的是请求响应时延，选择批量发送必然会影响RPC服务的时延。 这种情况，比较明智的方式就是通过并发来提升发送性能。

如果你的系统是一个离线分析系统，离线系统在性能上的需求是什么呢?它不关心时延，更注重整个系统的 吞吐量。发送端的数据都是来自于数据库，这种情况就更适合批量发送，你可以批量从数据库读取数据，然 后批量来发送消息，同样用少量的并发就可以获得非常高的吞吐量。

2. 消费端性能优化

使用消息队列的时候，大部分的性能问题都出现在消费端，如果消费的速度跟不上发送端生产消息的速度， 就会造成消息积压。如果这种性能倒挂的问题只是暂时的，那问题不大，只要消费端的性能恢复之后，超过 发送端的性能，那积压的消息是可以逐渐被消化掉的。

要是消费速度一直比生产速度慢，时间⻓了，整个系统就会出现问题，要么，消息队列的存储被填满无法提 供服务，要么消息丢失，这对于整个系统来说都是严重故障。

所以，我们在设计系统的时候，一定要保证消费端的消费性能要高于生产端的发送性能，这样的系统才能健 康的持续运行。

消费端的性能优化除了优化消费业务逻辑以外，也可以通过水平扩容，增加消费端的并发数来提升总体的消 费性能。特别需要注意的一点是，在扩容Consumer的实例数量的同时，必须同步扩容主题中的分区(也叫 队列)数量，确保Consumer的实例数和分区数量是相等的。

如果Consumer的实例数量超过分区数量，这 样的扩容实际上是没有效果的。原因我们之前讲过，因为对于消费者来说，在每个分区上实际上只能支持单线程消费。

我⻅到过很多消费程序，他们是这样来解决消费慢的问题的:

它收消息处理的业务逻辑可能比较慢，也很难再优化了，为了避免消息积压，在收到消息的OnMessage方法中，不处理任何业务逻辑，把这个消息放到一个内存队列里面就返回了。

然后它可以启动很多的业务线 程，这些业务线程里面是真正处理消息的业务逻辑，这些线程从内存队列里取消息处理，这样它就解决了单 个Consumer不能并行消费的问题。

这个方法是不是很完美地实现了并发消费?请注意，这是一个非常常⻅的错误方法! 为什么错误?因为会丢消息。如果收消息的节点发生宕机，在内存队列中还没来及处理的这些消息就会丢失。

三、消息积压了该如何处理?

还有一种消息积压的情况是，日常系统正常运转的时候，没有积压或者只有少量积压很快就消费掉了，但是 某一个时刻，突然就开始积压消息并且积压持续上涨。这种情况下需要你在短时间内找到消息积压的原因， 迅速解决问题才不至于影响业务。

导致突然积压的原因肯定是多种多样的，不同的系统、不同的情况有不同的原因，不能一概而论。但是，我 们排查消息积压原因，是有一些相对固定而且比较有效的方法的。

能导致积压突然增加，最粗粒度的原因，只有两种:要么是发送变快了，要么是消费变慢了。

大部分消息队列都内置了监控的功能，只要通过监控数据，很容易确定是哪种原因。如果是单位时间发送的 消息增多，比如说是赶上大促或者抢购，短时间内不太可能优化消费端的代码来提升消费性能，唯一的方法 是通过扩容消费端的实例数来提升总体的消费能力。

如果短时间内没有足够的服务器资源进行扩容，没办法的办法是，将系统降级，通过关闭一些不重要的业 务，减少发送方发送的数据量，最低限度让系统还能正常运转，服务一些重要业务。

还有一种不太常⻅的情况，你通过监控发现，无论是发送消息的速度还是消费消息的速度和原来都没什么变化，这时候你需要检查一下你的消费端，是不是消费失败导致的一条消息反复消费这种情况比较多，这种情况也会拖慢整个系统的消费速度。

如果监控到消费变慢了，你需要检查你的消费实例，分析一下是什么原因导致消费变慢。优先检查一下日志 是否有大量的消费错误，如果没有错误的话，可以通过打印堆栈信息，看一下你的消费线程是不是卡在什么 地方不动了，比如触发了死锁或者卡在等待某些资源上了。

四、小结

优化消息收发性能，预防消息积压的方法有两种，增加批量或者是增加并发，在发送端这两种方法都可以使 用，在消费端需要注意的是，增加并发需要同步扩容分区数量，否则是起不到效果的。

对于系统发生消息积压的情况，需要先解决积压，再分析原因，毕竟保证系统的可用性是首先要解决的问 题。快速解决积压的方法就是通过水平扩容增加Consumer的实例数量。