确保MQ消息的可靠性

解决消息丢失问题，保证MQ的可靠性，就必须从3个方面入手：

• 确保生产者一定把消息发送到MQ
• 确保MQ不会将消息弄丢
• 确保消费者一定要处理消息

生产者重试机制：生产者发送消息时，出现了网络故障，导致与MQ的连接中断，解决就是当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 # 最大重试次数

• 注意：当网络不稳定的时候，利用重试机制可以有效提高消息发送的成功率。不过SpringAMQP提供的重试机制是阻塞式的重试，也就是说多次重试等待的过程中，当前线程是被阻塞的。
  如果对于业务性能有要求，建议禁用重试机制。如果一定要使用，请合理配置等待时长和重试次数，当然也可以考虑使用异步线程来执行发送消息的代码。

生产者确认机制

• 问题：

  • MQ内部处理消息的进程发生了异常
  • 生产者发送消息到达MQ后未找到Exchange
  • 生产者发送消息到达MQ的Exchange后，未找到合适的Queue，因此无法路由

• 针对上述情况，RabbitMQ提供了生产者消息确认机制，包括Publisher Confirm和Publisher Return两种。在开启确认机制的情况下，当生产者发送消息给MQ后，MQ会根据消息处理的情况返回不同的回执。

• 总结如下：

  • 当消息投递到MQ，但是路由失败时，通过Publisher Return返回异常信息，同时返回ack的确认信息，代表投递成功
  • 临时消息投递到了MQ，并且入队成功，返回ACK，告知投递成功
  • 持久消息投递到了MQ，并且入队完成持久化，返回ACK ，告知投递成功
  • 其它情况都会返回NACK，告知投递失败

• ack和nack属于Publisher Confirm机制，ack是投递成功；nack是投递失败。而return则属于Publisher Return机制

• 开启生产者确认：在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异步回调返回回执

• 定义ReturnCallback,每个RabbitTemplate只能配置一个ReturnCallback，因此我们可以在配置类中统一设置。

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());
            }
        });
    }
}

• 定义ConfirmCallback:由于每个消息发送时的处理逻辑不一定相同，因此ConfirmCallback需要在每次发消息时定义。
• 新建一个测试，向系统自带的交换机发送消息，并且添加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);
}

• 只有对消息可靠性要求非常高的业务才需要开启，而且仅仅需要开启ConfirmCallback处理nack就可以了。

MQ的可靠性

• 数据持久化:交换机持久化,队列持久化,消息持久化

• LazyQueue：在默认情况下，RabbitMQ会将接收到的信息保存在内存中以降低消息收发的延迟。但在某些特殊情况下，这会导致消息积压，比如：

• 消费者宕机或出现网络故障
• 消息发送量激增，超过了消费者处理速度
• 消费者处理业务发生阻塞

一旦出现消息堆积问题，RabbitMQ的内存占用就会越来越高，直到触发内存预警上限。此时RabbitMQ会将内存消息刷到磁盘上，这个行为成为PageOut. PageOut会耗费一段时间，并且会阻塞队列进程。因此在这个过程中RabbitMQ不会再处理新的消息，生产者的所有请求都会被阻塞。
惰性队列的特征如下：接收到消息后直接存入磁盘而非内存，消费者要消费消息时才会从磁盘中读取并加载到内存（也就是懒加载），支持数百万条的消息存储

控制台配置Lazy模式：在添加队列的时候，添加x-queue-mod=lazy参数即可设置队列为Lazy模式。
代码配置Lazy模式：在利用SpringAMQP声明队列的时候，添加x-queue-mod=lazy参数也可设置队列为Lazy模式

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

• 基于注解来声明队列并设置为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);
}

• 对于已经存在的队列，也可以配置为lazy模式，但是要通过设置policy实现。

rabbitmqctl set_policy Lazy "^lazy-queue$" '{"queue-mode":"lazy"}' --apply-to queues  

消费者的可靠性

