Featured image of post 服务异步通讯 - MQ
后端技术栈

服务异步通讯 - MQ

服务异步通讯

MQ的一些常见问题

  • 消息可靠性问题: 确保发送的消息至少被消费一次

  • 延迟消息问题: 如何实现消息的延迟投递

  • 高可用性问题: ru和避免单点的MQ故障而导致的不可用问题

  • 消息堆积问题: 如何解决数百万消息堆积，无法即使消费的问题

消息可靠性

导致消息丢失的可能性:

  • 发送时丢失
    • 生产者发送的消息未送达exchange
    • 消息到达exchange后未到达queue
  • MQ宕机，queue将消息丢失
  • consumer接收到消息后未消费就宕机

生产者消息确认

RabbitMQ提供了publisher confirm机制来避免消息发送到MQ过程中丢失。消息发送到MQ后，会返回一个结果给发送者，表示消息是否处理成功。

  • publisher-confirm, 发送者确认
    • 消息成功投递到交换机，返回ack
    • 消息未成功到交换机，返回nack
  • publisher-return, 发送者回执
    • 消息投递到交换机了，但是没有路由到队列。返回ACK，及路由失败原因。

注意: 确认机制发送消息时，需要给每个消息设置一个全局唯一id,以区分不同消息，避免ack冲突

image-20240219211628578

SpringAMQP实现生产者确认

在publicsher这个微服务中的application.yml中添加配置

spring:
  rabbitmq:
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true

每个RabbitTemplate只能配置一个ReturnCallback，因此需要在启动项目时配置

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

发送消息，指定消息ID、消息ConfirmCallback

@Test
public void testSendMessage2SimpleQueue() throws InterruptedException {
    // 消息体
    String message = "hello, spring amqp!";
    // 消息ID, 确保消息唯一CorrelationData中
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    // 注册callback
    correlationData.getFuture().addCallback(
        result -> {
            if(result.isAck()){
                // ack, 消息成功
                log.debug("消息发送确认成功, ID:{}", correlationData.getId());
            }else{
                // nack, 消息失败
                log.error("消息发送确认失败, ID:{}, 原因{}",correlationData.getId(), result.getReason());
            }
        },
        ex -> log.error("消息发送异常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage())
    );
    // 发送消息
    rabbitTemplate.convertAndSend("amq.direct", "simple", message, correlationData);
}

消息持久化

MQ消息是存储在内存中的，开启持久化功能可以确保缓存在MQ中的消息不丢失

交换机持久化

@Bean
public DirectExchange simpleExchange() {
    // 三个参数：交换机名称，是否持久化，当没有queue与交换机绑定是否自动删除
    return new DirectExchange("simple.direct", true, false);
}

队列持久化

@Bean
public Queue simpleQueue() {
    // 使用QueueBuilder构建队列，durable意味着持久化
    return QueueBuilder.durable("simple.queue").build();
}

以上代码加入到config中，加@Configuration注解

消息持久化， SpringAMQP中的消息默认是持久的，可以通过MessageProperties中的DeliveryMode来指定

Message msg = MessageBuilder
    .withBody(message.getBytes(StandardCharsets.UTF_8)) // 消息体
    .setDeliveryMode(MessageDeliveryMode.PERSISTENT) // 持久化
    .build();

默认情况下，SpringBoot在上面三者都是默认持久化的

消费者消息确认

RabbitMQ支持消费者确认机制，即: 消费者处理消息后可以向MQ发送ack回执，MQ收到ack回执后才会删除该消息。

SpringAMQP则允许配置三种默认模式:

  • manual: 手动ack，需要在业务代码结束后，调用api发送ack
  • auto: 自动ack，由spring检测listener代码是否出现异常，没有异常则返回ack; 抛出异常则返回nack
  • none: 关闭ack, MQ假定消费者获取消息后会成功处理，因此消息投递后立即被删除
spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1
        acknowledge-mode: none # none, 不自动ack; manual, 手动ack; auto: 自动ack

失败重试机制

auto模式下消费者返回nack，MQ会重新投递给消费者。如果代码本身有问题，消费者无法处理消息，抛出异常后，消息会不断requeue(重新入队)到队列，再重新发送给消费者，然后再次异常，再次requeue，无限循环，导致mq的消息处理飙升，带来不必要的压力。

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

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

消费者失败消息处理策略

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

  • RejectAndDontRequeueRecoverer: 重试耗尽后，直接reject,丢弃消息。默认就是这种方式
  • ImmediateRequeueMessageRecoverer: 重试耗尽后，返回nack,消息重新入队
  • RepublishMessageRecoverer: 重试耗尽后，消费者将失败消息投递到指定的交换机
    • error.direct 和 error.queue 相当于error日志

首先，定义接收失败消息的交换机、队列及其绑定关系: 在Config中

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

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

@Bean
public Binding errorBinding() {
    return BindingBuilder.bind(errorQueue()).to(errorMessageExchange()).with("error");
}

然后，定义RepublishMessageRecoverer:

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

死信交换机

死信:

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

如果该队列配置了dead-letter-exchange属性，指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机称为死信交换机 (DeadLetterExchange,简称DLX)

dead-letter-exchange = dl.direct
dead-letter-routing-key = dl

消费者Reject后，队列将消息投递到指定的死信交换机

  • dl.direct 和 dl.queue

TTL

Time to live

如果一个队列中的消息TTL结束仍未消费，则会变为死信，ttl超时分为两种情况：

  • 消息所在的队列设置了存活时间
  • 消息本身设置了存活时间
  • 两者共存时，以时间短的ttl为准

死信交换机结合ttl实现延迟队列:

image-20240219222801269

声明死信交换机和队列 （在listener中)

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "dl.queue", durable = "true"),
    exchange = @Exchange(name = "dl.direct"),
    key = "dl"
))
public void listenDlQueue(String msg) {
    log.info("监听到 dl.queue 的消息内容: {}", msg);
}

