RabbitMQ特性

消息可靠性：典型的生产者-消费者模型，发送端有消息发送确认机制，服务端有消息持久化方案，消费端有消息Ack机制
灵活的消息路由分发：多种多样的交换机
多语言客户端开发AMQP：只要实现了AMQP，就能和RabbitMQ对接
友好管理界面：通过插件可以在浏览器通过界面来管理
灵活插件配置：

RabbitMQ缺点：

RabbitMQ不提供高可用的集群方案，需要手工实现HAProxy + keepalived方式进行集群搭建
性能方面、处理能力没有RocketMQ和Kafka高

RabbitMQ架构与核心组件

RabbitMQ是一个典型的CS架构，是一个典型的生产者-消费者模型。

Broker：每一台单独运行的RabbitMQ进程就是一个Broker。多个Broker可以通过技术方案组成集群环境。

virtual host：虚拟主机，一个broker中有多个vh，用来环境隔离，区分不同环境（比如用户系统和订单系统划分在两个vh上），同一个vh中每个队列对象是唯一的，而不同vh之间，队列对象是可以相同的

Exchange：消息路由的交换机，是消息接收的通道。一条消息由producer产生之后，发送给broker，必须要到达交换机，然后由交换机转交给队列

routing key：消息路由分发的关键字，与binding中的key进行匹配从而分发到交换机绑定的指定队列queue。

注：发送一条消息，必须指定ExchangeName，告诉发送给哪个交换机；其次需要指定routing key，消息路由分发的关键字；

connection：客户端producer/consumer与服务端broker连接的连接对象，通过指定Host和Port去连接到对应的broker，还需要指定连接到哪个virtual host。但是最终实际操作对象是channel对象，一个connect可以创建多个channel，可以把connection理解成是JDBC中的连接池。

channel：producer中的channel是负责真正操作的对象，和交换机进行交互；consumer中的channel直接监听在对应的queue上，从而获取消息。

RabbitMQ中提供的Exchange

direct-exchange：直连类型交换机，根据routing key和binding key进行全匹配找到对应的queue
topic-exchange：主题类型交换机，根据通配符模糊匹配
fanout-exchange：广播类型交换机，和交换机绑定的队列都会收到消息
headers-exchange：头交换机，根据header中的值匹配

direct-exchange直连交换机

根据routing key和binding key进行全匹配，匹配到几个队列，那么消息就会发送到几个队列。

topic-exchange主题交换机

提供两个通配符*和#，根据通配符来进行匹配对应的队列。

* 代表英文一个单词

# 代表0个或多个单词

fanout-exchange广播交换机

发送消息时不需要指定路由关键字routing key，交换机与队列之间的绑定关系也不需要binding key，因为是广播，需要将消息发送给交换机所绑定的所有队列。

headers-exchange头交换机

与header头信息进行配置，匹配其中一个则符合条件

订单超时（30分钟）未支付，怎么办？

定时任务+扫表（低级）：定时（比如3分钟）去扫描一次表，把未支付的订单退回到仓库

延时队列（高级）：30分钟后去判断当前订单是否支付，如果未支付退回到仓库

RabbitMQ高可用方案

Cluster普通模式：只同步元数据

Cluster单机多节点集群

Cluster多机多节点集群

镜像模式：同步内容

有状态和无状态

如果一个应用，复制多份，分别访问，可以得到一样的结果，请求一个应用修改了值，其他应用同时也会修改，则此应用是无状态的。

容器化水平扩容，需要用无状态的应用。

RabbitMQ消息可靠性

RabbitMQ中的死信问题及处理方案

死信 dead-letter如何产生？

RabbitMQ consumer主动拒绝消费
消息的TTL（Time To Live）过期，设置队列中的消息5秒钟过期，那么TTL就是5秒
消息在加入队列时超过队列的长度

解决方案

当消息成为死信之后，若该队列配置了死信交换机，则将消息按规则转发给该死信交换机（必须为主题交换机），如果没有配置死信交换机，则消息丢弃不再处理，消息消亡。

延时队列实现

通过死信来实现延时队列。

延时策略回调

方案一：定时任务 + 扫表

创建任务表 [callTime, status]

定时Job发起数据库查询SQL [callTime < now() and status = 1]

拿到任务若回调成功 [update status = 0]

拿到任务若回调失败 [update callTime = callTime + 间隔时间]

方案二：redis的zset数据类型

方案三：延时队列

执行一个任务，如果执行成功则任务完成，如果执行失败，会延时10秒后再次回调执行，如果还是失败，延时更长时间再次回调执行，如果依然失败，延时时间再次延长......