• 消费者确认机制:为了确认消费者是否成功处理消息，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;

• 通过下面的配置可以修改SpringAMQP的ACK处理方式：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: none # 不做处理,aoto，自动ack

• 失败重试机制：当消费者出现异常后，消息会不断requeue（重入队）到队列，再重新发送给消费者。如果消费者再次执行依然出错，消息会再次requeue到队列，再次投递，直到消息处理成功为止。
  极端情况就是消费者一直无法执行成功，那么消息requeue就会无限循环，导致mq的消息处理飙升，带来不必要的压力。

• 为了应对上述情况Spring又提供了消费者失败重试机制：在消费者出现异常时利用本地重试，而不是无限制的requeue到mq队列，修改consumer服务的application.yml文件

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

• 结论：

  • 开启本地重试时，消息处理过程中抛出异常，不会requeue到队列，而是在消费者本地重试
  • 重试达到最大次数后，Spring会返回reject，消息会被丢弃

• 失败处理策略:在之前的测试中，本地测试达到最大重试次数后，消息会被丢弃。这在某些对于消息可靠性要求较高的业务场景下，显然不太合适了。因此Spring允许我们自定义重试次数耗尽后的消息处理策略，这个策略是由MessageRecovery接口来定义的，它有3个不同实现：

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

• 在consumer服务中定义处理失败消息的交换机和队列,定义一个RepublishMessageRecoverer，关联队列和交换机

@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");
}
@Bean
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
    return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
}

业务幂等性

• 幂等是一个数学概念，用函数表达式来描述是这样的：f(x) = f(f(x))，例如求绝对值函数。在程序开发中，则是指同一个业务，执行一次或多次对业务状态的影响是一致的。例如：

  • 根据id删除数据
  • 查询数据
  • 新增数据

• 但数据的更新往往不是幂等的，如果重复执行可能造成不一样的后果,所以，我们要尽可能避免业务被重复执行。

• 在实际业务场景中，由于意外经常会出现业务被重复执行的情况，例如：

  • 页面卡顿时频繁刷新导致表单重复提交
  • 服务间调用的重试
  • MQ消息的重复投递

• 唯一消息ID

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

//SpringAMQP的MessageConverter自带了MessageID的功能，我们只要开启这个功能即可。
//以Jackson的消息转换器为例：
@Bean
public MessageConverter messageConverter(){
    // 1.定义消息转换器
    Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();
    // 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
    jjmc.setCreateMessageIds(true);
    return jjmc;
}

• 业务判断:基于业务本身的逻辑或状态来判断是否是重复的请求或消息。

//以支付修改订单的业务为例，需要修改OrderServiceImpl中的markOrderPaySuccess方法
    @Override
    public void markOrderPaySuccess(Long orderId) {
        // 1.查询订单
        Order old = getById(orderId);
        // 2.判断订单状态
        if (old == null || old.getStatus() != 1) {
            // 订单不存在或者订单状态不是1，放弃处理
            return;
        }
        // 3.尝试更新订单
        Order order = new Order();
        order.setId(orderId);
        order.setStatus(2);
        order.setPayTime(LocalDateTime.now());
        updateById(order);
    }

延迟消息:对于超过一定时间未支付的订单，应该立刻取消订单并释放占用的库存。

• 如何保证支付服务与交易服务之间的订单状态一致性?

  • 首先，支付服务会正在用户支付成功以后利用MQ消息通知交易服务，完成订单状态同步。
  • 其次，为了保证MQ消息的可靠性，我们采用了生产者确认机制、消费者确认、消费者失败重试等策略，确保消息投递和处理的可靠性。同时也开启了MQ的持久化，避免因服务岩机导致消息丢失；
  • 最后，我们还在交易服务更新订单状态时做了业务幂等判断，避免因消息重复消费导致订单状态异常。

• 如果交易服务消息处理失败，有没有什么兜底方案?

  • 可以在交易服务设置定时任务，定期查询订单支付状态。这样即便MQ通知失败，还可以利用定时任务作为兜底方案，确保订单支付状态的最终一致性。