设置超时时间 (在config中)

@Bean
public DirectExchange ttlExchange() {
    return new DirectExchange("ttl.direct");
}

@Bean
public Queue ttlQueue() {
    return QueueBuilder.durable("ttl.queue") // 持久化队列, 开始化
        .ttl(10000) // 设置队列消息的过期时间, 10秒
        .deadLetterExchange("dl.direct") // 设置死信交换机
        .deadLetterRoutingKey("dl") // 设置死信RoutingKey
        .build();
}

@Bean
public Binding simpleBinding() {
    return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("ttl");
}

延迟队列插件

利用TTL结合死信交换机，我们实现了消息发出后，消费者延迟收到消息的效果。这种消息模式就称为延迟队列 (Delay Queue)模式。

延迟队列的使用场景包括:

  • 延迟发送短信
  • 用户下单，如果用户在15分钟内未支付，则自动取消
  • 预约工作会议，20分钟后自动通知所有参会人员

DelayExchange插件

使用:

DelayExchange插件的原理是对官方原生的Exchange做了功能的升级:

  • 将DelayExchge接受到的消息暂存在内存中（官方的Exchange是无法存储消息的）
  • 在DelayExchange中计时，超时后才投递消息到队列中

image-20240219223705337

发送消息时在headers指定x-delay来指定具体的延迟时间

  1. 基于注解方式声明延迟队列: 设定delay属性为true即可

    @RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "delay.queue", durable = "true"),
    exchange = @Exchange(name = "delay.direct",delayed = "true"),
    key = "delay"
))
public void listenDlQueue(String msg) {
    log.info("监听到 delay.queue 的延迟消息: {}", msg);
}

  2. 基于Bean的方式

    @Bean
public DirectExchange delayedExchange() {
    return ExchangeBuilder
        .directExchange("delay.direct") // 延迟交换机延迟队列名称
        .delayed() // 设置delay属性 true
        .durable(true) // 持久化
        .build();
}

@Bean
public Queue delayedQueue() {
    return new Queue("delay.queue");
}

@Bean
public Binding delayedBinding() {
    return BindingBuilder.bind(delayedQueue()).to(delayedExchange()).with("delay");
}

