提到 RabbitMq,想必大家都不陌生,我们经常使用它来进行服务之间的异步通信,如短信,日志,邮件发送等业务场景都可以使用 RabbitMq 实现,但在使用的过程中我们会遇到消息丢失的问题,也会有延迟消费消息的需求,通过本文相信你会找到自己想要的答案,下面就让我们一起来看一下吧。
1. 消息可靠性
一条消息从发送到接收,在 RabbitMq 中会经历如下过程,如图:
上图中的每一步都可能存在消息丢失的情况,消息丢失原因包括以下几种:
- 发送时丢失:消息在发往 exchange 的过程中丢失;消息成功到达 exchange,但在发往 queue 的过程中丢失
- mq 服务宕机,queue 中的消息丢失
- 消费者成功接收消息,消息未消费完,服务宕机导致消息丢失
RabbitMq 针对以上问题,提供了自己的解决方案,如下:
- 生产者确认机制
- 消息持久化存储
- 消费者确认机制
- 失败重试机制
通过以上机制我们便可以解决消息丢失的问题,保证消息的可靠性,下面我们通过一个案例来演示上述四种机制(需要提前搭建好一个 RabbitMq 服务,不赘述),实现消息的可靠性传输。
1.项目搭建
首先创建 pom 父工程 mq-advanced-demo,pom.xml 配置如下:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>mq.demo</groupId><artifactId>mq-advanced-demo</artifactId><version>1.0-SNAPSHOT</version><modules><module>publisher</module><module>consumer</module></modules><packaging>pom</packaging><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.3.9.RELEASE</version><relativePath/></parent><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target></properties><dependencies><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><!--AMQP依赖,包含RabbitMQ--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><!--单元测试--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId></dependency><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId></dependency></dependencies></project>
然后分别创建子工程 publisher 和 consumer:
publisher 的 pom.xml:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><parent><artifactId>mq-advanced-demo</artifactId><groupId>mq.demo</groupId><version>1.0-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><artifactId>publisher</artifactId><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target></properties>
</project>
consumer 的 pom.xml:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><parent><artifactId>mq-advanced-demo</artifactId><groupId>mq.demo</groupId><version>1.0-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><artifactId>consumer</artifactId><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target></properties></project>
整体项目结构如下:
至此,项目模型基本搭建完毕(项目完整代码可于文末获取)。
2. 生产者确认机制
RabbitMQ 提供了生产者确认机制来避免消息发送到 MQ 过程中丢失。这种机制必须给每个消息指定一个唯一 ID。消息发送到 MQ 之后,会返回结果给发送者,表示消息是否处理成功。
返回结果有以下两种:
- publisher-confirm:发送者确认:消息成功投递到交换机,返回 ack;消息未投递到交换机,返回 nack
- publisher-return:发送者回执:消息成功投递到交换机,但是没有路由到队列。返回 ack,及路由失败原因
图示如下:
下面我们来进行编码实现,首先,修改 publisher 服务中的 application.yml,添加以下内容:
logging:pattern:dateformat: HH:mm:sslevel:mq: debug
spring:rabbitmq:# rabbitMQ的ip地址host: 127.0.0.1# 端口port: 5672# 集群模式配置# addresses: 127.0.0.1:8071, 127.0.0.1:8072, 127.0.0.1:8073username: adminpassword: 123456virtual-host: /# 消费者确认机制相关配置 # 开启publisher-confirm,# 这里支持两种类型:simple:同步等待confirm结果,直到超时;# correlated:异步回调,定义ConfirmCallback,MQ返回结果时会回调这个ConfirmCallbackpublisher-confirm-type: correlated# publish-returns:开启publish-return功能,同样是基于callback机制,不过是定义ReturnCallbackpublisher-returns: true# 定义消息路由失败时的策略。true,则调用ReturnCallback;false:则直接丢弃消息template:mandatory: true
然后在代码中定义 ReturnCallback 回调,每个RabbitTemplate只能配置一个ReturnCallback,因此需要在项目加载时配置,在 congfig 包下创建 CommonConfig 类:
package mq.config;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;@Slf4j
@Configuration
public class CommonConfig implements ApplicationContextAware {@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {// 获取RabbitTemplate对象RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);// 配置ReturnCallbackrabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {// 记录日志log.error("消息发送到队列失败,响应码:{}, 失败原因:{}, 交换机: {}, 路由key:{}, 消息: {}",replyCode, replyText, exchange, routingKey, message.toString());// 如果有需要的话,重发消息});}
}
接着定义 ConfirmCallback,ConfirmCallback 可以在发送消息时指定,因为每个业务处理 confirm 成功或失败的逻辑不一定相同。提前在 mq 控制台创建好 exchange 和 queue,在 publisher 服务下的 test 包下创建 SpringAmqpTest 类:
package mq.spring;import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;import java.nio.charset.StandardCharsets;
import java.util.UUID;@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testSendMessage2SimpleQueue() throws InterruptedException {// 1.准备消息String message = "hello, spring amqp!";// 2.准备CorrelationData// 2.1.消息IDCorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());// 2.2.准备ConfirmCallbackcorrelationData.getFuture().addCallback(result -> {// 判断结果if (result.isAck()) {// ACKlog.debug("消息成功投递到交换机!消息ID: {}", correlationData.getId());} else {// NACKlog.error("消息投递到交换机失败!消息ID:{},原因:{}", correlationData.getId(), result.getReason());// 重发消息}}, ex -> {// 记录日志log.error("消息发送异常, ID:{}, 原因{}", correlationData.getId(), ex.getMessage());// 可以重发消息});// 3.发送消息 需创建 mqtest 交换机,并绑定一个 routingKey 为 mqtest_queue 的队列,可根据实际情况修改rabbitTemplate.convertAndSend("mqtest", "mqtest_queue", message, correlationData);// 休眠一会儿,等待ack回执Thread.sleep(2000);}
}
执行测试类,打印 log 如下:
异常情况大家可以自行测试。
3. 消息持久化存储
生产者确认机制可以确保消息投递到 RabbitMQ 的队列中,但是消息发送到 RabbitMQ 以后,如果突然宕机,也可能导致消息丢失。
要想确保消息在 RabbitMQ 中安全保存,必须开启消息持久化机制,即交换机持久化,队列持久化,消息持久化。
默认情况下,springAmqp 声明的交换机,队列,消息都是持久化的,并不需要我们特意指定,即 Durability 属性都为 Durable。
4. 消费者确认机制
RabbitMQ 是阅后即焚机制,RabbitMQ 确认消息被消费者消费后会立刻删除。
RabbitMQ 是通过消费者回执来确认消费者是否成功处理消息的:消费者获取消息后,向 RabbitMQ 发送 ACK 回执,表明自己已经处理消息。
SpringAmqp 则允许配置三种确认模式:
- manual:手动 ack,需要在业务代码结束后,调用 api 发送 ack,虽灵活但会提高编码复杂度。
- auto:自动 ack,没有异常则返回 ack;抛出异常则返回
nack,消息重新入队,一直到没有异常为止,也可以设置最大重试次数,超过次数后发送到专门收集错误消息的队列进一步处理 - none:关闭ack,MQ 假定消费者获取消息后会成功处理,因此消息投递后立即被删除(消息投递是不可靠的,可能丢失)
一般情况下,我们选择使用 auto 模式即可,接下来进行编码实现。
向 consumer 服务的 application.yml 添加以下内容:
logging:pattern:dateformat: HH:mm:sslevel:mq: debug
spring:rabbitmq:#rabbitMQ的ip地址host: 127.0.0.1#端口port: 5672#集群模式配置#addresses: 127.0.0.1:8071, 127.0.0.1:8072, 127.0.0.1:8073username: adminpassword: 123456virtual-host: /listener:simple:prefetch: 1acknowledge-mode: auto
在 listener 包下添加 SpringRabbitListener 类:
package mq.listener;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Slf4j
@Component
public class SpringRabbitListener {@RabbitListener(queues = "mqtest_queue")public void listenSimpleQueue(String msg) {log.debug("消费者接收到mqtest_queue的消息:【" + msg + "】");//System.out.println(1 / 0);log.info("消费者处理消息成功!");}
}
启动服务,消息成功被消费:
但如果处理消息的过程中出现异常,消息会不断重新入队,不断消费,导致 cpu 飙升,这个时候便需要结合失败重试机制来对消息进行处理。
5. 失败重试机制
我们可以利用 Spring 的 retry 机制,在消费者出现异常时进行本地重试,而不是无限制的 requeue 到 mq 队列,修改 consumer 服务的 application.yml 文件:
logging:pattern:dateformat: HH:mm:sslevel:mq: debug
spring:rabbitmq:#rabbitMQ的ip地址host: 127.0.0.1#端口port: 5672#集群模式配置#addresses: 127.0.0.1:8071, 127.0.0.1:8072, 127.0.0.1:8073username: adminpassword: 123456virtual-host: /listener:simple:prefetch: 1acknowledge-mode: auto# 失败重试retry:# 开启消费者失败重试enabled: true# 初始的失败等待时长为1秒initial-interval: 1000# 失败的等待时长倍数,下次等待时长 = multiplier * last-intervalmultiplier: 3# 最大重试次数max-attempts: 4# true无状态;false有状态。如果业务中包含事务,这里改为falsestateless: true
接下来重新向队列中发送消息,故意让监听代码产生异常,启动 consumer 服务,打印 log 如下:
如上图,消息重试了 4 次最终抛出异常,但该消息同样被 mq 删除了,所以我们会得到以下结论:
- 开启本地重试时,消息处理过程中抛出异常,不会 requeue 到队列,而是在消费者本地重试
- 重试达到最大次数后,Spring会返回ack,消息会被丢弃
在开启重试模式后,重试次数耗尽,如果消息依然失败,为了防止消息被直接丢弃,需要有MessageRecovery 接口来处理,它包含三种不同的实现:
- RejectAndDontRequeueRecoverer:重试耗尽后,直接 reject,丢弃消息。默认就是这种方式
- ImmediateRequeueMessageRecoverer:重试耗尽后,返回nack,消息重新入队
- RepublishMessageRecoverer:重试耗尽后,将失败消息投递到指定的交换机
很显然,RepublishMessageRecoverer 方式更为友好,失败后可以将消息投递到一个指定的,专门存放异常消息的队列,后续集中处理。
在 consumer 服务中定义处理失败消息的交换机和队列:
@Bean
public DirectExchange errorMessageExchange(){return new DirectExchange("error.direct");
}
@Bean
public Queue errorQueue(){return new Queue("error.queue", true);
}
@Bean
public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
}
定义一个 RepublishMessageRecoverer,关联队列和交换机:
@Bean
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
}
重启 consumer 服务,可以发现错误消息被成功发到了指定的队列:
6. 总结
通过以上案例分析,我们可以总结出以下几点来确保 RabbitMq 的消息可靠性传输:
- 开启生产者确认机制,确保生产者的消息能到达队列
- 开启持久化功能,确保消息未消费前在队列中不会丢失
- 开启消费者确认机制为 auto,由 spring 确认消息处理成功后完成 ack
- 开启消费者失败重试机制,并设置 MessageRecoverer,多次重试失败后将消息投递到异常交换机,后续集中处理
2. 实现延迟队列
当一个队列中的消息满足下列情况之一时,可以成为死信(dead letter):
- 消费者使用 basic.reject 或 basic.nack 声明消费失败,并且消息的 requeue 参数设置为 false
- 消息是一个过期消息,超时无人消费
- 要投递的队列消息满了,无法投递
如果这个包含死信的队列配置了 dead-letter-exchange 属性,指定了一个交换机,那么队列中的死信就会投递到这个交换机中,而这个交换机称为死信交换机(Dead Letter Exchange,简称DLX)。
如果同时配置了 dead-letter-routing-key 参数,则消息会由死信交换机发送到指定的队列,也就是死信队列。
接下来在 consumer 服务中,定义一组死信交换机,死信队列:
@Beanpublic DirectExchange dlExchange() {// 声明死信交换机 dl.directreturn new DirectExchange("dl.direct", true, false);}@Beanpublic Queue dlQueue() {// 声明存储死信的队列 dl.queuereturn new Queue("dl.queue", true);}@Beanpublic Binding dlBinding() {// 将死信队列 与 死信交换机绑定return BindingBuilder.bind(dlQueue()).to(dlExchange()).with("dl");}
为了实现消息的延迟消费,我们需要再定义一组接收超时消息的交换机和队列并把两者绑定,设置超时时间为 10s:
@Bean
public DirectExchange ttlExchange(){return new DirectExchange("ttl.direct");
}@Bean
public Queue ttlQueue(){return QueueBuilder.durable("ttl.queue") // 设置队列的超时时间,10秒.ttl(10000).deadLetterExchange("dl.direct").deadLetterRoutingKey("dl").build();
}@Bean
public Binding ttlBinding(){return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("ttl");
}
监听超时死信队列:
@RabbitListener(queues = "dl.queue")public void listenDlQueue(String msg) {log.info("消费者接收到了dl.queue 的延迟消息:{}", msg);}
在 publisher 服务测试发送消息:
@Testpublic void testTTLMessage() throws InterruptedException {// 1.准备消息Message message = MessageBuilder.withBody("hello, ttl messsage".getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();// 2.发送消息rabbitTemplate.convertAndSend("ttl.direct", "ttl", message);// 3.记录日志log.info("消息已经成功发送!");}
重新启动 consumer 服务,执行测试代码,查看 log:
查看 consumer 服务 的 log:
通过对比消息发送和消费的时间,发现二者相差 10s,跟我们设置的队列超时时间相同,延迟消息得以实现。
我们也可以在发送消息的时候设置消息过期时间:
@Testpublic void testTTLMessage() throws InterruptedException {// 1.准备消息Message message = MessageBuilder.withBody("hello, ttl messsage".getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.PERSISTENT)//设置超时时间为5s.setExpiration("5000").build();// 2.发送消息rabbitTemplate.convertAndSend("ttl.direct", "ttl", message);// 3.记录日志log.info("消息已经成功发送!");}
执行测试代码,查看 log:
查看 consumer 服务 的 log:
可以看到,发送和消费的时间相差 5s,跟我们设置的消息超时时间相同。
下面列举几种延迟队列的使用场景:
- 延迟发送短信
- 用户下单,如果用户在一小时内未支付,自动取消
- 预约会议,半小时后通知所有参会人员
当然,这些我们也可以使用定时任务来实现,只是会徒增许多无意义的查询,数据量大的话会比较耗费性能。
总结一下,设置消息超时主要有以下两种方式:
- 给队列设置 ttl 属性,进入队列后超过 ttl 时间的消息变为死信
- 给消息设置 ttl 属性,队列接收到消息超过 ttl 时间后变为死信
实现延迟消费的步骤:
- 创建一个交换机作为死信交换机并绑定一个队列作为死信队列
- 给消息的目标队列设置队列超时时间并指定死信交换机和路由 key
- 将消息的目标队列绑定到死信交换机
- 消费者监听死信队列获取超时消息