RabbitMQ延时队列

某条消息交给某个交换机exchange1

路由分发到对应的消费队列consume queue1

消费者Consumer1拿到消息后调用远方服务Remote Server

如果RemoteServer返回正确结果，则消费者采用Ack机制，代表消息被成功消费

如果RemoteServer返回错误结果，则消费者在确认消费过程中执行Nack拒绝消费

拒绝消费的消息则会成为当前队列的死信

将死信转发给死信交换机，exchange2就是consume queue1的死信交换机

exchange2根据通配符 # 将消息发送给delay queue2延时队列

delay queue2设置当前队列消息过期时间TTL=10s，那么10秒后消息会自动过期，变成死信

将delay queue2的死信转发给对应的死信交换机exchange3

exchange3继续执行exchange1的相同流程，如此往复执行

RabbitMQ基本使用

生产者发送消息

public class Producer {public static void main(String[] args) {ConnectionFactory factory = new ConnectionFactory();// 设置连接Broker实例的属性factory.setHost("127.0.0.1");factory.setPort(5672);factory.setUsername("guest");factory.setPassword("guest");Connection connection = null;Channel channel = null;try {// 创建一个Producer与Broker实例之间的连接connection = factory.newConnection("producer");// 在连接中创建一个信道channel = connection.createChannel();// 声明交换机channel.exchangeDeclare("t-exchange1", "topic");// 声明队列channel.queueDeclare("t-queue1", true, false, false, null);// 绑定队列给交换机channel.queueBind("t-queue1", "t-exchange1", "t-rk*");channel.queueDeclare("t-queue2", true, false, false, null);channel.queueBind("t-queue2", "t-exchange1", "t-rk*");channel.queueDeclare("t-queue3", true, false, false, null);channel.queueBind("t-queue3", "t-exchange1", "t-rk#");String message = "Hello, this is message from producer";// 生产者发送消息// mandatory：如果根据消息routingKey无法找到与exchange消息类型匹配的queue，是否将消息返回给生产者// immediate：如果exchange将消息路由到对应的queue上时发现queue中没有消费者，是否将消息返回给生产者channel.basicPublish("t-exchange1", "t-rk-01", true, false, null, message.getBytes());System.out.println("消息已发送!");} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();} finally {try {if (null != channel && channel.isOpen()) {channel.close();}if (null != connection && connection.isOpen()) {connection.close();}} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}}}
}

消费者订阅消息

public class Consumer {public static void main(String[] args) {ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setUsername("guest");factory.setPassword("guest");Connection connection = null;Channel channel = null;try {connection = factory.newConnection("consumer");channel = connection.createChannel();channel.basicConsume("f-queue1", true, new DeliverCallback() {@Overridepublic void handle(String consumerTag, Delivery message) throws IOException {System.out.println("消费者" + consumerTag + "收到消息：" + new String(message.getBody(), "UTF-8"));}}, new CancelCallback() {@Overridepublic void handle(String consumerTag) throws IOException {System.out.println("消费者" + consumerTag + "取消消费!");}});System.out.println("开始接收消息...");System.in.read();} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();} finally {try {if (null != channel && channel.isOpen()) {channel.close();}if (null != connection && connection.isOpen()) {connection.close();}} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}}}
}

RabbitMQ在Spring中的使用

pom.xml中引入rabbitmq依赖

<dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><version>1.3.5.RELEASE</version>
</dependency>

applicationContext.xml相关配置

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsdhttp://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd"><import resource="classpath*:rabbitMQ.xml" /><!-- 扫描指定package下所有带有如 @Controller,@Service,@Resource 并把所注释的注册为Spring Beans --><context:component-scan base-package="com.gupaoedu.*" /><!-- 激活annotation功能 --><context:annotation-config /><!-- 激活annotation功能 --><context:spring-configured />
</beans>

