MQ高级-服务异步通信

消息队列在使用过程中，面临的实际问题：

• 消息可靠性（如何确保发送的消息至少被消费一次）
• 延迟消息问题（如何实现消息的延迟投递）
• 消息堆积问题（如何解决百万消息堆积，无法及时消费的问题）
• 高可用问题（如何避免单点的MQ故障而导致的不可用问题）

相关面试题

• Rabbitmq如何保证mq消息可靠性？（3大方面）
  1、如何保证生产者不丢（confirm、ack）
  2、如何保证消息到达MQ了之后不丢（持久化）
  3、如何保证消费者不丢（自行判断配置什么ack返回模式）
• Rabbitmq如何实现延时消息？（2种）（队列、消息）
• 什么是死信队列？什么样的消息会进入死信队列？（消息无法被消费）
• Rabbitmq如何解决消息堆积问题？（3种思路）（增加消费者or在消费者内开启线程池加快消息处理速度、扩大队列容积）

1、消息可靠性

1.1.生产者消息确认

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

返回结果有两种方式：

• publisher-confirm，发送者确认
  • 消息成功投递到交换机，返回ack
  • 消息未投递到交换机，返回nack
• publisher-return，发送者回执
  • 消息投递到交换机了，但是没有路由到队列。返回ACK，及路由失败原因。
• 多个ack避免冲突，确认机制发送消息时，会给每个消息设置一个全局唯一id，以区分不同消息

修改配置

修改生产者服务中的配置文件：

spring:
  rabbitmq:
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true
1
2
3
4
5
6

• publisher-comfirm-type：开启publisher-comfirm，支持两种类型：
  • simple：同步等待confirm结果，直至超时
  • correlated：异步回调，定义ComfirmCallback，MQ返回结果时会回调这个ComfirmCallback
• publisher-returns:开启publisher-return功能，同样基于Callback机制，定义ReturnCallback
• template.mandatory：定义消息路由失败时的策略。
  • true：调用ReturnCallback
  • false：直接丢弃消息

定义Return回调
每个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());
            // 如果有业务需要，可以重发消息
        });
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

定义ConfirmCallback
ComfirmCallback可以在发送消息时指定，因为每个业务处理成功或失败的逻辑不确定。

