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

[[344036]]

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(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的使用中时非常必要的，能够帮助我们追踪消息的投递情况，以及处理消息投递异常或者成功后的逻辑处理，为消息丢失进行一些兜底或者记录。但是请注意这个并不是发生在消费阶段，是否成功消费并不是由这两种回调来处理，我们有空再对消息的消费确认进行讲解。

本文转载自微信公众号「码农小胖哥」，可以通过以下二维码关注。转载本文请联系码农小胖哥公众号。

当前文章：RabbitMQ是如何确定消息是否投递到队列中的
网站地址：http://www.shufengxianlan.com/qtweb/news11/467361.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联