【RabbitMQ】消息可靠性投递（重点）

标签：可靠性 String 队列 RabbitMQ 投递 交换机 消息 import message

在使用消息队列时，消息丢失是一个常见的问题。了解导致消息丢失的原因后，有助于我们在实际开发中采取相应的措施来确保消息不丢失。以下是一些可能导致消息丢失的情况：

消息丢失的情况分析

如上图，正常的业务是，用户请求下单，后台服务器处理订单请求完成，发送订单消息到消息队列。更新购物车，更新库存，更新积分的其他服务接收到订单消息后，各自处理相关业务。

消息丢失情况一

用户在请求下单，后台处理完订单请求后，发送到消息队列时失败，失败的原因可能会有：网络异常或消息发送到一个还未建立的Exchange上，此时会导致消息丢失，后面一系列的操作都会失败。

消息丢失情况二

消息队列服务器宕机导致内存中消息丢失，或是消息已经发送到Exchange了，但是Exchange将消息根据Routing Key路由到对应的Queue时失败，例如这个Exchange根本就没有绑定Queue等等。

消息丢失情况三

消息成功存入消息队列，但是消费端出现问题，例如：宕机、抛异常等等 总之，在消息生产阶段->队列存储阶段->消息消费的整个过程中，每个环节都有可能出现异常情况，导致消息丢失

对症下药

所以，在了解消息丢失的情况分析后，我们就可以针对出现异常的各个环节进行修正处理，尽可能对症下药，保障消息能准确投递。

生产者到消息队列数据丢失

生产者弄丢了数据，生产者没有将消息数据发送到消息队列；

解决思路：

1、在生产端发送消息后，进行消息送达确认，分别针对交换机和队列来确认，如果没有成功发送到消息队列服务器上，那就可以尝试重新发送。

修改yml配置：

spring:
  rabbitmq:
    host: 10.0.70.101
    port: 5672
    username: guest
    password: 123456
    virtual-host: /
    publisher-confirm-type: CORRELATED # 交换机的确认
    publisher-returns: true # 队列的确认

在生产端进行配置，声明回调ConfirmCallback确认消息是否发送到交换机，声明ReturnsCallback 确认消息是否发送到队列

package com.cz.test.mq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

@Component
@Slf4j
public class MQProducerAckConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);
    }

    /**
     * 消息发送到交换机回调
     * @param correlationData 与发送消息时的CorrelationData关联
     * @param ack true：消息成功到达交换机，false：消息未到达交换机
     * @param cause 如果消息未到达交换机，此参数才有值，返回失败原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            log.info("消息发送到交换机成功！数据：" + correlationData);
        } else {
            log.info("消息发送到交换机失败！数据：" + correlationData + " 原因：" + cause);
        }
    }

    /**
     * 消息发送到队列失败回调
     * @param returned ReturnedMessage Message 消息主题；routingKey 路由键；exchange 交换机；replyCode 响应码；replyText 描述；
     */
    @Override
    public void returnedMessage(ReturnedMessage returned) {
        log.info("消息主体: " + new String(returned.getMessage().getBody()));
        log.info("应答码: " + returned.getReplyCode());
        log.info("描述：" + returned.getReplyText());
        log.info("消息使用的交换器 exchange : " + returned.getExchange());
        log.info("消息使用的路由键 routing : " + returned.getRoutingKey());
    }
}

2、为目标交换机指定备份交换机，当目标交换机投递失败时，把消息投递至备份交换机。

备份交换机主要是处理“发布者 ===》交换机”这个过程，保存没有被路由成功的消息。它相当于是交换机的备胎，专门用来应对普通交换机不能路由成功的消息。当我们为一个交换机声明一个对应的备份交换机的时候，就是给它创建了一个备胎。一旦交换机收到了一条无法路由的消息是，就会把这条消息转发给备份交换机，由备份交换机去进行转发处理。

通常情况下，备份交换机都是fanout类型的，这样可以方便将所有的消息都投递到与其绑定的队列当中，然后我们在这个队列下边去进行信息的处理，甚至还可以去创建一个报警队列，用独立的消费者来专门监测和报警，省掉每次都通过日志去查看消息情况！

引用一个架构图

定义备份交换机的配置,将确认交换机与备份交换机管理：.withArgument("alternate-exchange", BACKUP_EXCHANGE)

