消息可靠性投递

Posted 2022-08-04 skystep

confirm 确认消息

消息确认机制是消息的可靠性投递的核心保障：

• 消息的确认是指生产者投递消息后，如果 Broker 收到消息，则会给生产者一个应答；
• 生产者接收应答用来确认这条消息是否正常发送到 Broker。

confirm 确认消息实现

添加依赖

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

开启确认消息机制

# 设置是否到达交换机
spring.rabbitmq.publisher-confirm-type=correlated

• NONE 值是禁用发布确认模式，是默认值;
• CORRELATED 值是发布消息成功交换机后触发回调方法;
• SIMPLE 值有两种效果，其一效果和 CORRELATED 值一样会触发回调方法，其二在发布消息成功后使用 rabbitTemplate 调用 waitForConfirms 或 waitForConfirmsOrDie 方法等待 broker 节点返回发送结果。

设置 ConfirmCallback 回调通知

通过判断 ack 的值从而知道是否正常投递，cause 是对投递结果的说明，correlationData 是消息的时候设置的相关信息。

package com.skystep.rabbitmq.config.rabbitmq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.nio.charset.StandardCharsets;
import java.util.Objects;

@Configuration
@Slf4j
public class RabbitmqConfig 
    @Bean
    public RabbitTemplate creatRabbitTemplate(ConnectionFactory connectionFactory) 
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        //设置消息投递失败的策略，有两种策略：自动删除或返回到客户端。
        //我们既然要做可靠性，当然是设置为返回到客户端(true是返回客户端，false是自动删除)
        rabbitTemplate.setConnectionFactory(connectionFactory);
        rabbitTemplate.setMandatory(true);

        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> 
            log.info("ConfirmCallback 相关数据：", correlationData.getId());
            if (ack) 
                log.info("ConfirmCallback 投递成功,确认情况：", ack);
                // 在这里可以标记消息发送成功等更多处理
             else 
                log.info("ConfirmCallback 投递失败,确认情况：, 原因：", ack, cause);
                // 获取没有确认的消息
                Message returnedMessage = correlationData.getReturnedMessage();
                // 如果没有消息说明客户端发送消息的时候CorrelationData中没有设置ReturnedMessage
                if (Objects.isNull(returnedMessage)) 
                    log.warn("ConfirmCallback 获取不到退回的消息：", correlationData.getId());
                    return;
                
                byte[] body = returnedMessage.getBody();
                if (Objects.nonNull(body)) 
                    String msg = new String(body, StandardCharsets.UTF_8);
                    log.info("ConfirmCallback 退回的消息：", msg);
                    // 在这里可以标记消息发送失败等更多处理
                 else 
                    log.warn("ConfirmCallback 获取不到退回的消息：", correlationData.getId());
                
            
        );

        return rabbitTemplate;
    

编写消息发送者

此处模拟往错误的交换机发送消息，broker 无法找到对应的交换机发送消息。

@GetMapping("/message/sender/confirm")
public String messageConfirm() 
    String msgId = UUID.randomUUID().toString();
    String msg = "confirm";
    CorrelationData correlationData = new CorrelationData(msgId);
    MessageProperties messageProperties = new MessageProperties();
    messageProperties.setMessageId(msgId);
    messageProperties.setCorrelationId(msgId);
    Message message = new Message(msg.getBytes(StandardCharsets.UTF_8), messageProperties);
    correlationData.setReturnedMessage(message);
    // 设置错误的交换机
    rabbitTemplate.convertAndSend("testDirectExchangeError", "testDirectRouting", message, correlationData);
    return "ok";

调用接口验证结果

2022-07-24 14:08:57.611  INFO 83652 --- [nectionFactory2] c.s.r.config.rabbitmq.RabbitmqConfig     : ConfirmCallback 相关数据：6b9b495c-afa8-4b2d-9ba4-09d77ed91d26
2022-07-24 14:08:57.612  INFO 83652 --- [nectionFactory2] c.s.r.config.rabbitmq.RabbitmqConfig     : ConfirmCallback 投递失败,确认情况：false, 原因：channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'testDirectExchangeError' in vhost '/', class-id=60, method-id=40)
2022-07-24 14:08:57.612  INFO 83652 --- [nectionFactory2] c.s.r.config.rabbitmq.RabbitmqConfig     : ConfirmCallback 退回的消息：confirm

