RabbitMQ是如何确定消息是否投递到队列中的

码农小胖哥

发布于 2020-10-10 02:24:51

2.9K00

代码可运行

文章被收录于专栏：码农小胖哥的码农生涯码农小胖哥的码农生涯

运行总次数：0

代码可运行

1. 前言

在使用RabbitMQ消息中间件时，因为消息的投递是异步的，默认情况下，RabbitMQ会删除那些无法路由的消息。为了能够检出消息是否顺利投递到队列，我们需要相应的处理机制。今天就来验证一下相关的验证机制。

2. 消息投递失败

那么哪些情况消息会投递失败呢？RabbitMQ消息会先到达指定的交换机，然后由交换机路由到对应的队列。所以以下几种情况会导致消息投递失败。

投递的交换机不可用。
投递的交换机可用，但是没有匹配到队列。

3. 投递失败的处理机制

对应上面的两种情况，RabbitMQ提供了对应的解决方案。

ConfirmCallback

RabbitMQ提供了ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后进行确认回调。

在Spring Boot中需要开启：

spring:
  rabbitmq:
  # 通常选择 correlated
    publisher-confirm-type:

通常有三种选择：

NONE ，禁用发布确认模式，是默认值。
CORRELATED，发布消息时会携带一个CorrelationData，被ack/nack时CorrelationData会被返回进行对照处理，CorrelationData可以包含比较丰富的元信息进行回调逻辑的处理。
SIMPLE，当被ack/nack后会等待所有消息被发布，如果超时会触发异常，甚至关闭连接通道。

这里我使用CORRELATED模式，声明一个ConfirmCallback并设置到RabbitTemplate中

rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
    // correlationData 可能为空
    if (ack) {
        log.debug("消息发送到exchange成功,id: {}", correlationData.getId());
    } else {
        log.debug("消息发送到exchange失败,原因: {}", cause);
    }
});

当消息投递到一个不存在的交换机Exchange且ack=false时会输出日志：

- Publishing message [(Body:'"hello"' MessageProperties [headers={spring_listener_return_correlation=a088eb3f-a234-4e15-bb7a-3aa9a6f043e6, spring_returned_message_correlation=29975bc1-f363-4e3a-85ca-010d13888720, __TypeId__=java.lang.String}, contentType=application/json, contentEncoding=UTF-8, contentLength=7, deliveryMode=PERSISTENT, priority=0, deliveryTag=0])] on exchange [DIRECT_EXCHANGE1], routingKey = [DIRECT_ROUTING_KEY2]

- 消息发送到exchange失败,原因: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'DIRECT_EXCHANGE1' in vhost 'my_vhost', class-id=60, method-id=40)

这里实现的比较简单你可以增加一些消息投递到交换机失败后的操作处理逻辑。

ReturnCallback

ReturnCallback接口用于实现消息已经成功发送到RabbitMQ交换机，但没有匹配到队列时的回调。

在Spring Boot中需要同时开启：

spring:
  rabbitmq:
    publisher-returns: true
    template:
      mandatory: true

RabbitTemplate中的mandatory设置值优先级要高一些。

我们声明一个ReturnCallback并设置到RabbitTemplate中

rabbitTemplate.setMandatory(true);
rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
    String correlationId = message.getMessageProperties()
            .getHeader(PublisherCallbackChannel.RETURNED_MESSAGE_CORRELATION_KEY);
    log.debug("消息：{} 发送失败, 应答码：{} 原因：{} 交换机: {}  路由键: {}", correlationId,
            replyCode, replyText, exchange, routingKey);
});

当消息成功投递到交换机但是无法匹配到队列时：

- Publishing message [(Body:'"hello"' MessageProperties [headers={spring_listener_return_correlation=155648bd-fc3e-4c8b-a650-7b1ce720c7a6, spring_returned_message_correlation=7029ee49-357a-42fc-8532-dc41b4bb8e87, __TypeId__=java.lang.String}, contentType=application/json, contentEncoding=UTF-8, contentLength=7, deliveryMode=PERSISTENT, priority=0, deliveryTag=0])] on exchange [DIRECT_EXCHANGE], routingKey = [DIRECT_ROUTING_KEY2]

- 消息：7029ee49-357a-42fc-8532-dc41b4bb8e87 发送失败, 应答码：312 原因：NO_ROUTE 交换机: DIRECT_EXCHANGE  路由键: DIRECT_ROUTING_KEY2
- 消息发送到exchange成功,id: 7029ee49-357a-42fc-8532-dc41b4bb8e87

从上面我们也可以看出ReturnCallback只处理投递到队列失败的情况，并不像ConfirmCallback既能处理失败的情况也能处理成功的情况。

4. 总结

消息投递失败的处理在使用RabbitMQ的使用中时非常必要的，能够帮助我们追踪消息的投递情况，以及处理消息投递异常或者成功后的逻辑处理，为消息丢失进行一些兜底或者记录。但是请注意这个并不是发生在消费阶段，是否成功消费并不是由这两种回调来处理，我们有空再对消息的消费确认进行讲解。多多关注：码农小胖哥 获取更多的编程干货。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2020-09-27，如有侵权请联系 cloudcommunity@tencent.com 删除

rabbitmq