RabbitMQ直播讲义（讲师：Old Jia）

RabbitMQ直播讲义（讲师：Old Jia）

第一部分内容回顾

第二部分作业讲解

1 任务一：消息中间件概述

1.1、消息中间件选型原则包括：（ABCDE）

A、开源

B、流行

C、消息的可靠性

D、横向扩展

E、性能

1.2、RabbitMQ的优缺点不包括：（E）

A、轻量级、快速部署，使用方便

B、灵活的路由配置

C、支持大多数编程语言

D、大量消息堆积，性能急剧下降

E、RabbitMQ吞吐量比Kafka高

1.3、不是消息中间件使用场景的是：（D）

A、消息的异步处理

B、系统的解耦

C、高并发流量削峰

D、热点数据缓存

任务二：RabbitMQ架构与实战

2.1、交换器类型有哪些？（C）

A、fanout类型交换器

B、direct类型交换器

C、Handler类型交换器

D、topic类型交换器

2.2、不是RabbitMQ组件的是：（A）

A、生产者、消费者

B、绑定

C、交换器

D、消息队列

2.3、下述命令使用错误的是：（AB）

A、rabbitmqctl enable rabbitmq_management

B、rabbitmqctl set_permission —vhost hello zhangsan “.“ “.“ “.*”

C、rabbitmqctl set_user_tags zhangsan policymaker

D、rabbitmq-plugins —list —enabled

2.4、RabbitMQ的工作模式有哪些？（ABCE）

A、Work Queue

B、Publish/Subscribe

C、Routing

D、Publisher Confirms

E、RPC

任务三：RabbitMQ高级特性解析

3.1、如何保证消息可靠性？（ABCDE）

A、异常捕获机制

B、使用消息事务

C、生产者确认机制

D、消费者确认机制

E、持久化存储机制

3.2、关于TTL机制说法不正确的是：（ABD）

A、RabbitMQ支持消息的过期时间，只能给消息指定过期时间

B、java中同过x-message-ttl属性指定消息的过期时间

C、消息队列通过x-expires设置过期时间

D、如果TTL设置为0，则消息永不过期

3.3、消息在什么情况下会变为死信？（ABC）

A、消息过期

B、队列达到最大长度

C、消息被拒绝并设置requeue为false

D、消息被消费

3.4、是延时交换器工作步骤的是：（ABC）

A、生产者将消息和路由键发送到指定的延时交换器上

B、延时交换器存储消息，等待消息到期，根据路由键将消息发送到绑定的队列中

C、队列把消息发送给消费者

D、RabbitMQ中启用自带的延时交换器。

任务四：RabbitMQ集群与运维

4.1、关于镜像队列说法正确的是：（ABC）

A、每个镜像队列包含一个master，若干个镜像

B、master存在于称为master的节点上

C、所有的操作都是首先对master执行，然后master将操作广播到镜像

D、镜像队列提供了高可用和负载均衡

4.2、关于HAProxy负载均衡说法正确的是：（AC）

A、HAProxy提供VIP供上游应用直接连接

B、HAProxy用于防止RabbitMQ的单点故障

C、HAProxy支持四层、七层负载均衡。

D、可以使用KeepAlived替换HAProxy做负载均衡。

4.3、下列属于RabbitMQ集群架构的是：（ABC）

A、Warren模式

B、镜像队列模式

C、Federation模式

D、两地三中心

任务五：RabbitMQ源码解析

5.1、消息发送流程包括哪些？（ABCD）

A、生产者连接RabbitMQ，建立TCP连接（Connection），声明通道（Channel）

B、声明交换器，设置相关属性

C、声明队列，设置相关属性，并使用绑定键绑定交换器和队列

D、生产者发送消息到RabbitMQ，消息中包含路由键。

5.2、如何推送消息？（ABCD）

A、使用Channel的basicConsume方法，并设置回调用于接收消息

B、等待RabbitMQ Broker 回应并投递相应队列中的消息，消费者接收消息。

C、消费者确认( ack) 接收到的消息。

D、RabbitMQ 从队列中删除相应己经被确认的消息。

5.3、RabbitMQ启动过程包括：（ABCD）

A、check erlang的版本

B、启动rabbit_sup，并没有启动任何child

C、注册rabbit这个进程

D、构造boot steps有向无环图，并按顺序启动boot steps

实操

基于RabbitMQ的TTL以及死信队列，使用SpringBoot实现延迟付款，手动补偿操作。

