RabbitMQ确保消息可靠性

消息丢失的可能性

支付服务先扣减余额和更新支付状态（这俩是同步调用），然后通过RabbitMq异步调用支付服务更新订单状态。但是有些情况下，可能订单已经支付 ，但是更新订单状态却失败了，这就出现了消息丢失。

1. 发送者在发送的过程中出现了网络故障
2. RabbitMQ在发送消息的过程中出现了问题
3. 消费者在更新订单状态的时候出现了问题

发送者的可靠性

发送者确认机制需要与MQ进行通信和确认，会影响消息发送的效率且一般出现的概率极低，所以一般不用这个。

---

方法1. 发送者重连

确保发送者与MQ之间连接的可靠性。有的时候由于网络波动，可能出现发送者连接MQ失败的情况，这个配置是关闭的，可以开启连接失败后的重连机制：

spring:rabbitmq:connection-timeout: 1s # 设置MQ的连接超时时间template:retry:enabled: true # 开启超时重试机制(默认是false)initial-interval: 1000ms # 失败后的初始等待时间multiplier: 1 # 失败后下次的等待时长倍数，下次等待时长 = initial-interval * multipliermax-attempts: 3 # 最大重试次数

【注】：当网络不稳定时，利用重试机制可以提高消息发送的成功率，但是SpringAMQP提供的重试机制是阻塞式的重试，如果需要多次重试等待，当前线程被阻塞，会影响性能。
如果对业务性能有要求，建议禁用重试机制，如果一定要使用，要合理的配置等待时常和重试次数，或使用异步线程来执行发送消息的代码。

方法2. 发送者确认

确保消息发送的可靠性。SpringAMQP提供了Publisher Confirm和Publisher Return两种机制，开启确认机制后，当发送者发送消息给MQ后，MQ会返回确认结果给发送者，返回的结果有以下几种情况：

• 消息投递到MQ，但是路由失败，此时通过PublisherReturn返回路由异常信息，然后返回ACK，告知投递成功。例如：
  • 消息发送给图中的exchange1，但是RoutingKey写错了，没有匹配到正确的队列，也会导致路由失败。
  • 消息发送给图中的exchange2，但是它底下没有绑定新的队列，就会导致路由失败。
• 临时消息【不需要往磁盘做持久化的消息】投递到MQ，并入队成功，返回ACK，告知投递成功。
• 持久消息投递到MQ，并入队完成持久化，返回ACK，告知投递成功。
• 其他情况都会返回NACK，告知投递失败。
  
  步骤：

1. 在发送方publisher所在的微服务的application.yml中配置：

spring:rabbitmq:publisher-confirm-type: correlated # 开启publisher confirm机制，并设置confirm类型publisher-returns: true # 开启publisher return机制

publisher-confirm-type有三种模式：

• none：关闭confirm机制
• simple：同步阻塞等待MQ回执消息
• correlated：MQ异步回调方式返回回执消息（常用）

2. 开启回调机制：每个RabbitTemplate只能配置一个ReturnCallback，在发送者publisher所在的项目启动时配置即可。

@Configuration
@RequiredArgsConstructor
@Slf4j
public class MqConfig {private final RabbitTemplate rabbitTemplate;@PostConstruct // 在Bean初始化完成后调用这个方法（只会调用一次）public void init() {// 返回ACK，但是此时路由失败，就会走这个方法rabbitTemplate.setReturnsCallback(returnedMessage -> {log.error("监听到了消息return callback");log.debug("exchange: {}", returnedMessage.getExchange());log.debug("routingKey: {}", returnedMessage.getRoutingKey());log.debug("message: {}", returnedMessage.getMessage());log.debug("replyCode: {}", returnedMessage.getReplyCode());log.debug("replyText: {}", returnedMessage.getReplyText());});}
}

3. 开启消息确认机制：发送消息、指定消息ID、每次发送消息都需要配置一个ConfirmCallback

