消息可靠性是什么？
消息可靠性（Message Reliability）是指在消息传递系统中，确保消息从发送方传递到接收方的过程中不丢失、不重复、按顺序到达且不会被篡改的一系列技术和机制。
在消息传输过程中，不免会出现消息丢失的情况。
生产者可靠性
生产者重连
有时候由于网络波动，可能出现客户端连接MQ失败的情况，我们可以通过配置开启连接失败后的重连机制
spring:
rabbitmq:
host: 192.168.150.101
port: 5672
virtual-host: /hmall
username: hmall
password: 123
connection-timeout: 1s #设置MQ的连接超时时间
template:
retry:
enabled: true #开启超时重试机制
initial-interval：1000ms #失败后的初始等待时间
multiplier: 2 #失败后下次的等待时常倍数，下次等待时长=initial-interval *
multiplier max-attempts: 3 #最大重试次数
注意：
当网络不稳定时，利用重试机制可以有效提高消息发送的成功率。不过Spring AMQP提供的重试机制是阻塞式的重试，也就是说多次重试等待的过程中，当前线程是被阻塞的，会影响业务性能。
如果对与业务性能有要求，建议禁用重试机制，如果一定要用，需要合理配置等待时长和重试次数，当然也可以考虑使用异步线程来执行发送消息的代码。
生产者确认
RabbitMQ提供了Publisher Confirm 和 Publisher Return两种确认机制。开启确认机制后，在MQ成功收到消息后会返回确认消息给生产者。返回结果如下：
- 消息投递到了MQ，但是路由失败（通常由程序编写导致），此时会通过Publisher Return 返回路由异常原因，然后返回ACK，告知投递成功。
- 临时消息投递到了MQ，并且入队成功，返回ACK，告知投递成功。
- 持久消息投递到了MQ，并且入队完成持久化，返回ACK，告知投递成功。
- 其他情况都会返回NACK，告知投递失败。
Publisher Confirm：用于确认消息是否被 RabbitMQ 成功接收和持久化。
Publisher Return：用于捕获由于路由键不匹配而无法投递到任何队列的消息。
spring:
rabbitmq:
publisher-confirm-type:
correlated #none：关闭confirm机制，simple：#同步阻塞等待MQ的回执消息，correlated：#MQ异步回调方式返回回执消息
publisher-returns: true #开启Publisher Return机制
@Slf4j
@Configuration
public class MqConfirmConfig implements ApplicationContextAware {
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class); // 配置回调
rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
@Override
public void returnedMessage(ReturnedMessage returned) {
log.debug("收到消息的return callback，exchange:{}, key:{}, msg:{}, code:{}, text:{}",
returned.getExchange(), returned.getRoutingKey(), returned.getMessage(),returned.getReplyCode(), returned.getReplyText());}
});
}
}
@Test
void testConfirmCallback() throws InterruptedException {
// 1.创建cd
CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
// 2.添加ConfirmCallback
cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
@Override
public void onFailure(Throwable ex) {
log.error("消息回调失败", ex); }
@Override
public void onSuccess(CorrelationData.Confirm result) {
log.debug("收到confirm callback回执");
if(result.isAck()){
// 消息发送成功
log.debug("消息发送成功，收到ack");
}else{
// 消息发送失败
log.error("消息发送失败，收到nack， 原因：{}", result.getReason());
}
}
});
rabbitTemplate.convertAndSend("hmall.direct123", "red2", "hello", cd);
Thread.sleep(2000);
}
注：
生产者确认需要额外的网络和系统资源开销，尽量不要使用
如果一定要用，无需开启Publisher Return机制，因为一般路由失败是自己的业务问题导致的
对于nack消息可以有限次数重试，依然失败则记录异常消息
MQ可靠性
在默认情况下，RabbitMQ会将接收到的消息保存在内存中以降低消息收发的延迟，这样会导致两个问题：
- 一旦MQ宕机，内存中的消息会丢失2. 内存空间有限，当消费者故障或者处理过慢时，会导致消息积压，引发MQ阻塞
生产者向队列中发送消息，队列满了之后，MQ会将旧消息持久化到磁盘，但在这个过程中，MQ是阻塞状态
数据持久化
MQ实现数据持久化包括三个方面
- 交换机持久化
- 队列持久化 （使用Spring创建的交换机和队列默认就是持久化的）
- 消息持久化 （发现性能还是会有所下降，与理想状态不同）
Lazy Queue
从RabbitMQ的3.6.0版本开始，就增加了Lazy Queue的概念，也就是惰性队列
惰性队列特征如下：
- 接受消息后直接存入磁盘而非内存（内存中只保留最近的消息，默认2048条）2. 消费者要消费消息时才会从磁盘中读取并加载到内存3. 支持数百万条到消息存储
在3.12版本之后，所有队列都是Lazy Queue模式，无法更改。
消费者可靠性
消费者确认机制
为了确认消费者是否成功处理消息，RabbitMQ提供了消费者确认机制。当消费者处理消息结束后，应该向RabbitMQ发送一个回执，告知RabbitMQ自己消息处理状态，回执有三种可选值：
- ack：成功处理消息，RabbitMQ从队列中删除该消息
- nack：消息处理失败，RabbitMQ需要再次投递消息
- reject：消息处理失败并拒绝该消息，RabbitMQ从队列中删除该消息
SpringAMQP已经实现了消息确认功能。并允许我们通过配置文件选择ACK处理方式，有三种：
- none：不处理，即消息投递给消费者后立刻ack，消息会立刻从MQ删除，非常不安全，不建议使用
- manual：手动模式，需要自己在业务代码中调用API，发送ack或reject，存在业务入侵，但更灵活
- auto：自动模式，SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强， 1. 当业务正常执行时则自动返回ack，2. 当业务出现异常时，根据异常判断返回不同结果 1. 如果时业务异常，会自动返回nack2. 如果是消息处理或校验异常，自动返回reject
spring:
rabbitmq:
host: 192.168.150.101
port: 5672 virtual-host: /hmall
username: hmall
password: 123
listener:
simple:
prefetch: 1
acknowledge-mode: auto # 确认机制
失败重试机制
当消费者出现异常后，消息会不断requeue（重新入队）到队列，再重新发送给消费者，然后再次异常，再次requeue，无限循环，导致mq的消息处理飙升，带来不必要的压力。
我们可以利用Spring的retry 机制，在消费者出现异常时利用本地重试，而不是无限制requeue到mq队列
spring:
rabbitmq:
host: 192.168.150.101
port: 5672 virtual-host: /hmall
username: hmall
password: 123
listener:
simple:
prefetch: 1
acknowledge-mode: auto # 确认机制
retry:
enabled: true # 开启重试机制
失败消息处理策略
在开启重试模式后，重试次数耗尽，如果消息依然失败，则需要有MessageRecoverer接口来处理，它包含三种不同的实现：
- RejectAndDontRequeueRecoverer：重试耗尽后，直接reject，丢弃消息。默认
- ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入对
- RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机
业务幂等性
在程序开发中，则指同一个业务，执行一次或者多次对业务状态的影响是一致的。
唯一消息ID
给每个消息都设置一个唯一ID，利用ID区分是否是重复消息：
- 每一条消息都生成一个唯一的ID，与消息一起投递给消费者
- 消费者接收到消息后处理自己的业务，业务处理成功后将消息ID保存到数据库
- 如果下次又收到相同消息，去数据库查询判断是否存在，存在则为重复消息放弃处理
@Bean
public MessageConverter jacksonMessageConvertor(){
Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();
jjmc.setCreateMessageIds(true);
return jjmc;
}
业务判断
结合业务逻辑，基于业务本身做判断，已订单业务为例：支付后修改订单状态为已支付，应该在修改订单状态前先查询订单状态，判断状态是否是未支付。只有未支付订单才需要修改状态，其他状态不做处理。
相关问题
如果交易服务消息处理失败，有没有什么兜底方案？
- 使用失败消息处理策略，可以将失败的消息进行人工处理
- 也可以在交易服务设置定时任务，不再是订单支付之后向交易服务发送消息，而是由交易服务主动去拉去消息，定期查询订单支付状态。即便MQ通知失败，还可以利用定时任务作为兜底方案，确保订单支付状态的最终一致性。
延迟消息
什么是延迟消息？
生产者发送消息时指定一个时间，消费者不会立刻收到消息，而是在指定时间之后才收到消息
什么是延迟任务？
设置在一定时间之后才执行任务
死信交换机
延迟消息插件
RabbitMQ 的官方也推出了一个插件，原声支持延迟消息功能。该插件的原理是设计了一种支持延迟消息功能的交换机，当消息投递到交换机后可以暂存一定时间，到期后再投递到队列。（定时任务对cpu占用较大，适用于延时时间相对较短的）
取消超时订单
可以将30分钟进行拆分，形成一个延时时间组
版权归原作者 不止会Java 所有， 如有侵权，请联系我们删除。