package com.cz.test.mq;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class PublisherConfirmConfig {

    private static final String CONFIRM_EXCHANGE = "confirm.exchange";
    private static final String CONFIRM_QUEUE = "confirm.queue";
    private static final String BACKUP_EXCHANGE = "backup.exchange";
    private static final String BACKUP_QUEUE = "backup.queue";
    private static final String WARNING_QUEUE = "warning.queue";

    /**
     * 管理确认交换机，需要关联上备份交换机
     */
    @Bean(CONFIRM_EXCHANGE)
    public DirectExchange confirmExchange() {
        //关联上备份交换机
        return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE).durable(true)
                .withArgument("alternate-exchange", BACKUP_EXCHANGE).build();
    }

    /**
     * 管理备份交换机
     */
    @Bean(BACKUP_EXCHANGE)
    public FanoutExchange backupExchange() {
        return new FanoutExchange(BACKUP_EXCHANGE);
    }

    /**
     * 管理确认队列
     */
    @Bean(CONFIRM_QUEUE)
    public Queue confirmQueue() {
        return QueueBuilder.durable(CONFIRM_QUEUE).build();
    }

    /**
     * 管理备份队列
     */
    @Bean(BACKUP_QUEUE)
    public Queue backupQueue() {
        return QueueBuilder.durable(BACKUP_QUEUE).build();
    }

    /**
     * 管理告警队列
     */
    @Bean(WARNING_QUEUE)
    public Queue warningQueue() {
        return QueueBuilder.durable(WARNING_QUEUE).build();
    }

    /**
     * 绑定确认交换机和确认队列
     *
     * @param queue    确认队列
     * @param exchange 确认交换机
     */
    @Bean
    public Binding confirmQueueBindingConfirmExchange(
            @Qualifier(CONFIRM_QUEUE) Queue queue,
            @Qualifier(CONFIRM_EXCHANGE) Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("businessKey").noargs();
    }

    /**
     * 绑定备份交换机和备份队列
     *
     * @param queue    备份队列
     * @param exchange 备份交换机
     */
    @Bean
    public Binding backupQueueBindingBackupExchange(
            @Qualifier(BACKUP_QUEUE) Queue queue,
            @Qualifier(BACKUP_EXCHANGE) Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("").noargs();
    }

    /**
     * 绑定备份交换机和告警队列
     *
     * @param queue    告警队列
     * @param exchange 备份交换机
     */
    @Bean
    public Binding warningQueueBindingBackupExchange(@Qualifier(WARNING_QUEUE) Queue queue,
                                                     @Qualifier(BACKUP_EXCHANGE) Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("").noargs();
    }
}

定义消息发送者：

package com.cz.test.mq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.Date;

@RestController
@RequestMapping("/publish")
@Slf4j
public class Publisher {
    private static final String CONFIRM_EXCHANGE = "confirm.exchange";
    
    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 发布正常业务消息
     */
    @GetMapping("/{message}")
    public String sendMessage(@PathVariable("message")String message){
        String date = new Date().toString();
        log.info("生产者在：{}，发布了消息：{}",date,message);
        //发送一条路由key正确、id为1的消息
        rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE,"businessKey",message,new CorrelationData("1"));
        return "生产者在："+date+",发布了一条消息："+message;
    }
 
    /**
     * 发布消息:不可路由的消息
     */
    @GetMapping("/error/{message}")
    public String sendMessage2(@PathVariable("message")String message){
        String date = new Date().toString();
        log.info("生产者在：{}，发布了消息：{}",date,message);
        //发送一条路由key不正确，id为2的消息
        rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE,"key2",message,new CorrelationData("2"));
        return "生产者在："+date+",发布了一条消息："+message;
    }
}

监听消费情况：

package com.cz.test.mq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Date;

@Component
@Slf4j
public class Comsumer {
    private static final String CONFIRM_QUEUE = "confirm.queue";
    private static final String BACKUP_QUEUE = "backup.queue";
    private static final String WARNING_QUEUE = "warning.queue";
    /**
     * 监听确认队列当中的消息
     */
    @RabbitListener(queues = CONFIRM_QUEUE)
    public void confirmMessage(Message message){
        String date = new Date().toString();
        log.info("消费者C1在：{}，收到了确认队列当中的消息：{}",date,new String(message.getBody()));
    }


    /**
     * 监听备份队列当中的消息
     */
    @RabbitListener(queues = BACKUP_QUEUE)
    public void backupMessage(Message message){
        String date = new Date().toString();
        log.info("备用消费者在：{}，收到了消息：{}",date,new String(message.getBody()));
    }

    /**
     * 监听警告队列当中的消息
     */
    @RabbitListener(queues = WARNING_QUEUE)
    public void warningMessage(Message message){
        String date = new Date().toString();
        log.info("告警在：{},发现了消息：{}",date,new String(message.getBody()));
    }
}

在分别发送一条正常消息，和一条异常消息后，可以看到，当消息无法被路由到正确的queue时，使用“备份交换机”机制之后，confirm交换机不再把消息直接回退、通知发布者，而是将消息转发给备份交换机，备份消费者、告警消费者从与备份交换机绑定的队列来消费消息；

生产者在：Fri Oct 11 11:17:08 CST 2024，发布了消息：normal message
消息发送到交换机成功！数据：CorrelationData [id=1]
消费者C1在：Fri Oct 11 11:17:08 CST 2024，收到了确认队列当中的消息：normal message