1、用户下单后展示等待付款页面

2、在页面上点击付款的按钮，如果不超时，则跳转到付款成功页面

3、如果超时，则跳转到用户历史账单中查看因付款超时而取消的订单。

第三部分大厂笔试题

2.1 上千万条消息在RabbitMQ中积压了几个小时还没解决：

先修复consumer的问题，确保其恢复消费速度，然后将现有consumer都停掉；
新建⼀个topic，partition是原来的10倍，临时建⽴好原先10倍或者20倍的queue数量；
然后写⼀个临时的分发数据的consumer程序，这个程序部署上去消费积压的数据；消费之后不做耗时的处理，直接均匀轮询写⼊临时建⽴好的10倍数量的queue；
接着临时征⽤10倍的机器来部署consumer，每⼀批consumer消费⼀个临时queue的数据；
这种做法相当于是临时将queue资源和consumer资源扩⼤10倍，以正常的10倍速度来消费数据；
等快速消费完积压数据之后，得恢复原先部署架构，重新⽤原先的consumer机器来消费消息。

总结：

修复并停掉consumer；
新建⼀个topic，partition是原来的10倍，建⽴临时queue，数量是原来的10倍或20倍；
写临时consumer程序，临时征⽤10倍的机器去消费数据；
消费完成之后，恢复原先consumer；

2.2、rabbitmq设置过期时间，部分消息丢失：

采取批量重导⽅法：将丢失的那批数据查询导⼊到mq⾥⾯。

2.3、RabbitMQ 上的⼀个 queue 中存放的 message 是否有数量限制？

可以认为是⽆限制，因为限制取决于机器的内存，但是消息过多会导致处理效率的下降。

2.4、分布式部署：

RabbitMQ⽆法容忍不同数据中⼼之间⽹络延迟，但是可以通过3种⽅式实现分布式部署：Federation和Shovel。

2.5、如何确保消息正确地发送⾄RabbitMQ？

RabbitMQ使⽤发送⽅确认模式，确保消息正确地发送到RabbitMQ。

发送⽅确认模式：将信道设置成confirm模式（发送⽅确认模式），则所有在信道上发布的消息

都会被指派⼀个唯⼀的ID。⼀旦消息被投递到⽬的队列后，或者消息被写⼊磁盘后（可持久化

的消息），信道会发送⼀个确认给⽣产者（包含消息唯⼀ID）。如果RabbitMQ发⽣内部错误

从⽽导致消息丢失，会发送⼀条nack（not acknowledged，未确认）消息。

发送⽅确认模式是异步的，⽣产者应⽤程序在等待确认的同时，可以继续发送消息。当确认消息到达⽣产者应⽤程序，⽣产者应⽤程序的回调⽅法就会被触发来处理确认消息。

2.6、如何确保消息接收⽅消费了消息？

接收⽅消息确认机制：消费者接收每⼀条消息后都必须进⾏确认（消息接收和消息确认是两个不同操作）。只有消费者确认了消息，RabbitMQ才能安全地把消息从队列中删除。

这⾥并没有⽤到超时机制，RabbitMQ仅通过Consumer的连接中断来确认是否需要重新发送消息。也就是说，只要连接不中断，RabbitMQ给了Consumer⾜够⻓的时间来处理消息。

特殊情况：

如果消费者接收到消息，在确认之前断开了连接或取消订阅，RabbitMQ会认为消息没有被分发，然后重新分发给下⼀个订阅的消费者。（可能存在消息重复消费的隐患，需要根据bizId去重）
如果消费者接收到消息却没有确认消息，连接也未断开，则RabbitMQ认为该消费者繁忙，将不会给该消费者分发更多的消息。

2.7、如何避免消息重复投递或重复消费？

在消息⽣产时，MQ内部针对每条⽣产者发送的消息⽣成⼀个inner-msg-id，作为去重和幂等的依据（消息投递失败并重传），避免重复的消息进⼊队列；在消息消费时，要求消息体中必须要有⼀个bizId（对于同⼀业务全局唯⼀，如⽀付ID、订单ID、帖⼦ID等）作为去重和幂等的依据，避免同⼀条消息被重复消费。

2.8、消息基于什么传输？

由于TCP连接的创建和销毁开销较⼤，且并发数受系统资源限制，会造成性能瓶颈。

RabbitMQ使⽤信道的⽅式来传输数据。信道是建⽴在真实的TCP连接内的虚拟连接，且每条