return 回退消息

return 消息回退机制用于处理一些不可以路由的消息：

生产者通过交换机和路由键指定发送到对应的队列，然后消费者进行监听和处理。但是某些情况下，交换机和或者路由键错误，致使消息无法到达指定的队列。如果开启了消息回退监听，就可以监听这种不可到达的消息进行对应的处理。

添加依赖

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

开启回退消息机制

# 设置发送失败是否回退
spring.rabbitmq.publisher-returns=true

设置 ReturnCallback 回调通知

@Configuration
@Slf4j
public class RabbitmqConfig 
    @Bean
    public RabbitTemplate creatRabbitTemplate(ConnectionFactory connectionFactory) 
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        //设置消息投递失败的策略，有两种策略：自动删除或返回到客户端。
        //我们既然要做可靠性，当然是设置为返回到客户端(true是返回客户端，false是自动删除)
        rabbitTemplate.setConnectionFactory(connectionFactory);
        rabbitTemplate.setMandatory(true);

        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> 
            MessageProperties messageProperties = message.getMessageProperties();
            String correlationId = messageProperties.getCorrelationId();
            String body = new String(message.getBody(), StandardCharsets.UTF_8);
            log.info("ReturnCallback 消息：,消息id；", body, correlationId);
            log.info("ReturnCallback 回应码：, 回应信息：, 交换机：, 路由键：", replyCode, replyText, exchange, routingKey);
            // 在这里可以标记消息发送失败等更多处理
        );

        return rabbitTemplate;
    

编写消息发送者

此处模拟往错误的路由键发送消息，broker 无法路由到对应的队列。

@GetMapping("/message/sender/return")
public String messageReturn() 
    String msg = "return";
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    MessageProperties messageProperties = new MessageProperties();
    Message message = new Message(msg.getBytes(StandardCharsets.UTF_8), messageProperties);
    correlationData.setReturnedMessage(message);
    rabbitTemplate.convertAndSend("testDirectExchange", "testDirectRoutingError", msg, correlationData);
    return "ok";

调用接口验证结果

2022-07-24 20:53:36.876  INFO 80356 --- [nectionFactory3] c.s.r.config.rabbitmq.RabbitmqConfig     : ReturnCallback 消息：return,消息id；a9f2a580-ca3b-44ea-8b62-07a76ac2bfd0
2022-07-24 20:53:36.877  INFO 80356 --- [nectionFactory3] c.s.r.config.rabbitmq.RabbitmqConfig     : ReturnCallback 回应码：312, 回应信息：NO_ROUTE, 交换机：testDirectExchange, 路由键：testDirectRoutingError

可靠性投递方案

生产端消息可靠性传递是为了解决消息投递时丢失的问题，导致这种问题的原比较多，比如rabbitMQ 宕机，投递过程中交换机发生问题，路由键错误等。

消息落库方案

方案介绍

如图，该方案对消息进行落库处理，每次操作之后修改消息的状态。启动定时任务对消息库中的获取投递失败的消息，请求发送者重新投递消息。数据库中核心存储消息的唯一id，消息状态，重试次数，重试时间，主要的步骤如下：

• step1: 发送消息前，存储消息至消息库；

• step2: 发送消息至 MQ Broker；

• step3: 生产者监听到MQ的确认消息；

• step4: 如果 ack = true，更改数据库中消息的状态；

• step5: 定时任务（定时去获取数据库中消息状态为投递中状态，并且重试时间小于当前时间的消息）

• step6:如果重试次数不超过3次，消息的重试次数+1，更新重试时间为当前时间+指定超时时间

• step7:如果重试次数超过3次，更新状态为投递失败，不能重试

方案缺点

• 消息持久化数据库且涉及多从更新消息状态，大并发场景下，数据库可能成为瓶颈；
• 引进了定时任务，在分布式架构下，需要支持分布式定时任务。

在需要保证消息投递 100% 可靠，消息并发不大的场景下，推荐使用。