public void testSendMessage2SimpleQueue() throws InterruptedException {
    // 1.消息体
    String message = "hello, spring amqp!";
    // 2.全局唯一的消息ID，需要封装到CorrelationData中
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    // 3.添加callback
    correlationData.getFuture().addCallback(
        result -> {
            if(result.isAck()){
                // 3.1.ack，消息成功
                log.debug("消息发送成功, ID:{}", correlationData.getId());
            }else{
                // 3.2.nack，消息失败
                log.error("消息发送失败, ID:{}, 原因{}",correlationData.getId(), result.getReason());
            }
        },
        ex -> log.error("消息发送异常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage())
    );
    // 4.发送消息
    rabbitTemplate.convertAndSend("task.direct", "task", message, correlationData);

    // 休眠一会儿，等待ack回执
    Thread.sleep(2000);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

1.2.消息持久化

生产者确认可以确保消息投递到RabbitMQ的队列中，但是消息发送到RabbitMQ以后，如果突然宕机，也可能导致消息丢失。

要想确保消息在RabbitMQ中安全保存，必须开启消息持久化机制（durable方法）。

RabbitMQ中交换机默认是非持久化的，mq重启后就丢失，但是SpringAMQP中通过代码去定义的交换机、队列和消息都是持久化的。

• 交换机持久化
• 队列持久化
• 消息持久化

用普通的方式去定义交换机、队列和消息，系统也是自动默认持久化的。

1.3.消费者消息确认

RabbitMQ是阅后即焚机制，RabbitMQ确认消息被消费者消费后会立刻删除。（默认）

而RabbitMQ是通过消费者回执来确认消费者是否成功处理消息的：消费者获取消息后，应该向RabbitMQ发送ACK回执，表明自己已经处理消息。

如果出现场景：

• 1）RabbitMQ投递消息给消费者
• 2）消费者获取消息后，返回ACK给RabbitMQ
• 3）RabbitMQ删除消息
• 4）消费者宕机，消息尚未处理

这样，消息就丢失了。因此消费者返回ACK的时机非常重要。

SpringAMQP则允许配置三种确认模式：

•manual：手动ack，需要在业务代码结束后，调用api发送ack。

•auto：自动ack，由spring监测listener代码是否出现异常，没有异常则返回ack；抛出异常则返回nack

•none：关闭ack，MQ假定消费者获取消息后会成功处理，因此消息投递后立即被删除

因此，一般在配置文件中设置消息确认的方式：

• none模式下，消息投递是不可靠的，可能丢失
• auto模式类似事务机制，出现异常时返回nack，消息回滚到mq；没有异常，返回ack（MQ压力会很大，消费端代码会无限执行，不断在消费端和队列之间跳转）
• manual：自己根据业务情况，判断什么时候该ack

一般，我们都是使用默认的auto即可。

1.4.消费失败重试机制

当消费者消息确认模式设置为auto的时候，出现异常时，消息会不断从消费者端到队列再到消费者端，给mq带来不必要的压力。
解决：
本地重试
利用Spring的retry机制，在消费者出现异常时利用本地重试，然后再进行其他处理。

• 修改消费者配置：

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

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

失败策略
在之前的测试中，达到最大重试次数后，消息会被丢弃，这是由Spring内部机制决定的。

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

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

比较优雅的一种处理方案是RepublishMessageRecoverer，失败后将消息投递到一个指定的，专门存放异常消息的队列，后续由人工集中处理。

@Configuration
public class ErrorMessageConfig {
//定义指定交换机
    @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");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

面试题1
如何确保RabbitMQ消息的可靠性？

• 开启生产者确认机制，确保生产者的消息能到达队列
• 开启持久化功能，确保消息未消费前在队列中不会丢失
• 开启消费者确认机制为auto，由spring确认消息处理成功后完成ack
• 开启消费者失败重试机制，并设置MessageRecoverer，多次重试失败后将消息投递到异常交换机，交由人工处理

2、死信交换机

2.1.初始死信交换机

死信
当一个队列中满足下列情况之一时，可以成为死信（dead letter）：

• 消费者使用basic.reject（默认）或者basic.nack声明消费失败
• 消息超时未消费
• 要投递的队列消息满了，踢出最早前没有运行的代码。

死信交换机
如果这个包含死信的队列配置了dead-letter-exchange属性，指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机称为死信交换机（Dead Letter Exchange，检查DLX）。

队列将死信投递给死信交换机时，必须知道以下两点：

• 死信交换机名称
• 死信交换机与死信队列绑定的RoutingKey

利用死信交换机接收死信
在失败重试策略中，默认的RejectAndDontRequeueRecoverer会在本地重试次数耗尽后，发送reject给RabbitMQ，消息变成死信，被丢弃。
我们给该队列添加一个死信交换机，给死信交换机绑定一个队列。这样消息变成死信后也不会被丢弃，而最终投递到死信交换机，路由到与死信交换机绑定的队列。

// 声明普通的 simple.queue队列，并且为其指定死信交换机：dl.direct
@Bean
public Queue simpleQueue2(){
    return QueueBuilder.durable("simple.queue") // 指定队列名称，并持久化
        .deadLetterExchange("dl.direct") // 指定死信交换机
        .build();
}
// 声明死信交换机 dl.direct
@Bean
public DirectExchange dlExchange(){
    return new DirectExchange("dl.direct", true, false);
}
// 声明存储死信的队列 dl.queue
@Bean
public Queue dlQueue(){
    return new Queue("dl.queue", true);
}
// 将死信队列 与 死信交换机绑定
@Bean
public Binding dlBinding(){
    return BindingBuilder.bind(dlQueue()).to(dlExchange()).with("simple");
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

死信交换机的使用场景

• 如果队列中绑定了死信交换机，那么当队列中生成死信时，将其投递到死信交换机
• 利用死信交换机将生成的死信做人工处理，进一步提高消息队列的可靠性

2.2.TTL

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

• 消息所在的队列设置了超时时间
• 消息本身设置了超时时间

接收超时死信的死信交换机

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "dl.ttl.queue", durable = "true"),
    exchange = @Exchange(name = "dl.ttl.direct"),
    key = "ttl"
))
public void listenDlQueue(String msg){
    log.info("接收到 dl.ttl.queue的延迟消息：{}", msg);
}
1
2
3
4
5
6
7
8

声明一个队列，并且指定TTL

//给队列设置超时时间
@Bean
public Queue ttlQueue(){
    return QueueBuilder.durable("ttl.queue") // 指定队列名称，并持久化
        .ttl(10000) // 设置队列的超时时间，10秒
        .deadLetterExchange("dl.ttl.direct") // 指定死信交换机
        .build();
}
//声明交换机
@Bean
public DirectExchange ttlExchange(){
    return new DirectExchange("ttl.direct");
}
//将ttl与交换机绑定
@Bean
public Binding ttlBinding(){
    return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("ttl");
}

//发送消息，但是不要指定TTL
@Test
public void testTTLQueue() {
    // 创建消息
    String message = "hello, ttl queue";
    // 消息ID，需要封装到CorrelationData中
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    // 发送消息
    rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);
    // 记录日志
    log.debug("发送消息成功");
}
//发送消息时，指定TTL
@Test
public void testTTLMsg() {
    // 创建消息
    Message message = MessageBuilder
        .withBody("hello, ttl message".getBytes(StandardCharsets.UTF_8))
        .setExpiration("5000")
        .build();
    // 消息ID，需要封装到CorrelationData中
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    // 发送消息
    rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);
    log.debug("发送消息成功");
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

