生产者的可靠性

生产者重连

有的时候由于网络波动，可能会出现客户端连接MQ失败的问题，我们可以通过配置重连机制来解决这种情况：

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

生产者确认

RabbitMQ 提供了 Publisher Confirm 和 Publisher Return 两种确认机制。开启确机制后，在 MQ 成功收到消息后会返回确认消息给生产者。返回的结果有以下几种情况：

• 消息投递到了 MQ，但是路由失败。此时会通过 Publisher Return 返回路由异常原因，然后返回 ACK，告知投递成功
• 临时消息投递到了 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，因此需要在项目启动过程中配置：

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;
import org.springframework.beans.BeansException;

@Slf4j
@Configuration
public class CommonConfig implements ApplicationContextAware {

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        // 获取RabbitTemplate
        RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
        // 设置ReturnCallback
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("消息发送失败，应答码={}，原因={}，交换机={}，路由键={}，消息={}",
                    replyCode, replyText, exchange, routingKey, message.toString());
        });
    }
}

3.发送消息，指定ID,消息ConfirmCallback:
 

@Test
void testPublisherConfirm() throws InterruptedException {
    // 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("handle message ack 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", "red1", "hello", cd);
}

MQ可靠性

数据持久化

在默认情况下，RabbitMQ 会将接收到的信息保存在内存中以降低消息收发的延迟。这样会导致两个问题：

• 一旦 MQ 宕机，内存中的消息会丢失
• 内存空间有限，当消费者故障或处理过慢时，会导致消息积压，引发 MQ 阻塞

消息持久化需要把delivery_mode改成2

java在发送消息时默认是持久化的，如果想要发送临时消息，需要自己定义一个message.

LazyQueue

从 RabbitMQ 的 3.6.0 版本开始，就增加了 Lazy Queue 的概念，也就是惰性队列。

惰性队列的特征如下：

• 接收到消息后直接存入磁盘而非内存（内存中只保留最近的消息，默认 2048 条）
• 消费者要消费消息时才会从磁盘中读取并加载到内存
• 支持数百万条的消息存储

在 3.12 版本后，所有队列都是 Lazy Queue 模式，无法更改。

RabbitMQ 如何保证消息的可靠性

• 首先通过配置可以让交换机、队列、以及发送的消息都持久化。这样队列中的消息会持久化到磁盘，MQ 重启消息依然存在。
• RabbitMQ 在 3.6 版本引入了 LazyQueue，并且在 3.12 版本后会称为队列的默认模式。LazyQueue 会将所有消息都持久化。
• 开启持久化和生产者确认时，RabbitMQ 只有在消息持久化完成后才会给生产者返回 ACK 回执。

消费者可靠性

消费者确认机制

为了确认消费者是否成功处理消息，RabbitMQ 提供了消费者确认机制（Consumer Acknowledgement）。当消费者处理消息结束后，应该向 RabbitMQ 发送一个回执，告知 RabbitMQ 自己消息处理状态。回执有三种可选值：

• ack：成功处理消息，RabbitMQ 从队列中删除该消息
• nack：消息处理失败，RabbitMQ 需要再次投递消息
• reject：消息处理失败并拒绝该消息，RabbitMQ 从队列中删除该消息

失败重试机制

当消费者出现异常后，消息会不断 requeue（重新入队）到队列，再重新发送给消费者，然后再次异常，再次 requeue，无限循环，导致 mq 的消息处理飙升，带来不必要的压力。

我们可以利用 Spring 的 retry 机制，在消费者出现异常时利用本地重试，而不是无限制的 requeue 到 mq 队列：

失败消息处理机制

在开启重试模式后，重试次数耗尽，如果消息依然失败，则需要有 MessageRecoverer 接口来处理，它包含三种不同的实现：

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

建议使用第三种方案：

那么重试耗尽后，最终将失败消息投递到error.queue中

使用步骤

1.

将失败处理策略改为 RepublishMessageRecoverer：

① 首先，定义接收失败消息的交换机、队列及其绑定关系，此处略；

② 然后，定义 RepublishMessageRecoverer：

@Bean
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate) {
    return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
}

2.

@Configuration
public class ErrorConfiguration {

    @Bean
    public DirectExchange errorExchange() {
        return new DirectExchange("error.direct");
    }

    @Bean
    public Queue errorQueue() {
        return new Queue("error.queue");
    }

    @Bean
    public Binding errorBinding(Queue errorQueue, DirectExchange errorExchange) {
        return BindingBuilder.bind(errorQueue).to(errorExchange).with("error");
    }

    @Bean
    public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate) {
        return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
    }
}

消费者如何保证消息一定被消费？

• 开启消费者确认机制为 auto，由 spring 确认消息处理成功后返回 ack，异常时返回 nack
• 开启消费者失败重试机制，并设置 MessageRecoverer，多次重试失败后将消息投递到异常交换机，交由人工处理

业务幂等性

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

唯一消息 id

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

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

@Bean
public MessageConverter messageConverter() {
    // 1. 定义消息转换器
    Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();
    // 2. 配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
    jjmc.setCreateMessageIds(true);
    return jjmc;
}

业务判断

方案二，是结合业务逻辑，基于业务本身做判断。以我们的业务为例：我们要在支付后修改订单状态为已支付，应该在修改订单状态前先查询订单状态，判断状态是否是未支付。只有未支付订单才需要修改，其它状态不做处理。

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

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

---

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

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

延迟消息

延迟消息：生产者发送消息时指定一个时间，消费者不会立刻收到消息，而是在指定时间之后才收到消息。

死信交换机

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

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

如果队列通过 dead-letter-exchange 属性指定了一个交换机，那么该队列中的死信就会投递到这个交换机中。这个交换机称为死信交换机（Dead Letter Exchange，简称 DLX）。

当生产者生产的消息路由到simple.queue后，由于该队列没有绑定消费者，实践到了之后消息自动变成死信，最后发送给consumer，从而实现了延迟消息。

但是一般不使用这种方案，因为过于繁琐了。

延迟消息插件

步骤

1.创建消费者，接收消息

2.发送消息，用setDelay设置延迟时间

取消超时订单

设置 30 分钟后检测订单支付状态实现起来非常简单，但是存在两个问题：

• 如果并发较高，30 分钟可能堆积消息过多，对 MQ 压力很大
• 大多数订单在下单后 1 分钟内就会支付，但是却需要在 MQ 内等待 30 分钟，浪费资源

解决方法，将30min拆分成多个较短的时间周期，减少消息的堆积。

业务流程