众所周知，RabbitMQ在保证消息可靠投递的实现过程中有个参数mandatory。该参数的作用是，当消息的mandatory设置为true时，消息投递到Exchange之后，如果Exchange无法将该消息路由到任何一个队列，那么该消息将返回给生产者。当设置为false，RabbitMQ将直接丢弃该消息。
- 在了解了这个背景之后，分为使用和不使用
spring-boot-starter-amqp两种场景。
- 在未使用
spring-boot-starter-amqp的场景下，我们直接给channel设置监听器并且将消息的mandatory设置为true，即可实现消息无法路由之后通过该channel将消息return给生产者。即:
- channel设置监听回调函数。
- 消息的
mandatory设置为true。
- 而在使用
spring-boot-starter-amqp的场景下，除了设置mandatory，还需要设置spring.rabbitmq.publisher-returns，这个参数的作用是什么呢。并且在SpringBoot下尽管设置了mandatory为true，但是同时spring.rabbitmq.publisher-returns为false，还是无法监听到路由失败return的消息。这是为什么？
演示过程:
定义一个Exchange，不绑定任何Queue，向该Exchange投递mandatory为true消息，那么肯定是路由失败并且需要返回生产者。
# 不使用spring-boot-starter-amqp场景:
ConnectionFactory connectionFactory = new ConnectionFactory(); //connectionFactory.setUsername("futao"); //connectionFactory.setPassword("123456789"); connectionFactory.setVirtualHost("/tech-sharing"); //connectionFactory.setHost("localhost"); //connectionFactory.setPort(5672); connectionFactory.setUri("amqp://futao:123456789@localhost:5672"); // 创建TCP连接 Connection connection = connectionFactory.newConnection(); // 创建通道 final Channel channel = connection.createChannel(); // 监听被return的消息 channel.addReturnListener(new ReturnCallback() { @Override public void handle(Return returnMessage) { log.info("消息被退回:{}", returnMessage); } }); // 定义交换机 channel.exchangeDeclare("X_SIMPLE_NO_MATCH_QUEUE", BuiltinExchangeType.TOPIC, false, true, new HashMap<>(0)); AMQP.BasicProperties basicProperties = new AMQP.BasicProperties(); // 向Exchange投递mandatory为true的消息 channel.basicPublish("X_NO_MATCH_QUEUE", "", true, false, basicProperties, "123".getBytes(StandardCharsets.UTF_8)); TimeUnit.MINUTES.sleep(5L); channel.close(); connection.close();
消息被投递到Exchange之后立即被退回并被我们的程序监听到。
- 使用wireshark来查看这一过程。
接下来看看使用SpringBootStarter的方式。
# 使用spring-boot-starter-amqp:
- 配置(注意
spring.rabbitmq.publisher-returns):
spring.rabbitmq.host=localhost spring.rabbitmq.port=5672 spring.rabbitmq.username=futao spring.rabbitmq.password=123456789 spring.rabbitmq.virtual-host=/tech-sharing # 当exchange无法找到任何一个合适的queue时，将消息return给生产者 spring.rabbitmq.template.mandatory=true # 必须设置为true，否则消息消息路由失败也无法触发Return回调 spring.rabbitmq.publisher-returns=true
- 设置回调函数
@Slf4j @Component public class RabbitTemplateEnhance implements BeanPostProcessor { @Override public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { if (bean instanceof RabbitTemplate) { log.debug("增强 RabbitTemplate"); RabbitTemplate rabbitTemplate = (RabbitTemplate) bean; // return回调函数 rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() { @Override public void returnedMessage(ReturnedMessage returnedMessage) { log.error("消息被退回:{}", returnedMessage); } }); return rabbitTemplate; } return bean; } }
- 发送消息
public void send() { log.info("发送消息"); Order order = new Order(1, BigDecimal.TEN, OrderStatusEnum.UN_PAY.getStatus()); Message message = MessageBuilder .withBody(JSON.toJSONString(order).getBytes(StandardCharsets.UTF_8)) .setContentEncoding(StandardCharsets.UTF_8.displayName()) .setContentType(MessageProperties.CONTENT_TYPE_JSON) .build(); rabbitTemplate.convertAndSend(EXCHANGE_NAME, "", message); }
如果将spring.rabbitmq.publisher-returns设置为flase进行测试，会惊奇的发现并不会触发回调方法。那么排查过程是首先是看RabbitMQ到底有没有通知生产者有消息被退回。使用wireshark抓包发现有一条Basic.Return报文，消息确实被退回了，那么我们的程序为什么没有触发回调函数呢？
Debug:
还是回到不使用SpringBoot的代码，在回调函数出打断点查看调用链。
监听器是被ChannelIN.processAsync()方法触发的。
ChannelIN.processAsync()中遍历了所有的监听器。那么查看监听器是如何被加入returnListeners集合的即可。
接下来分析为什么将spring.rabbitmq.publisher-returns设置为flase的场景下我们的监听器为什么没有被加入集合。
在添加回调监听器的地方打上断点
回到SpringBoot环境下Debug: 分析监听器是如何被加入到集合的。
发现这段代码会根据confirmsOrReturnsCapable的值判断是否需要向Channel添加监听器，而confirmsOrReturnsCapable的值来自于RabbitProperties的isPublisherReturns
即：只有将spring.rabbitmq.publisher-returns设置为true，才会向Channel添加我们设置的监听回调函数。
spring.rabbitmq.publisher-returns参数的作用分析完了，那么mandatory属性是如何被设置到Message的呢：
通过rabbitTemplate.convertAndSend()一路往下debug会发现mandatory的取值过程：
即bool mandatory = (RabbitTemplate.this.returnsCallback != null || (correlationData != null && StringUtils.hasText(correlationData.getId()))) && isMandatoryFor(message)
RabbitTemplate.this.returnsCallback != null我们在第二步设置了ReturnsCallback，所以为true- 那么就取决于
isMandatoryFor(message)这个方法的返回值
- 而
isMandatoryFor(message)的值实际上来自:
如果mandatory != null，则取mandatory，否则取PublisherReturns的值。
那么是不是意味着如果将设置mandatory为null，publisherReturns的作用可以完全替代mandatory?
# 当exchange无法找到任何一个合适的queue时，将消息return给生产者 spring.rabbitmq.template.mandatory= # 必须设置为true，否则消息消息路由失败也无法触发Return回调 spring.rabbitmq.publisher-returns=true
- 经过测试发现，确实是这样。如果将
spring.rabbitmq.template.mandatory设置为null，那么mandatory的效果可以完全被spring.rabbitmq.publisher-returns来控制。
# 总结
SpringBootStarter环境下要触发路由失败消息的回调可以有两种设置方式:
spring.rabbitmq.template.mandatory=true spring.rabbitmq.publisher-returns=true
OR
spring.rabbitmq.template.mandatory= spring.rabbitmq.publisher-returns=true
SpringBoot为什么要加这个参数？？？