public void testConfirmCallback() {CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {// 【几乎不可能发生】Future发生异常时的处理逻辑@Overridepublic void onFailure(Throwable ex) {log.error("spring ampq处理确认结果异常", ex);}// 成功，拿到MQ结果，判断是ACK还是NACK@Overridepublic void onSuccess(CorrelationData.Confirm result) {if(result.isAck()) {// ACKlog.debug("收到ACK，消息发送成功");}else {// NACKlog.debug("收到NACK，消息发送失败，失败原因：{}",result.getReason());}}});rabbitTemplate.convertAndSend("hmall.direct", "blue", "hello world", cd); // 发送消息
}

MQ的可靠性

RabbitMQ一般会将收到的信息保存到内存（速度快）中，降低消息收发的延迟，这样会导致：

1. MQ宕机，内存中的消息会丢失。
2. 内存空间有限，消费者故障或处理过慢，会导致消息积压，引发MQ阻塞。
   

【案例】发送者往MQ发消息，MQ会把数据保存到内存中，如果内存满了，MQ就会把一部分数据迁移到磁盘中暂时进行持久化存储，移动到磁盘的这段时间发送者发送的消息就会产生丢失。

方法1. 数据持久化

数据持久化就是把数据持久化到磁盘，但是不是向上边那个案例，等满了再去持久化（被动），而是提前进行持久化。

1. 交换机的持久化（默认开启的）
   
2. 队列的持久化（默认开启的）
   
3. 消息持久化（默认是非持久的）
   在发送消息的时候设定的
   
   【案例】：比较一下持久化和非持久化的性能。
   发100w条消息给MQ：
   这是非持久化的方式：使用纯内存的方式存储，每次内存满之后，MQ就会把消息写到磁盘中，此时就会出现阻塞状态，处理速度降低到0
   
   【问题】可能出现消息丢失和MQ阻塞
   【解决办法】使用持久化的方式：

public void testSendPersistentMsg() {// 自定义构建消息Message msg = MessageBuilder.withBody("hello world".getBytes(StandardCharsets.UTF_8)) // 消息体.setDeliveryMode(MessageDeliveryMode.PERSISTENT) // 投递模式（持久化）.build();for (int i = 0; i < 100000; ++i) {rabbitTemplate.convertAndSend("simple.queue", msg);}
}

Mq并没有阻塞，每发一条消息就赶紧把它存到磁盘中，和纯内存方式相比，不会有个中断的过程。

方法2. Lazy Queue（推荐）

【问题】：由于使用了消息持久化的方式，发到MQ的消息不仅要到内存，还要在磁盘中写一份，这会导致整体的并发能力下降
【特征】：

• 接收到消息后直接入磁盘，不再存储到内存
• 在写磁盘的时候也对写入磁盘的操作进行一些优化，比传统的写操作高很多
• 消费者要消费消息时，才会从磁盘中读取并加载到内存
  • 【问题】：可能会影响消费者处理消息的速度
  • 【解决】：可以提前缓存部分消息到内存，最多2048条

控制台声明Lazy Queue队列

Java代码添加

声明Bean

@Bean
public Queue lazyQueue(){return QueueBuilder.durable("lazy.queue").lazy() // 开启Lazy模式.build();
}

@RabbitListener注解

@RabbitListener(queuesToDeclare = @Queue(name = "lazy.queue",durable = "true",arguments = @Argument(name = "x-queue-mode", value = "lazy") // 开启Lazy模式
))
public void listenLazyQueue(String msg){log.info("接收到 lazy.queue的消息：{}", msg);
}

消费者的可靠性

消费者确认机制

为了确认消费者是否成功处理消息，当消费者处理消息结束后，应该向MQ发送一个回执，告知MQ自己的消息处理状态。有如下几种消息处理状态：

• ack：处理消息成功，RabbitMQ从队列中删除该消息
• nack：消息处理失败，RabbitMQ需要再次投递消息
• reject：消息处理失败并拒绝该消息，RabbitMQ从队列中删除该消息【在处理的过程中，发现消息的内容有问题，没有重试的必要，直接拒绝就行】

【注意】：不管是哪种情况，都应该等消息处理完后得到结果再返回，不要一拿到消息就返回
返回消息处理状态的过程，类似于处理事务，事务处理成功，返回ACK；处理失败，返回NACK

SpringAMQP允许通过在消费者的配置文件选择ACK的处理方式，有三种：

• none：不处理，消息投递给消费者后立刻ack，消息会立刻从MQ中删除，别用
• manual：手动模式，需要在业务代码中调用api，发送ack或reject，存在业务入侵，但是更灵活。
• auto：自动模式，利用AOP对消息处理逻辑进行了环绕增强
  • 业务处理正常：自动返回ack
  • 业务处理异常：自动返回nack
  • 消息处理或校验异常【MessageConversionException】：自动返回reject

spring:rabbitmq:listener:simple:acknowledge-mode: auto# 不做处理

失败重试策略

在消费者出现异常时，利用本地重试，而不是无限的重新入队到mq，可以在消费者的yaml文件中添加配置来开启重试机制。

spring:rabbitmq:listener:simple:retry:enabled: true # 开启消费者失败重试initial-interval: 1000ms # 初识的失败等待时长为1秒multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-intervalmax-attempts: 3 # 最大重试次数（达到最大重试次数后，MQ会把消息丢弃）stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false

【问题】：在开启重试模式后，重试次数耗尽，如果消息仍然失败，默认会把消息进行丢弃。
【解决】：因此需要有MessageRecoverer接口来处理，包含三种不同的实现：

• RejectAndDontRequeueRecoverer（默认）：重试耗尽后，直接reject，丢弃消息。
• ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队。
• RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机 。
  

修改失败重试策略为RepublishMessageRecoverer

1. 定义接收失败的交换机、队列、
2. 定义RepublishMessageRecoverer

@Configuration
public class ErrorMessageConfiguration {// 定义接收失败的交换机@Beanpublic DirectExchange errorExchange() {return new DirectExchange("error.direct");}// 定义接收失败的队列@Beanpublic Queue errorQueue() {return new Queue("error.queue");}// 定义绑定关系@Beanpublic Binding errorQueueBinding() {return BindingBuilder.bind(errorQueue()).to(errorExchange()).with("error");}// 定义失败处理策略@Beanpublic MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate) {return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");}
}

业务幂等性

f(x) = f(f(x))，指同一个业务，执行一次或多次对业务状态的影响是一致的。

• 幂等业务：查询业务、删除业务
• 非幂等业务：用户下单需要扣减库存、用户退款业务需要恢复余额

方案1. 唯一消息id

给每个消息设置一个唯一id，利用id区分是否是重复消息：

• 每条消息都生成一个唯一id，与消息一起投递给消费者
• 消费者接收到消息后处理自己的业务，业务处理成功后将消息id保存到数据库中
• 如果下次又收到相同消息，去数据库查询判断是否存在，存在则视为重复消息放弃处理

1. 在发送方配置Bean用来自动创建消息id

@Configuration
public class MqConfig {@Beanpublic MessageConverter messageConverter() {Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();converter.setCreateMessageIds(true); // 配置自动创建消息idreturn converter;}
}

2. 在接收方接收消息id

@RabbitListener(queues = "simple.queue")public void listenSimpleQueue(Message msg) { // 使用字符串发送，就用字符串接收log.info("监听到simple.queue的消息：{}", msg);log.info("消息id：{}", msg.getMessageProperties().getMessageId());// throw new RuntimeException("故意的");}

方案2. 业务判断（常用）

结合业务逻辑，基于业务本身做判断。
【案例】：当用户下单成功后，通过MQ通知交易服务来修改订单状态为已支付（这里记作消息1），修改成功后交易服务返回ACK给MQ，此时出现了网络的故障，MQ没有收到交易服务发送的ACK，MQ认为交易服务宕机，消息又重新入队。
就在此刻，用户点击了申请退款，直接向交易服务修改订单状态为退款中（这个操作没有走MQ，此时订单状态是退款中，但是消息1还在消息队列中）。
此时网络恢复了，MQ又将消息1发送给交易服务，此时交易服务又把订单状态标记为已支付（订单申请退款中的状态又被覆盖了）。

【解决】：通知来的时候，先判断订单的状态，再进行操作。

@Component
@RequiredArgsConstructor
public class PayStatusListener {private final IOrderService orderService;@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "trade.pay.success.queue", durable = "true"),exchange = @Exchange(name = "pay.direct"),key = "pay.success"))public void listenPaySuccess(Long orderId) {// 1.查询订单Order order = orderService.getById(orderId);// 2.判断订单状态是否为未支付if(order == null || order.getStatus() != 1) {// 不做处理return;}// 3.标记订单状态为已支付orderService.markOrderPaySuccess(orderId);}
}

延迟消息

延迟消息：发送者发送消息时指定一个时间，消费者不会立刻收到消息，而是在指定时间之后才收到消息。
延迟任务：设置在一定时间后才执行的任务。

方案1. 死信交换机

当一个队列中的消息满足下列情况之一的，就会成为死信：

• 消费者使用basic.reject或basic.nack声明消费失败，并且消息的requeue参数设置为false。
• 消息是一个过期消息（达到队列设置的过期时间 或 消息本身设置的过期时间），超时无人消费。
• 要投递的队列消息堆积满了，最早的消息可能成为死信。

队列通过dead-letter-exchange属性指定了一个交换机，那么该队列中的死信就会投递到这个交换机中，这个交换机就叫做死信交换机（DLX）。

1. 声明死信队列、死信交换机、它们之间的绑定关系：

@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "dlx.queue", durable = "true"), // 死信队列exchange = @Exchange(name = "dlx.direct", type = ExchangeTypes.DIRECT), // 死信交换机key = {"hi"} ))public void listenDlxQueue(String msg) {log.info("消费者监听到dlx.queue的消息: " + msg);}

