在之前的练习作业中，我们改造了余额支付功能，在支付成功后利用RabbitMQ通知交易服务，更新业务订单状态为已支付。
但是大家思考一下，如果这里MQ通知失败，支付服务中支付流水显示支付成功，而交易服务中的订单状态却显示未支付，数据出现了不一致。
此时前端发送请求查询支付状态时，肯定是查询交易服务状态，会发现业务订单未支付，而用户自己知道已经支付成功，这就导致用户体验不一致。
因此，这里我们必须尽可能确保MQ消息的可靠性，即：消息应该至少被消费者处理1次
那么问题来了：
- 我们该如何确保MQ消息的可靠性？
- 如果真的发送失败，有没有其它的兜底方案？
1.发送者的可靠性
首先，我们一起分析一下消息丢失的可能性有哪些。
消息从发送者发送消息，到消费者处理消息，需要经过的流程是这样的：
消息从生产者到消费者的每一步都可能导致消息丢失：
- 发送消息时丢失：- 生产者发送消息时连接MQ失败- 生产者发送消息到达MQ后未找到
Exchange- 生产者发送消息到达MQ的Exchange后，未找到合适的Queue- 消息到达MQ后，处理消息的进程发生异常 - MQ导致消息丢失：- 消息到达MQ，保存到队列后，尚未消费就突然宕机
- 消费者处理消息时：- 消息接收后尚未处理突然宕机- 消息接收后处理过程中抛出异常
综上，我们要解决消息丢失问题，保证MQ的可靠性，就必须从3个方面入手：
- 确保生产者一定把消息发送到MQ
- 确保MQ不会将消息弄丢
- 确保消费者一定要处理消息
1.1.生产者重试机制
首先第一种情况，就是生产者发送消息时，出现了网络故障，导致与MQ的连接中断。
为了解决这个问题，SpringAMQP提供的消息发送时的重试机制。即：当
RabbitTemplate
与MQ连接超时后，多次重试。
修改
publisher
模块的
application.yaml
文件，添加下面的内容：
spring:
rabbitmq:
connection-timeout: 1s # 设置MQ的连接超时时间
template:
retry:
enabled: true # 开启超时重试机制
initial-interval: 1000ms # 失败后的初始等待时间
multiplier: 1 # 失败后下次的等待时长倍数，下次等待时长 = initial-interval * multiplier
max-attempts: 3 # 最大重试次数
我们利用命令停掉RabbitMQ服务：
docker stop mq
然后测试发送一条消息，会发现会每隔1秒重试1次，总共重试了3次。消息发送的超时重试机制配置成功了！
注意：当网络不稳定的时候，利用重试机制可以有效提高消息发送的成功率。不过SpringAMQP提供的重试机制是阻塞式的重试，也就是说多次重试等待的过程中，当前线程是被阻塞的。通俗来说在等待期间，代码卡在那里了，不会向下运行了。
如果对于业务性能有要求，建议**禁用**重试机制。如果一定要使用，请合理配置等待时长和重试次数，当然也可以考虑使用异步线程来执行发送消息的代码。
1.2.生产者确认机制
相比于生产者重试机制，生产者确认机制侧重的是：消息发送时，失败了怎么办。
一般情况下，只要生产者与MQ之间的网路连接顺畅，基本不会出现发送消息丢失的情况，因此大多数情况下我们无需考虑这种问题。
不过，在少数情况下，也会出现消息发送到MQ之后丢失的现象，比如：
- MQ内部处理消息的进程发生了异常
- 生产者发送消息到达MQ后未找到
Exchange - 生产者发送消息到达MQ的
Exchange后，未找到合适的Queue，因此无法路由
针对上述情况，RabbitMQ提供了生产者消息确认机制，包括
Publisher Confirm
和
Publisher Return
两种。在开启确认机制的情况下，当生产者发送消息给MQ后，MQ会根据消息处理的情况返回不同的回执。
具体如图所示：
总结如下：
- 当消息投递到MQ，但是路由失败时，通过Publisher Return返回异常信息，同时返回ack的确认信息，代表投递成功。这种情况一般是RoutingKey填写失败，跟我MQ发送没有关系！
- 临时消息投递到了MQ，并且入队成功，返回ACK，告知投递成功
- 持久消息投递到了MQ，并且入队完成持久化（保存到磁盘）后，才会返回ACK ，告知投递成功
- 其它情况都会返回NACK，告知投递失败
比如说：消息投递到交换机，结果还没来得及持久化到磁盘，结果磁盘因为某种故障 (比如说磁盘满了)；MQ内部出现异常 (比如说内存爆满)导致消息丢失。这些情况都会返回NACK。
其中
ack
和
nack
属于Publisher Confirm机制，
ack
是投递成功；
nack
是投递失败。而
return
则属于Publisher Return机制。
默认两种机制都是关闭状态，需要通过配置文件来开启。
1.3.实现生产者确认
在我们上面分析的生产者确认机制里面：生产者发消息，MQ给我们回值，我们应该去接收这个值。这里就有两种方法：1.同步等待，发了消息就开始等待，等着MQ给我回值。2.采用异步回调的方式，生产者发了消息，就干别的事了。当MQ回值来了以后我再去处理就行了。
当然这种异步的方式会好一点！接下来我们就采用这种方式。
1.3.1.开启生产者确认
在publisher模块的
application.yaml
中添加配置：
spring:
rabbitmq:
publisher-confirm-type: correlated # 开启publisher confirm机制，并设置confirm类型
publisher-returns: true # 开启publisher return机制，专门用来返回路由失败消息的
这里
publisher-confirm-type
有三种模式可选：
none：关闭confirm机制simple：同步阻塞等待MQ的回执- **
correlated**：MQ异步回调返回回执
一般我们推荐使用**
correlated
，异步回调机制**。
1.3.2.定义ReturnCallback (回调函数)
**每个
RabbitTemplate
只能配置一个
ReturnCallback
，因此我们可以在配置类中统一设置。我们在publisher模块定义一个配置类：**
内容如下：
package com.itheima.publisher.config;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Configuration;
import javax.annotation.PostConstruct;
@Slf4j
@AllArgsConstructor
@Configuration
public class MqConfig {
private final RabbitTemplate rabbitTemplate;
@PostConstruct
public void init(){
rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
@Override
public void returnedMessage(ReturnedMessage returned) {
log.error("触发return callback,");
log.debug("exchange: {}", returned.getExchange());
log.debug("routingKey: {}", returned.getRoutingKey());
log.debug("message: {}", returned.getMessage());
log.debug("replyCode: {}", returned.getReplyCode());
log.debug("replyText: {}", returned.getReplyText());
}
});
}
}
1.3.3.定义ConfirmCallback
由于每个消息发送时的处理逻辑不一定相同，因此ConfirmCallback需要在每次发消息时定义。具体来说，是在调用RabbitTemplate中的convertAndSend方法时，多传递一个参数：
这里的CorrelationData中包含两个核心的东西：
id：消息的唯一标示，MQ对不同的消息的回执以此做判断，避免混淆SettableListenableFuture：回执结果的Future对象
将来MQ的回执就会通过这个
Future
来返回，我们可以提前给
CorrelationData
中的
Future
添加回调函数来处理消息回执：
我们新建一个测试，向系统自带的交换机发送消息，并且添加
ConfirmCallback
：
@Test
void testPublisherConfirm() {
// 1.创建CorrelationData
CorrelationData cd = new CorrelationData();
// 2.给Future添加ConfirmCallback
cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
@Override
public void onFailure(Throwable ex) {
// 2.1.Future发生异常时的处理逻辑，基本不会触发
log.error("send message fail", ex);
}
@Override
public void onSuccess(CorrelationData.Confirm result) {
// 2.2.Future接收到回执的处理逻辑，参数中的result就是回执内容
if(result.isAck()){ // result.isAck()，boolean类型，true代表ack回执，false 代表 nack回执
log.debug("发送消息成功，收到 ack!");
}else{ // result.getReason()，String类型，返回nack时的异常描述
log.error("发送消息失败，收到 nack, reason : {}", result.getReason());
}
}
});
// 3.发送消息
rabbitTemplate.convertAndSend("hmall.direct", "q", "hello", cd);
}
CollelationData对象：这个对象里面有一个唯一ID（UUID）当前消息的一个标识。每次发消息都有一个CollelationData对象，因为每个消息都有自己的消息id。将来消息到MQ以后，MQ也能区分每个消息谁是谁。将来做回调的时候，每个消息的回调函数可能不同。
注意这个onFailure和onSuccess方法，是指回调有没有成功。不是指消息执行有没有成功。而onSuccess里根据ack还是nack，才能知道消息有没有发送成功！
执行结果如下：
可以看到，由于传递的
RoutingKey
是错误的，路由失败后，触发了
return callback
，同时也收到了ack。
当我们修改为正确的
RoutingKey
以后，就不会触发
return callback
了，只收到ack。
而如果连交换机都是错误的，则只会收到nack。
注意：
开启生产者确认比较消耗MQ性能，一般不建议开启。而且大家思考一下触发确认的几种情况：
- 路由失败：一般是因为RoutingKey错误导致，往往是编程导致
- 交换机名称错误：同样是编程错误导致
- MQ内部故障：这种需要处理，但概率往往较低。因此只有对消息可靠性要求非常高的业务才需要开启，而且仅仅需要开启ConfirmCallback处理nack就可以了。
2.MQ的可靠性
消息到达MQ以后，如果MQ不能及时保存，也会导致消息丢失，所以MQ的可靠性也非常重要。
2.1.数据持久化
为了提升性能，默认情况下MQ的数据都是在内存存储的临时数据，重启后就会消失。为了保证数据的可靠性，必须配置数据持久化，包括：
- 交换机持久化
- 队列持久化
- 消息持久化
我们以控制台界面为例来说明。
2.1.1.交换机持久化
在控制台的
Exchanges
页面，添加交换机时可以配置交换机的
Durability
参数：
设置为
Durable
就是持久化模式，
Transient
就是临时模式。
2.1.2.队列持久化
在控制台的Queues页面，添加队列时，同样可以配置队列的
Durability
参数：
除了持久化以外，你可以看到队列还有很多其它参数，有一些我们会在后期学习。
2.1.3.消息持久化
在控制台发送消息的时候，可以添加很多参数，而消息的持久化是要配置一个
properties
：
说明：在开启持久化机制以后，如果同时还开启了生产者确认，那么MQ会在消息持久化以后才发送ACK回执，进一步确保消息的可靠性。
不过出于性能考虑，为了减少IO次数，发送到MQ的消息并不是逐条持久化到数据库的，而是每隔一段时间批量持久化。一般间隔在100毫秒左右，这就会导致ACK有一定的延迟，因此建议生产者确认全部采用异步方式。
2.2.LazyQueue
在默认情况下，RabbitMQ会将接收到的信息保存在内存中以降低消息收发的延迟。但在某些特殊情况下，这会导致消息积压，比如：
- 消费者宕机或出现网络故障
- 消息发送量激增，超过了消费者处理速度
- 消费者处理业务发生阻塞
一旦出现消息堆积问题，RabbitMQ的内存占用就会越来越高，直到触发内存预警上限。此时RabbitMQ会将内存消息刷到磁盘上，这个行为成为
PageOut
.
PageOut
会耗费一段时间，并且会阻塞队列进程。因此在这个过程中RabbitMQ不会再处理新的消息，生产者的所有请求都会被阻塞。
为了解决这个问题，从RabbitMQ的3.6.0版本开始，就增加了Lazy Queues的模式，也就是惰性队列。惰性队列的特征如下：
- 接收到消息后直接存入磁盘而非内存
- 消费者要消费消息时才会从磁盘中读取并加载到内存（也就是懒加载）
- 支持数百万条的消息存储
而在3.12版本之后，LazyQueue已经成为所有队列的默认格式。因此官方推荐升级MQ为3.12版本或者所有队列都设置为LazyQueue模式。
2.2.1.控制台配置Lazy模式
在添加队列的时候，添加
x-queue-mod=lazy
参数即可设置队列为Lazy模式：
2.2.2.代码配置Lazy模式
在利用SpringAMQP声明队列的时候，添加
x-queue-mod=lazy
参数也可设置队列为Lazy模式：
@Bean
public Queue lazyQueue(){
return QueueBuilder
.durable("lazy.queue")
.lazy() // 开启Lazy模式
.build();
}
这里是通过
QueueBuilder
的
lazy()
函数配置Lazy模式，底层源码如下：
当然，我们也可以基于注解来声明队列并设置为Lazy模式：
@RabbitListener(queuesToDeclare = @Queue(
name = "lazy.queue",
durable = "true",
arguments = @Argument(name = "x-queue-mode", value = "lazy")
))
public void listenLazyQueue(String msg){
log.info("接收到 lazy.queue的消息：{}", msg);
}
2.2.3.更新已有队列为lazy模式
对于已经存在的队列，也可以配置为lazy模式，但是要通过设置policy实现。
可以基于命令行设置policy：
rabbitmqctl set_policy Lazy "^lazy-queue$" '{"queue-mode":"lazy"}' --apply-to queues
命令解读：
rabbitmqctl：RabbitMQ的命令行工具set_policy：添加一个策略Lazy：策略名称，可以自定义"^lazy-queue$"：用正则表达式匹配队列的名字'{"queue-mode":"lazy"}'：设置队列模式为lazy模式--apply-to queues：策略的作用对象，是所有的队列
当然，也可以在控制台配置policy，进入在控制台的
Admin
页面，点击
Policies
，即可添加配置：
2.1.消费者确认机制
为了确认消费者是否成功处理消息，RabbitMQ提供了消费者确认机制（Consumer Acknowledgement）。即：当消费者处理消息结束后，应该向RabbitMQ发送一个回执，告知RabbitMQ自己消息处理状态。回执有三种可选值：
- ack：成功处理消息，RabbitMQ从队列中删除该消息
- nack：消息处理失败，RabbitMQ需要再次投递消息
- reject：消息处理失败并拒绝该消息，RabbitMQ从队列中删除该消息
一般reject方式用的较少，除非是消息格式有问题，那就是开发问题了。因此大多数情况下我们需要将消息处理的代码通过
try catch
机制捕获，消息处理成功时返回ack，处理失败时返回nack.
由于消息回执的处理代码比较统一，因此SpringAMQP帮我们实现了消息确认。并允许我们通过配置文件设置ACK处理方式，有三种模式：
- **
none**：不处理。即消息投递给消费者后立刻ack，消息会立刻从MQ删除。非常不安全，不建议使用 - **
manual**：手动模式。需要自己在业务代码中调用api，发送ack或reject，存在业务入侵，但更灵活 auto：自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强，当业务正常执行时则自动返回ack. 当业务出现异常时，根据异常判断返回不同结果：- 如果是业务异常，会自动返回nack；- 如果是消息处理或校验异常，自动返回reject;
版权归原作者 一只特立独行的猪611 所有， 如有侵权，请联系我们删除。