TCP连接上的信道数量没有限制。

RabbitMQ采⽤类似NIO（Non-blocking I/O）做法，选择TCP连接复⽤，不仅可以减少性能开销，同时也便于管理。
每个线程把持⼀个信道，所以信道服⽤了Connection的TCP连接。同时RabbitMQ可以确保每个线程的私密性，就像拥有独⽴的连接⼀样。

2.9、消息如何分发？

若该队列⾄少有⼀个消费者订阅，消息将以循环（round-robin）的⽅式发送给消费者。每条消息只会分发给⼀个订阅的消费者（前提是消费者能够正常处理消息并进⾏确认）。

2.10、消息怎么路由？

从概念上来说，消息路由必须有三部分：交换器、路由、绑定。⽣产者把消息发布到交换器上；绑定决定了消息如何从交换器路由到特定的队列；消息最终到达队列，并被消费者接收。

消息发布到交换器时，消息将拥有⼀个路由键（routing key），在消息创建时设定。
通过队列路由键，可以把队列绑定到交换器上。
消息到达交换器后，RabbitMQ会将消息的路由键与队列的路由键进⾏匹配（针对不同的交换器有不同的路由规则）。
如果能够匹配到队列，则消息会投递到相应队列中；如果不能匹配到任何队列，消息将进⼊ “⿊洞”。

2.11、如何确保消息不丢失？

消息持久化的前提是：将交换器/队列的durable属性设置为true，表示交换器/队列是持久交换器/队列，在服务器崩溃或重启之后不需要重新创建交换器/队列（交换器/队列会⾃动创建）。

如果消息想要从Rabbit崩溃中恢复，那么消息必须：

在消息发布前，通过把它的 “投递模式” 选项设置为2（持久）来把消息标记成持久化
将消息发送到持久交换器
消息到达持久队列

RabbitMQ确保持久性消息能从服务器重启中恢复的⽅式是，将它们写⼊磁盘上的⼀个持久化⽇志⽂件，当发布⼀条持久性消息到持久交换器上时，Rabbit会在消息提交到⽇志⽂件后才发送响应（如果消息路由到了⾮持久队列，它会⾃动从持久化⽇志中移除）。⼀旦消费者从持久队列中消费了⼀条持久化消息，RabbitMQ会在持久化⽇志中把这条消息标记为等待垃圾收集。如果持久化消息在被消费之前RabbitMQ重启，那么Rabbit会⾃动重建交换器和队列（以及绑定），并重播持久化⽇志⽂件中的消息到合适的队列或者交换器上。

2.12、使⽤RabbitMQ有什么好处？

应⽤解耦（系统拆分）
异步处理（预约挂号业务处理成功后，异步发送短信、推送消息、⽇志记录等，可以⼤⼤减⼩响应时间）
消息分发
流量削峰：将请求发送到队列中，短暂的⾼峰期积压是允许的。
消息缓冲

2.13、消息队列有什么缺点？

系统可⽤性降低：消息队列出问题影响业务；
系统复杂性增加：加⼊消息队列，需要考虑很多⽅⾯的问题，⽐如：⼀致性问题、如何保证消息不被重复消费、如何保证消息可靠性传输等。

2.14、MQ如何选型？

Rabbit回顾 - 图1

中⼩型公司⾸选RabbitMQ：管理界⾯简单，⾼并发。
⼤型公司可以选择RocketMQ：更⾼并发，可对rocketmq进⾏定制化开发。
⽇志采集功能，⾸选kafka，专为⼤数据准备。

2.15、如何保证消息队列⾼可⽤？

1. 集群：

Rabbit回顾 - 图2

集群可以扩展消息通信的吞吐量，但是不会备份消息，备份消息要通过镜像队列的⽅式解决。
队列存储在单个节点、交换器存储在所有节点。

2. 镜像队列：将需要消费的队列变为镜像队列，存在于多个节点，这样就可以实现RabbitMQ

的HA⾼可⽤性。作⽤就是消息实体会主动在镜像节点之间实现同步，⽽不是像普通模式那样，

在consumer消费数据时临时读取。缺点就是，集群内部的同步通讯会占⽤⼤量的⽹络带宽。

Rabbit回顾 - 图3

2.16、如何保证消息的顺序性？