2.3.延迟队列

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

使用场景

• 延迟发送短信
• 用户下单15分钟未付款自动取消订单
• 预约会议开放会议室

支持延迟队列的插件

• 这个插件就是DelayExchange插件。参考RabbitMQ的插件列表页面

• 使用方式可以参考官网地址

DelayExchange原理
DelayExchange需要将一个交换机声明为delayed类型。当我们发送到delayExchange时，流程：

• 接收消息
• 判断消息是否具备x-delay属性
• 如果有x-delay属性，说明是延迟消息，持久化到硬盘，读取x-delay值，作为延迟时间
• 返回routing not found结果给消息发送者
• x-delay时间到期后，重新投递消息到指定队列

使用DelayExchange

• 声明一个交换机，交换机的类型可以是任意类型
• 只需要设定delayed属性为true
• 声明队列与其绑定
• 基于注解或者bean都行
• 发送消息时，一定要携带x-delay属性头，指定延迟的时间

3、惰性队列

3.1.消息堆积问题

定义
当生产者发送消息的速度超过了消费者处理消息的速度，就会导致队列中的消息堆积，知道队列存储消息达到上限。之后发送的消息就会成为死信，可能会被丢弃。

解决思路

• 增加更多消费者，提高消费速度。也就是我们之前说的work queue模式
• 扩大队列容积，提高堆积上限（提升队列容器吧消息保存在内存中显然是不行的）

3.2.惰性队列

RabbitMQ3.6.0版本开始，增加了Lazy Queues的概念，也就是惰性队列。
特征

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

设置lazy-queue

• 基于命令行设置

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

• rabbitmqctl ：RabbitMQ的命令行工具
• set_policy ：添加一个策略
• Lazy ：策略名称，可以自定义
• “^lazy-queue$” ：用正则表达式匹配队列的名字
• ‘{“queue-mode”:“lazy”}’ ：设置队列模式为lazy模式
• –apply-to queues：策略的作用对象，是所有的队列

• 基于bean声明lazy-queue
  • 指定x-queue-mode属性为lazy即可
• 基于RabbitListener声明LazyQueue
  • 在声明交换机的注解上，添加

arguments = @Argument(name = "x-queue-mode",value = "lazy"
1

惰性队列的优点

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

惰性队列的缺点

• 基于磁盘存储，消息时效性会降低
• 性能受限于磁盘的IO

4、MQ集群

另起一篇。