生产者在：Fri Oct 11 11:17:10 CST 2024，发布了消息：error message
备用消费者在：Fri Oct 11 11:17:10 CST 2024，收到了消息：error message
消息发送到交换机成功！数据：CorrelationData [id=2]
告警在：Fri Oct 11 11:17:10 CST 2024,发现了消息：error message

CorrelationData 参数在使用 RabbitTemplate 发送消息时用于关联发送的消息和其确认结果。具体作用包括：

• 消息追踪：通过设置唯一的 CorrelationData 值（通常是消息ID，唯一），可以在消息发送后跟踪该消息的状态。
• 发布确认：当消息被投递到一个或多个队列后，RabbitMQ 会发送一个确认回调给应用。CorrelationData 在这个过程中用来标识哪个确认回调对应哪条消息。
• 错误处理：如果消息未能正确到达队列，可以通过 CorrelationData 进行错误处理和日志记录

消息未持久化

消息队列服务器宕机导致内存中消息丢失；

解决思路：消息持久化到硬盘上，哪怕服务器重启也不会导致消息丢失。

我们可以通过将durable的值设置为true来保证持久化。要想做到消息持久化，必须满足以下三个条件，缺一不可。

    Exchange 设置持久化，将durable设置为true

    Queue 设置持久化，durable参数设置，默认为true

参数说明：

    queue：queue 的名称
    exclusive：排他队列，如果一个队列被声明为排他队列，该队列仅对首次申明它的连接可见，并在连接断开时自动删除。这里需要注意三点：
        排他队列是基于连接可见的，同一连接的不同信道是可以同时访问同一连接创建的排他队列；
        “首次”，如果一个连接已经声明了一个排他队列，其他连接是不允许建立同名的排他队列的，这个与普通队列不同；
        即使该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会被自动删除的，这种队列适用于一个客户端发送读取消息的应用场景。
    autoDelete：自动删除，如果该队列没有任何订阅的消费者的话，该队列会被自动删除。这种队列适用于临时队列。

    Message持久化发送：发送消息时默认模式deliveryMode=2，代表持久化消息

消费者异常

消费端异常导致消息没有成功被消费；

解决思路：消费端消费消息成功，给服务器返回ACK信息；

修改yml配置：

spring:
  rabbitmq:
    host: 10.0.70.101
    port: 5672
    username: guest
    password: 123456
    virtual-host: /
    publisher-confirm-type: CORRELATED # 交换机的确认
    publisher-returns: true # 队列的确认
    listener:
      simple:
        acknowledge-mode: manual # 把消息确认模式改为手动确认

改造下上面正常消费的监听器：

 /**
     * 监听确认队列当中的消息
     */
    @RabbitListener(queues = CONFIRM_QUEUE)
    public void confirmMessage(String dataMsg,Message message, Channel channel) throws IOException{
        // 1、获取当前消息的 deliveryTag 值备用
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            // 2、正常业务操作
            log.info("消费端接收到消息内容：" + dataMsg);
             System.out.println(10 / 0);
            // 3、给 RabbitMQ 服务器返回 ACK 确认信息
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {

            // 4、获取信息，看当前消息是否曾经被投递过
            Boolean redelivered = message.getMessageProperties().getRedelivered();

            if (!redelivered) {
                // 5、如果没有被投递过，那就重新放回队列，重新投递，再试一次
                channel.basicNack(deliveryTag, false, true);
            } else {
                // 6、如果已经被投递过，且这一次仍然进入了 catch 块，那么返回拒绝且不再放回队列
                channel.basicReject(deliveryTag, false);
            }
        }
    }

当消息正常到达消费监听后，如果业务处理正常，直接给 RabbitMQ 服务器返回 ACK 确认信息（代码3），第一个参数deliveryTag，消息的唯一标识，第二个参数multiple，消息是否支持批量确认，如果是true，代表可以一次性确认标识小于等于当前标识的所有消息，如果是false，只会确认当前消息

当消息监听器接收消息后，此时业务逻辑发生了异常（如上：System.out.println(10 / 0)），可以判断消息是否曾经被投递过，如果没有被投递过，那就重新放回队列，重新投递，再试一次（代码4、5）。basicNack(deliveryTag, false, true)，第一个参数消息唯一标识，第二个是否支持批量确认，第三个表示是否重新入列

当消息再次被消费到，此时还是业务异常，这条消息显然已被投递过一次，将执行basicReject(deliveryTag, false)（代码6），此时可以拒绝再次入列（第二个参数，为true时，消息会再次入列），以免形成循环入列，造成消息积压。

标签：可靠性,String,队列,RabbitMQ,投递,交换机,消息,import,message
From： https://blog.csdn.net/chenzhong373095238/article/details/142826142