发送消息时，添加一个header: x-delay

// 创建消息
Message message = MessageBuilder
    .withBody("hello, delayed message".getBytes(StandardCharsets.UTF_8))
    .setHeader("x-delay", 10000)
    .build();

// 消息ID, 需要持久化CorrelationData #
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

// 发送消息
rabbitTemplate.convertAndSend("delay.direct", "delay", message, correlationData);

log.debug("发送消息成功");

旧版MQ可能会有报错，因为理论上消息不应该在交换机中存储，延迟消息会让它误以为没有发送成功。

消息堆积和惰性队列

当生产者发送消息的速度过了消费者处理消息的速度，就会导致队列中的消息堆积，直到队列存储消息达到上限。最早接收到的消息，可能就会成为死信，会被丢弃，这就是消息堆积问题。

解决消息堆积有三种种思路：

  • 增加更多消费者，提高消费速度
  • 在消费者内开启线程池加快消息处理速度 // 线程池会增加cpu切换开销
  • 扩大队列容积，提高堆积上限

惰性队列的特征:

  • 接收到消息后直接存入磁盘而非内存
  • 消费者要消费消息时才会从磁盘中读取并加载到内存
  • 支持数百万条的消息存储

设置一个队列为惰性队列:

  • 声明队列时指定x-queue-mode属性为lazy

  • 将一个运行中的队列修改为惰性队列

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

用SpringAMQP声明惰性队列

  • 基于Bean的方式

  • @Bean
public Queue lazyQueue() {
    return QueueBuilder
        .durable("lazy.queue")
        .lazy() // 开启queue-mode为lazy
        .build();
}

  • 基于注解的方式

  • @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);
}

惰性队列的优点:

  • 基于磁盘存储，消息上限高
  • 没有间歇性的page-out，性能比较稳定

惰性队列的缺点:

  • 基于磁盘存储，消息时效性会降低

  • 性能受限于磁盘的IO

集群

RabbitMQ的是基于Erlang语言编写，而Erlang又是一个面向并发的语言，天然支持集群模式。

RabbitMQ的集群由两种模式

  • 普通集群: 是一种分布式集群，将队列分散到集群的各个节点，从而提高整个集群的并发能力。
  • 镜像集群: 是一种主从集群，普通集群的基础上，添加了主从备份功能，提高集群的数据可用性。

镜像集群虽然支持主从，但主从同步并不是强一致的，某些情况下可能有数据丢失的风险。因此在RabbitMQ的3.8版本以后，推出了新的功能：仲裁队列来代替镜像集群，底层采用Raft协议确保主从的数据一致性。

普通集群

也叫标准集群 classic cluster

特征:

  • 会在集群的各个节点间共享部分数据，包括：交换机、队列元信息。不包含队列中的消息

  • image-20240219230651550

  • 当访问集群某节点时，如果队列不在该节点，会从数据所在节点传递到当前节点并返回

  • image-20240219230729558

  • 队列所在的节点宕机，队列中的消息就会丢失

镜像集群

特征

  • 交换机、队列、队列中的消息会在各个mq的镜像节点之间同步备份。

  • 创建队列的节点被称为该队列的主节点，备份到的其它节点叫做该队列的镜像节点。

  • 一个队列的主节点可能是另一个队列的镜像节点

  • 所有操作都是主节点完成，然后同步给镜像节点

  • 主节点宕机后，镜像节点会替代成新的主节点

image-20240219231114245

仲裁队列

  • 与镜像队列一样，都是主从模式，支持主从数据同步
  • 使用非常简单，没有复杂的配置
  • 主从同步基于Raft协议，强一致

控制台添加仲裁队列:

image-20240219231354788

在SpringAMQP中创建仲裁队列

  • @Bean
public Queue quormQueue() {
    return QueueBuilder
        .durable("quorum.queue") // 持久化
        .quorum() //仲裁队列
        .build();
}

  • spring:
  rabbitmq:
    addresses: 192.168.150.105:8071, 192.168.150.105:8072, 192.168.150.105:8073 # 集群
    username: xxx
    password: xxx
    virtual-host: /