我们知道，消息在RabbitMQ的整个生命周期是生产者投递消息到Exchange，Exchange根据路由键将消息路由到合适的Queue，Queue再将消息推(或消费者主动拉)给消费者。
在这个过程当中，Exchange根据路由键将消息路由到合适的Queue的过程，可能发生诸如
- Exchange没有任何Queue与其绑定，
- 或者根据消息的路由键，没有任何一个合适的Queue来投递消息，
从而导致消息路由失败。
对于这些路由失败的消息应该如何处理呢？
有两种方式：
- 将消息返回给投递该条消息的生产者。
- 使用备份交换机 alternate-exchange（AE）。
方式1：将消息返回给投递该条消息的生产者
- 配置
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 NoMatchQueue {
/**
* 交换机名称
*/
public static final String EXCHANGE_NAME = "X_NO_MATCH_QUEUE";
@Autowired
private RabbitTemplate rabbitTemplate;
@PostConstruct
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);
}
}
@Configuration
class ExchangeDeclare {
/**
* 只定义一个交换机，但是不绑定任何Queue，所以发送到该Exchange的消息都会路由失败
*
* @return
*/
@Bean
public Exchange noMatchQueueExchange() {
return ExchangeBuilder
.topicExchange(NoMatchQueue.EXCHANGE_NAME)
.durable(true)
.build();
}
}
- 设置回调函数
rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
@Override
public void returnedMessage(ReturnedMessage returnedMessage) {
log.error("消息被退回:{}", returnedMessage);
}
});
-
消息被退回：且可以看到原因是无法路由
图片.png 详细过程请参考: 你可知道publisherReturns参数在spring-boot-starter-amqp中的作用？
方式2：使用备份交换机
使用方式1需要我们在程序中进行编码设置回调函数监听，增加了生产者代码的复杂性，那么为了消息不丢失还有没有其他方式来处理路由失败的消息呢： 答案是使用备份交换机。
- 相较于使用回调函数，使用备份交换机只需要给交换机绑定一个备份交换机即可，当消息路由失败之后，消息将投递到备份交换机，再由备份交换机路由消息到备份队列。这样我们只需要关注这个备份队列就能知道/获取到路由失败的消息。通常情况下备份交换的Type应该设置为
fanout。 - 配置
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=false
# 必须设置为true，否则消息消息路由失败也无法触发Return回调
spring.rabbitmq.publisher-returns=false
- 注意: 使用备份交换机模式，mandatory将无效，即就算mandatory设置为false，路由失败的消息同样会被投递到绑定的备份交换机。
- 正常业务交换机(不绑定队列，使得消息一定会路由失败)
/**
* 业务交换机
*
* @return
*/
@Bean
public Exchange noMatchQueueExchange() {
return ExchangeBuilder
.topicExchange(NoMatchQueueAlternateExchange.EXCHANGE_NAME)
.durable(true)
// 绑定备份交换机
.alternate(X_ALTERNATE)
.build();
}
- 备份交换机/队列/绑定
/**
* 备份队列
*
* @return
*/
@Bean
public Queue alternateQueue() {
return QueueBuilder
.durable("Q_ALTERNATE")
.build();
}
/**
* 备份交换机
*
* @return
*/
@Bean
public Exchange alternateExchange() {
return ExchangeBuilder
.fanoutExchange(X_ALTERNATE)
.durable(true)
.build();
}
/**
* 备份绑定
*
* @param alternateExchange
* @param alternateQueue
* @return
*/
@Bean
public Binding alternateBinding(Exchange alternateExchange, Queue alternateQueue) {
return BindingBuilder
.bind(alternateQueue)
.to(alternateExchange)
.with("")
.noargs();
}
- 消息投递
/**
* 正常业务交换机
*/
public static final String EXCHANGE_NAME = "X_NO_MATCH_QUEUE_ALTERNATE";
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 发送消息
*/
@PostConstruct
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);
}
-
结果是消息被路由到备份交换机的备份队列
图片.png - 且：如果你同时使用了两种方式，即（mandatory为true+Listener监听）和（备份交换机AlternateExchange），消息将只会路由到备份交换机，不会Return回生产者。
- 在原生RabbitMQ-client中演示这一过程:
@Slf4j
public class AeTest {
/**
* 获取Channel
*/
private static final Channel CHANNEL = MqChannelUtils.getChannel();
/**
* 备份交换机
*/
private static final String X_AE = "X_AE";
/**
* 备份交换机绑定的队列
*/
private static final String Q_AE = "Q_AE";
/**
* 正常业务的交换机
*/
private static final String X_1 = "X_1";
public static void main(String[] args) throws IOException {
// 定义备份交换机-其实也是一个正常的交换机
CHANNEL.exchangeDeclare(X_AE, BuiltinExchangeType.FANOUT, true);
// 定义备份队列
CHANNEL.queueDeclare(Q_AE, true, false, false, null);
// 绑定备份
CHANNEL.queueBind(Q_AE, X_AE, "");
HashMap<String, Object> arguments = new HashMap<>();
// 绑定的备份交换机
arguments.put("alternate-exchange", X_AE);
// 定义交换机
CHANNEL.exchangeDeclare(X_1, BuiltinExchangeType.TOPIC, false, false, arguments);
// 添加监听器，看看是否还会return消息
CHANNEL.addReturnListener(new ReturnCallback() {
@Override
public void handle(Return returnMessage) {
log.error("消息被退回{}", returnMessage);
}
});
// 尝试向交换机发送消息（无法路由）- mandatory参数无效
CHANNEL.basicPublish(X_1, "", false, false,
new AMQP.BasicProperties(), "阿依古丽".getBytes(StandardCharsets.UTF_8));
}
}
-
两个交换机，正常的交换机
X_1和备份交换机X_AE
图片.png
图片.png
图片.png -
备份交换机绑定的队列已经接收到了路由失败的消息
图片.png -
其他要注意的点：
- 备份交换机的Type设置为fanout比较合适，这样可以忽略RoutingKey，避免备份交换机又路由失败。
- 被投递到备份交换机的RoutingKey为消息投递到MQ时的原始RoutingKey，不会变，这一点在其他场景下也是一样的。
- 使用备份交换机模式，mandatory将无效，即就算mandatory设置为false，路由失败的消息同样会被投递到绑定的备份交换机。