通过某种算法，将需要保持先后顺序的消息放到同⼀个消息队列中(kafka中就是partition,rabbitMq中就是queue)。然后只⽤⼀个消费者去消费该队列。
可以在消息体内添加全局有序标识来实现。

2.17、使用RabbitMQ增加rest服务吞吐量。

Rabbit回顾 - 图4

2.18、RabbitMQ交换器有哪些类型？

fanout交换器：它会把所有发送到该交换器的消息路由到所有与该交换器绑定的队列中；
direct交换器：direct类型的交换器路由规则很简单，它会把消息路由到哪些BindingKey和RoutingKey完全匹配的队列中；
topic交换器：匹配规则⽐direct更灵活。
headers交换器：根据发送消息内容的headers属性进⾏匹配（由于性能很差，不实⽤）

常⽤的交换器主要分为以下三种：

direct：如果路由键完全匹配，消息就被投递到相应的队列
fanout：如果交换器收到消息，将会⼴播到所有绑定的队列上
topic：可以使来⾃不同源头的消息能够到达同⼀个队列。使⽤topic交换器时，可以使⽤通配符，⽐如：“*” 匹配特定位置的任意⽂本， “.” 把路由键分为了⼏部分，“#” 匹配所有规则等。特别注意：发往topic交换器的消息不能随意的设置选择键（routing_key），必须是由”.”隔开的⼀系列的标识符组成

2.19、RabbitMQ如何保证数据⼀致性？

⽣产者确认机制：消息持久化后异步回调通知⽣产者，保证消息已经发出去；
消息持久化：设置消息持久化；
消费者确认机制：消费者成功消费消息之后，⼿动确认，保证消息已经消费。

2.20、RabbitMQ消费者自动扩展数量

SimpleMessageListenerContainer可根据RabbitMQ消息堆积情况⾃动扩展消费者数量。

2.21、RabbitMQ结构

Rabbit回顾 - 图5

Broker：简单来说就是消息队列服务器实体。
Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。
Queue：消息队列载体，每个消息都会被投⼊到⼀个或多个队列。
Binding：绑定，它的作⽤就是把exchange和queue按照路由规则绑定起来。
Routing Key：路由关键字，exchange根据这个关键字进⾏消息投递。
vhost：虚拟主机，⼀个broker⾥可以开设多个vhost，⽤作不同⽤户的权限分离。
producer：消息⽣产者，就是投递消息的程序。
consumer：消息消费者，就是接受消息的程序。
channel：消息通道，在客户端的每个连接⾥，可建⽴多个channel，每个channel代表⼀个会话任务。

2.22、rabbitmq队列与消费者的关系？

⼀个队列可以绑定多个消费者；
消息默认以循环的⽅式发送给消费者；
消费者收到消息默认⾃动确认，也可以改成⼿动确认。

第四部分疑问解答

4.1 消费端用推模式消费消息时候，消费端往往需要多线程进行并发处理消息，

如果同时需要保证消息消费的顺序性，有什么方案？

如果设置了factory.setConcurrentConsumers(10); 10个并发处理的线程，有办法保证有序吗？

我的思路是单线程监听队列，构建一个内存队列：BlockingQueue＜Future＜T>>，使用Callable＜V>创建线程，然后多线程处理可以并发操作的内容，通过Future＜T>返回处理的结果，再构建一个单线程，处理BlockingQueue中的结果，把需要保证顺序性的操作放到单线程位置处理。

以下是一个简单代码实现举例子：

// Result自定义的处理结果
private BlockingQueue＜Future＜Result>> blockQueue = new ArrayBlockingQueue＜>(1000);
// 单线程监听队列
@RabbitHandler
public void process(String message){
 doDealMessage(message)
｝
// 多线程并发处理可以并发操作的部分，返回Result
public void doDealMsg(String message){
    while (true){
        CallableTask callableTask = new CallableTask(message);
        //consumerPool是一个线程池
        Future＜Result> submit = consumerPool.submit(callableTask);
        blockQueue.add(submit);
    ｝
｝
// 单线程顺序处理队列中的Result
public void update(){
     while(true){
         try {
             Future＜Result> take = blockQueue.take();
             while (true){
                 Result result = take.get();
                 if(result ！= null){
                     doResult(result);
                 ｝ else {
                     TimeUnit.MILLISECONDS.sleep(100);
                 ｝
             ｝
         ｝ catch (InterruptedException ¦ ExecutionException e) {
             log.error("InterruptedException！", e);
         ｝
       ｝
 ｝

Rabbit回顾