rabbitMQ.xml相关配置

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans-3.0.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit-1.2.xsd"><!--配置connection-factory，指定连接rabbit server参数 --><rabbit:connection-factory id="connectionFactory" virtual-host="/" username="guest" password="guest" host="127.0.0.1" port="5672" /><!--通过指定下面的admin信息，当前producer中的exchange和queue会在rabbitmq服务器上自动生成 --><rabbit:admin id="connectAdmin" connection-factory="connectionFactory" /><!--######分隔线######--><!--定义queue --><rabbit:queue name="MY_FIRST_QUEUE" durable="true" auto-delete="false" exclusive="false" declared-by="connectAdmin" /><!--定义direct exchange，绑定MY_FIRST_QUEUE --><rabbit:direct-exchange name="MY_DIRECT_EXCHANGE" durable="true" auto-delete="false" declared-by="connectAdmin"><rabbit:bindings><rabbit:binding queue="MY_FIRST_QUEUE" key="FirstKey"></rabbit:binding></rabbit:bindings></rabbit:direct-exchange><!--定义rabbit template用于数据的接收和发送 --><rabbit:template id="amqpTemplate" connection-factory="connectionFactory" exchange="MY_DIRECT_EXCHANGE" /><!--消息接收者 --><bean id="messageReceiver" class="com.gupaoedu.consumer.FirstConsumer"></bean><!--queue listener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象 --><rabbit:listener-container connection-factory="connectionFactory"><rabbit:listener queues="MY_FIRST_QUEUE" ref="messageReceiver" /></rabbit:listener-container><!--定义queue --><rabbit:queue name="MY_THIRD_QUEUE" durable="true" auto-delete="false" exclusive="false" declared-by="connectAdmin" /><!-- 定义topic exchange，绑定MY_THIRD_QUEUE，注意关键词是pattern --><rabbit:topic-exchange name="MY_TOPIC_EXCHANGE" durable="true" auto-delete="false" declared-by="connectAdmin"><rabbit:bindings><rabbit:binding queue="MY_THIRD_QUEUE" pattern="#.Third.#"></rabbit:binding></rabbit:bindings></rabbit:topic-exchange><!--定义rabbit template用于数据的接收和发送 --><rabbit:template id="amqpTemplate2" connection-factory="connectionFactory" exchange="MY_TOPIC_EXCHANGE" /><!-- 消息接收者 --><bean id="receiverThird" class="com.gupaoedu.consumer.ThirdConsumer"></bean><!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象 --><rabbit:listener-container connection-factory="connectionFactory"><rabbit:listener queues="MY_THIRD_QUEUE" ref="receiverThird" /></rabbit:listener-container>
</beans>

消息生产者

@Service
public class MessageProducer {private Logger logger = LoggerFactory.getLogger(MessageProducer.class);@Autowired@Qualifier("amqpTemplate")private AmqpTemplate amqpTemplate; // 直连交换机@Autowired@Qualifier("amqpTemplate2")private AmqpTemplate amqpTemplate2; // 主题交换机/*** 演示三种交换机的使用** @param message*/public void sendMessage(Object message) {// Exchange 为 direct 模式，直接指定routingKeyamqpTemplate.convertAndSend("FirstKey", "[Direct,FirstKey] "+message);amqpTemplate.convertAndSend("SecondKey", "[Direct,SecondKey] "+message);// Exchange模式为topic，通过topic匹配关心该主题的队列amqpTemplate2.convertAndSend("msg.Third.send","[Topic,msg.Third.send] "+message);// 广播消息，与Exchange绑定的所有队列都会收到消息，routingKey为空amqpTemplate2.convertAndSend("MY_FANOUT_EXCHANGE",null,"[Fanout] "+message);}
}

消息订阅者/消费者

public class FirstConsumer implements MessageListener {private Logger logger = LoggerFactory.getLogger(FirstConsumer.class);public void onMessage(Message message) {logger.info("The first consumer received message : " + message.getBody());}
}

RabbitMQ常见问题

1、如何保证消息可靠性？

用事务保证消息正确传递（但是会有性能问题，不推荐）
生产端：开启confirm监听器（推荐）
Broker实例：开启持久化
消费端：使用手动ACK机制（默认是自动ACK）

从以下方面考虑可靠性：

1) 生产端发送可靠性：confirm机制确保消息成功发送到Broker

a. 事务模式：阻塞的，性能低
b. 服务端普通确认：发送一条确认一条，效率低
c. 服务端批量确认：没有达到一定条件，会一直不确认
d. 服务端异步确认：边发送边确认

2) Broker实例存储可靠性：Broker对消息持久化，确保消息不会丢失

a. 交换机持久化
b. 队列持久化
c. 消息持久化

3) 消费端消费可靠性：消费者在消费的同时，将autoAck设置为false，然后通过手动确认的方式去确认消息被正确消费，以免引起不必要的消息丢失，出现异常则nack。

4) 消费者回调：消费者消费完成发送一条响应消息给生产者

5) 补偿机制：消息落库，定时扫描，重发(衰减机制)，设置一定重发次数

6) 消息幂等性：每个消息发送时就有一个唯一的标识ID，类似银行流水号，不管消息发送/消费了多少次，最终处理结果必须相同