• 在RabbitMQ中实现延迟消息也有两种方案：

  • 死信交换机+TTL
  • 延迟消息插件

• 死信交换机:当一个队列中的消息满足下列情况之一时，可以成为死信（dead letter）

  • 消费者使用basic.reject或 basic.nack声明消费失败，并且消息的requeue参数设置为false
  • 消息是一个过期消息，超时无人消费
  • 要投递的队列消息满了，无法投递

• 如果一个队列中的消息已经成为死信，并且这个队列通过dead-letter-exchange属性指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机就称为死信交换机（Dead Letter Exchange）。而此时加入有队列与死信交换机绑定，则最终死信就会被投递到这个队列中。

• 死信交换机的作用：收集那些因处理失败而被拒绝的消息；收集那些因队列满了而被拒绝的消息；收集因TTL（有效期）到期的消息

• 注意：

  • RabbitMQ的消息过期是基于追溯方式来实现的，也就是说当一个消息的TTL到期以后不一定会被移除或投递到死信交换机，而是在消息恰好处于队首时才会被处理。
  • 当队列中消息堆积很多的时候，过期消息可能不会被按时处理，因此你设置的TTL时间不一定准确。

• DelayExchange插件:基于死信队列虽然可以实现延迟消息，但是太麻烦了。因此RabbitMQ社区提供了一个延迟消息插件来实现相同的效果。

• 下载并安装插件：docker exec -it mq rabbitmq-plugins enable rabbitmq_delayed_message_exchange

• 声明延迟交换机

//基于注解的方式
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "delay.queue", durable = "true"),
        exchange = @Exchange(name = "delay.direct", delayed = "true"),
        key = "delay"
))
public void listenDelayMessage(String msg){
    log.info("接收到delay.queue的延迟消息：{}", msg);
}

//基于@Bean的方式
@Slf4j
@Configuration
public class DelayExchangeConfig {

    @Bean
    public DirectExchange delayExchange(){
        return ExchangeBuilder
                .directExchange("delay.direct") // 指定交换机类型和名称
                .delayed() // 设置delay的属性为true
                .durable(true) // 持久化
                .build();
    }

    @Bean
    public Queue delayedQueue(){
        return new Queue("delay.queue");
    }
    
    @Bean
    public Binding delayQueueBinding(){
        return BindingBuilder.bind(delayedQueue()).to(delayExchange()).with("delay");
    }
}

• 发送延迟消息

@Test
void testPublisherDelayMessage() {
    // 1.创建消息
    String message = "hello, delayed message";
    // 2.发送消息，利用消息后置处理器添加消息头
    rabbitTemplate.convertAndSend("delay.direct", "delay", message, new MessagePostProcessor() {
        @Override
        public Message postProcessMessage(Message message) throws AmqpException {
            // 添加延迟消息属性
            message.getMessageProperties().setDelay(5000);
            return message;
        }
    });
}

• 延迟消息插件内部会维护一个本地数据库表，同时使用Elang Timers功能实现计时。如果消息的延迟时间设置较长，可能会导致堆积的延迟消息非常多，会带来较大的CPU开销，同时延迟消息的时间会存在误差。
  因此，不建议设置延迟时间过长的延迟消息。

在交易服务中利用延迟消息实现订单超时取消功能

• 定义常量

package com.hmall.trade.constants;

public interface MQConstants {
    String DELAY_EXCHANGE_NAME = "trade.delay.direct";
    String DELAY_ORDER_QUEUE_NAME = "trade.delay.order.queue";
    String DELAY_ORDER_KEY = "delay.order.query";
}

• 配置MQ

  <!--amqp-->
  <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>

• 加MQ的配置

spring:
  rabbitmq:
    host: 192.168.150.101
    port: 5672
    virtual-host: /hmall
    username: hmall
    password: 123

• 改造下单业务，发送延迟消息，查询支付状态