2. 声明普通队列、普通交换机、它们之间的绑定关系，并把队列绑定到死信交换机上（此时就不需要把它绑定消费者了）：

@Configuration
public class NormalConfiguration {@Beanpublic DirectExchange normalExchange() { // 普通交换机return ExchangeBuilder.directExchange("normal.direct").build();}@Beanpublic Queue normalQueue() { // 普通队列return QueueBuilder.durable("normal.direct") // 队列名字.deadLetterExchange("dlx.direct") // 死信交换机名字.build();}@Beanpublic Binding normalQueueBinding(Queue normalQueue, DirectExchange normalExchange) { // 绑定关系// 把队列绑定到交换机return BindingBuilder.bind(normalQueue) // 队列.to(normalExchange) // 交换机.with("hi");// 这里绑定关系要和普通队列的绑定关系保持一致}
}

3. 发送延迟消息：

@Test
public void testSendDelayMsg() {rabbitTemplate.convertAndSend("normal.direct", "hi", "hello world", message -> {// 当消息被转成Message对象后，还可以进一步做加工message.getMessageProperties().setExpiration("10000"); // 设置消息过期时间(10s)return message;});
}

【注】：normal.direct和normal.queue之间绑定的BindingKey 与 dlx.direct和dlx.queue之间绑定的BindingKey要一致

方案2. 延迟消息插件DelayExchange（推荐）

这个插件可以将普通交换机改造为支持延迟消息功能的交换机，当消息投递到交换机后，可以暂存一段时间，到后期再投递到队列。

一、安装插件

1. 插件下载地址：DelayExchange
2. 需要把插件放在RabbitMQ插件目录对应的数据卷下

docker volume inspect mq-plugins

3. 执行命令，安装插件

docker exec -it rabbitmq rabbitmq-plugins enable rabbitmq_delayed_message_exchange

二、使用插件

1. 声明延迟交换机：只要设置delay的属性为true即可

@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "delay.queue", durable = "true"),exchange = @Exchange(name = "delay.direct", delayed = "true", type = ExchangeTypes.DIRECT), // 只要设置一个delayed属性为true即可key = {"hi"}
))
public void listenDelayQueue(String msg) {log.info("消费者监听到delay.queue的消息: " + msg);
}

2. 发送延迟消息：通过消息头x-delay来设置过期时间

@Test
public void testSendDelayMsgByPlugin() {rabbitTemplate.convertAndSend("delay.direct", "delay", "hello world", message -> {message.getMessageProperties().setDelay(10000);// 添加延迟消息属性return message;});
}

延迟消息的实现需要记录消息的过期时间，计时的时钟需要依赖cpu，是个cpu密集型任务。因此使用延迟消息时，需要避免同一时刻在mq里存在大量的延迟消息（尽可能地让延迟消息的延迟时间不要太长）。