7) 最终一致性：如果以上做法都没有成功，重发也失败，那么需要人工干预实现最终一致性

8) 消息顺序性：生产者发送消息的顺序必须和消费者消费消息的顺序一致，所以一般一个队列只对应一个消费者消费

2、如何保证消息不重复消费？

消息确认机制和幂等机制：每个消息生成一个全局唯一MessageID，每次消费前判断下当前消息是否已经被消费，如果消费过了，返回ack，则会把消息从队列中删除。

3、如何保证顺序消费？

同一个交换机，同一个队列，同一个消费者，单线程消费

4、消息重试机制？

消费端在处理消息过程中可能会出错，此时需要设置消息重试机制，解决方案有以下两种：

在redis或者数据库中记录重试次数，达到最大重试次数以后消息进入死信队列或者其他队列，再单独针对这些消息进行处理；
使用spring-rabbit中自带的retry功能；

spring:rabbitmq:listener:simple:retry:enabled: true #是否开启重试initial-interval: 3000ms #重试时间间隔max-attempts: 3 #重试次数max-interval: 15000ms #重试最大时间间隔multiplier: 2 #倍数，间隔时间*倍数=下一次的间隔时间，最大不能超过设置的最大间隔时间

5、死信队列和延时队列

死信消息：

消息被拒绝消费（Basic.Reject或Basic.Nack），并且设置 requeue 参数的值为 false
消息过期了，超过了队列中设置的TTL时间仍然没有被消费
消息加入队列时，队列达到了最大的长度

死信队列：当消息变成死信后，如果这个消息所在的队列存在x-dead-letter-exchange参数，就会被发送到x-dead-letter-exchange对应值的交换机上，这个交换机就称为死信交换机，与这个死信交换机绑定的队列就是死信队列。

延时队列：实际上RabbitMQ中不存在延时队列，但是我们可以通过设置消息过期时间TTL和死信队列来构建延时队列。

6、消息积压，如果保证消息不丢失？

如果是consumer消费速度落后于producer生产的速度，可以扩容消费者群组。

如果消息积压比较严重，上百万条，那么写个临时consumer程序处理积压消息。

7、分布式事务

8、订单超时如何处理？

移到死信队列

9、监控

自研（通过rest api）：适合大厂有能力做监控平台的

prometheus：适合中小企业，节约成本

小项目用rabbitmq-management

10、mandatory和immediate

mandatory和immediate是AMQP协议中basic.publish方法中的两个标识位，它们都有当消息传递过程中不可达目的地时是否将消息返回给生产者的功能。

mandatory：为true时，如果exchange根据自身类型与消息routingKey无法找到与之匹配的queue，那么会调用basic.return方法将消息返回给生产者（Basic.Return + Content-Header + Content-Body）；当mandatory设置为false时，出现上述情形broker会直接将消息扔掉。

immediate：为true时，如果exchange将消息路由到对应的queue上时发现queue中没有消费者，那么这条消息不会放入到队列中，该消息会通过basic.return返回给生产者，不用消息入队列等待消费者消费。

11、消息队列的作用和应用场景

异步、解耦、削峰、广播

12、Channel和Vhost的作用

vhost资源隔离，提升硬件资源利用率

13、RabbitMQ消息路由方式和应用场景

直连direct(通过唯一routing key匹配)：

主题topic(通配符*/#)：

广播fanout：只要和这个交换机绑定的队列都发送

头信息headers：

14、交换机与队列，队列与消费者的绑定关系

交换机与队列通过binding key，队列与消费者通过routing key

15、一个队列可存放多少条消息

没有规定，但是可以设置队列长度，maxLength

16、集群节点类型

内存节点ram：只存储元数据，交换机名字类型，队列名字等

磁盘节点disc：既存元数据，也存消息内容

17、AmqpTemplate和RabbitTemplate的区别

父子关系，rabbitmq遵循amqp协议，amqp是amqp协议的一种抽象，而rabbitmq是amqp的一种实现

18、Spring AMQP消息怎么封装，怎么转换

用Message封装，MessageConverter转换

19、如何动态创建队列和消费者

container.setQueues()通过配置文件动态创建

20、多个消费者监听同一个队列，消息会重复消费吗？

不会，但是会导致消息乱序，所以最好一个消费者对应一个队列，避免消息乱序

21、无法被路由的消息，去了哪里？

要么直接丢弃，要么设置备份交换机接收消息，或者消息回发给生产者

22、消息在什么时候变成死信Dead Letter？

以下情况都会变成死信：消息过期未消费，消息超过队列长度，消息被拒绝消费

23、如果一个项目需要从多个服务器接收消息，怎么做？

定义多个ConnectionFactory，或者将Template定义为多例，类似配置多数据源

24、RabbitMQ如何实现延时队列？

1) 数据库 + 定时器; 2) TTL + 死信; 3) 安装延时队列插件

