你可知道publisherReturns参数在spring-boot-starter-amqp中的作用？
你可知道publisherReturns参数在spring-boot-starter-amqp中的作用？
2021-02-21 万寿宫. NC.
众所周知，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();
图片.png
消息被投递到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为什么要加这个参数？？？
欢迎在评论区留下你看文章时的思考，及时说出，有助于加深记忆和理解，还能和像你一样也喜欢这个话题的读者相遇~