当前位置：首页 > article >正文

RabbitMQ 消息确认机制

article 2024/12/2 19:22:24

RabbitMQ 消息确认机制

本文总结了RabbitMQ消息发送过程中的一些代码片段，详细分析了回调函数和发布确认机制的实现，以提高消息传递的可靠性。

返回回调机制的代码分析

主要用途

这个代码主要用于设置RabbitMQ消息发送过程中的回调函数，即在消息不能被成功投递到目标队列时的返回回调处理。具体来说，它使用RabbitTemplate对象来设置ReturnsCallback回调函数，这个回调函数可以帮助开发者在消息无法正确路由到队列时获得相关的反馈信息。

代码详细分析

类与注解

@Slf4j：用来生成一个用于日志记录的log对象，方便日志输出。
@AllArgsConstructor：自动生成全参构造器，使得可以方便地初始化类的所有字段。
@Configuration：表明这是一个Spring配置类，用于定义一些Bean。

RabbitTemplate的初始化

private final RabbitTemplate rabbitTemplate;：这是一个RabbitTemplate对象，用于与RabbitMQ进行交互操作。
@PostConstruct：标注的方法将在所有依赖项注入完成后自动调用，通常用于初始化逻辑。

`init()` 方法

rabbitTemplate.setReturnsCallback(...)：设置一个回调函数，当消息不能正确路由时会触发这个回调。
new RabbitTemplate.ReturnsCallback()：匿名内部类，用于定义ReturnsCallback的逻辑。
public void returnedMessage(ReturnedMessage returned)：这是实现ReturnsCallback接口中的方法，它负责处理返回的消息。

`returnedMessage()` 方法中的具体逻辑

log.error("触发return callback,")：表示触发了返回回调，记录一条错误日志。
log.debug(...)：输出了消息在返回时的各种属性，包括：
- exchange：消息发送到的交换机名称。
- routingKey：用于路由消息的路由键。
- message：具体的消息内容。
- replyCode 和 replyText：返回码及返回的文本，表明返回的原因。

方法执行的时机

当RabbitMQ消息没有成功被路由到指定的队列时，这个回调函数将被调用。典型的场景是队列不存在、交换机配置错误或路由键不匹配等情况。
这种情况下，RabbitMQ会触发返回回调，通过returnedMessage()方法通知应用程序该消息未能被成功投递。

代码示例

@Slf4j
@AllArgsConstructor
@Configuration
public class MqConfig {
    private final RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init() {
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                log.error("触发return callback,");
                log.debug("exchange: {}", returned.getExchange());
                log.debug("routingKey: {}", returned.getRoutingKey());
                log.debug("message: {}", returned.getMessage());
                log.debug("replyCode: {}", returned.getReplyCode());
                log.debug("replyText: {}", returned.getReplyText());
            }
        });
    }
}

代码的用途

当RabbitMQ消息没有成功被路由到指定的队列时，这个回调函数将被调用。
典型的场景是队列不存在、交换机配置错误或路由键不匹配等情况。
通过这些日志输出，开发人员可以方便地定位问题所在，了解为什么消息未能成功发送到队列中。
这个回调可以提高系统的健壮性，确保消息发送的可靠性，如果消息丢失或出现其他问题，可以通过日志及时发现并进行处理。

总结：这个代码主要用于设置RabbitMQ的返回回调机制，以便在消息发送失败时可以及时进行处理和日志记录。它是消息可靠传输的一部分，尤其适用于需要高可靠性的消息传递场景。

发布确认机制的代码分析

主要用途

这个代码演示了如何在RabbitMQ中使用**发布确认（Publisher Confirm）**的机制，确保消息成功到达交换机并获得确认（ack）。通过这种机制，应用程序可以在消息发送到交换机后获得反馈，确认是否发送成功，从而提高消息传输的可靠性。

代码详细分析

方法注解与声明

@Test：表示这是一个测试方法，通常用JUnit等测试框架来执行，测试消息发送的过程。
void testPublisherConfirm()：方法名表明该方法用于测试发布者的确认（Publisher Confirm）。

创建`CorrelationData`对象

CorrelationData cd = new CorrelationData();
- CorrelationData对象用于追踪每个消息的唯一标识。
- 在发布消息时，相关的确认信息与该对象进行关联。
- 通过该对象，可以区分出每条消息的确认状态。

设置`ConfirmCallback`回调

cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {...})
- 使用了一个ListenableFutureCallback来监听消息的确认结果。
- addCallback()方法用于指定回调逻辑，当消息被确认（或失败）时，执行对应的逻辑。
onFailure(Throwable ex)
- 如果消息确认过程中发生异常，则会进入onFailure()方法。
- log.error("handle message ack fail", ex)：记录一条错误日志，表明消息的确认失败，并输出具体异常信息。
onSuccess(CorrelationData.Confirm result)
- 如果消息成功被确认，则会进入onSuccess()方法。
- result.isAck()：判断确认结果是否为ack，即成功确认。
- 如果isAck()返回true，则记录日志，表示消息成功接收。
- 如果返回false，则表示消息未被确认（即nack），需要处理相应的失败情况，记录错误日志并输出原因。

发送消息

rabbitTemplate.convertAndSend("hmall.direct", "red1", "hello", cd)
- rabbitTemplate是用于发送消息的模板对象。
- convertAndSend()方法用于发送消息，参数分别为：
  - "hmall.direct"：交换机名称。
  - "red1"：路由键。
  - "hello"：消息内容。
  - cd：之前创建的CorrelationData对象，用于追踪该消息的状态。

方法执行的时机

onFailure(Throwable ex)方法会在消息确认过程中出现异常时被调用，例如由于网络中断或RabbitMQ服务器不可用等情况，导致消息无法被正确发送或确认。
onSuccess(CorrelationData.Confirm result)方法会在RabbitMQ成功处理消息时被调用，返回确认结果（ack）或者未确认结果（nack）。如果交换机成功接收了消息但队列出现问题，则可能返回nack。

代码示例

@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 {
                log.error("发送消息失败, 收到 nack, reason: {}", result.getReason());
            }
        }
    });

    // 3. 发送消息
    rabbitTemplate.convertAndSend("hmall.direct", "red1", "hello", cd);
}