25、哪些情况会导致消息丢失？如何解决？

1) Producer到Broker出现异常：事务模式或者确认模式(普通确认，批量确认、异步确认)保证消息可靠投递

2) exchange到队列出现异常：消息回发给生产者、备份交换机接收消息

3) 队列自身出现异常：队列持久化，交换机持久化，消息持久化

4) 队列到消费者出现异常：手动ack

保证消息可靠性的其他方式：

消费者回调：消费者消费完成发送一条响应消息给生产者

补偿机制：重发

消息幂等性：通过流水号或消息唯一id保证幂等性，唯一消费

最终一致性：

消息顺序性：一个队列只对应一个消费者消费

26、可以用队列的x-max-length最大消息数来实现限流吗？例如秒杀场景

不可以，如果队列满了，就会删除先入队列的消息，导致消息丢失。正确方法是调整内存告警阈值 + 集群部署，提高负载能力

27、如何提高消息的消费速率？

用多线程，增加消费者

28、如何保证消息顺序性？

一个队列只对应一个消费者，或者分布式锁

29、如何保证RabbitMQ的高可用？

集群(镜像集群) + 负载(HAproxy + Keepalived)

30、大量消息堆积怎么办?

增加消费者

31、设计一个MQ，你的思路是什么？

MQ主要是存储消息和分发消息

存储：内存 / 磁盘

分发：pull / push

通信：HTTP / TCP / AMQP

RabbitMQ介绍与延时队列相关推荐

（转） RabbitMQ学习之延时队列
http://blog.csdn.net/zhu_tianwei/article/details/53563311 在实际的业务中我们会遇见生产者产生的消息,不立即消费,而是延时一段时间在消费.Rab ...
Docker安装RabbitMQ并安装延时队列插件
一.RabbitMQ简介 RabbitMQ是由erlang语言开发,基于AMQP(Advanced Message Queue 高级消息队列协议)协议实现的消息队列,它是一种应用程序之间的通信方法,消 ...
【SpringBoot】43、SpringBoot中整合RabbitMQ实现延时队列（延时插件篇）
死信队列实现篇,参考文章:[SpringBoot]60.SpringBoot中整合RabbitMQ实现延时队列(死信队列篇) 一.介绍 1.什么是延时队列? 延时队列即就是放置在该队列里面的消息是不需 ...
延时队列的几种实现方式
延时队列的几种实现方式何为延迟队列? 顾名思义,首先它要具有队列的特性,再给它附加一个延迟消费队列消息的功能,也就是说可以指定队列中的消息在哪个时间点被消费. 延时队列能做什么? 延时队列多用于需要 ...
基于Redis实现延时队列的优化方案
一.延时队列的应用近期在开发部门的新项目,其中有个关键功能就是智能推送,即根据用户行为在特定的时间点向用户推送相应的提醒消息,比如以下业务场景: 在用户点击充值项后,半小时内未充值,向用户推送充值未 ...
关于延时队列的一些思考
最近由于项目的需要原因,需要做一个延时队列,比如用户登录X秒后需要发送一些系统消息.或者要做一个小游戏,需要有操作超时检测,如果超时,则自动跳到下一个玩家操作.这些,都用到了定时检测,而又想到了red ...
RabbitMQ延时队列原理讲解
RabbitMQ延时消息队列延时队列介绍延时队列即放置在该队列里面的消息是不需要立即消费的,而是等待一段时间之后取出消费. 那么,为什么需要延迟消费呢?我们来看以下的场景网上商城下订单后30分钟 ...
RabbitMQ 的延时队列和镜像队列原理与实战
在阿里云栖开发者沙龙PHP技术专场上,掌阅资深后端工程师.掘金小测<Redis深度历险>作者钱文品为大家介绍了RabbitMQ的延时队列和镜像队列的原理与实践,重点比较了RabbitMQ提 ...
RabbitMQ通过TTL和DLX实现延时队列
RabbitMQ实现延时队列一.介绍 1.TTL 如何设置TTL(2种方式): 2.Dead Letter Exchanges 二.实现延时队列的思路三.SpringBoot+RabbitMQ实现 ...

RabbitMQ介绍与延时队列