RabbitMQ消息路由失败的处理方案(回调与备份交换机AE)

简介: RabbitMQ消息路由失败的处理方案(回调与备份交换机AE)

我们知道,消息在RabbitMQ的整个生命周期是生产者投递消息ExchangeExchange根据路由键消息路由到合适的QueueQueue再将消息推(或消费者主动拉)给消费者

在这个过程当中,Exchange根据路由键将消息路由到合适的Queue的过程,可能发生诸如


  1. Exchange没有任何Queue与其绑定,
  2. 或者根据消息的路由键,没有任何一个合适的Queue来投递消息,

从而导致消息路由失败。

对于这些路由失败的消息应该如何处理呢?

有两种方式:

  1. 将消息返回给投递该条消息的生产者。
  2. 使用备份交换机 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);
    }
});


  • 消息被退回:且可以看到原因是无法路由


image.png


方式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);
}


  • 结果是消息被路由到备份交换机的备份队列


image.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


image.png

image.png

image.png


image.png


备份交换机绑定的队列已经接收到了路由失败的消息


image.png


  • 其他要注意的点:
  • 备份交换机的Type设置为fanout比较合适,这样可以忽略RoutingKey,避免备份交换机又路由失败。
  • 被投递到备份交换机的RoutingKey为消息投递到MQ时的原始RoutingKey,不会变,这一点在其他场景下也是一样的。
  • 使用备份交换机模式,mandatory将无效,即就算mandatory设置为false,路由失败的消息同样会被投递到绑定的备份交换机。


# 源代码

https://gitee.com/FutaoSmile/tech-sharing-mq

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
2月前
|
消息中间件 存储 缓存
RabbitMQ中的交换机
RabbitMQ中的交换机
34 2
|
2月前
|
消息中间件 JSON 缓存
RabbitMQ快速学习之WorkQueues模型、三种交换机、消息转换器(SpringBoot整合)
RabbitMQ快速学习之WorkQueues模型、三种交换机、消息转换器(SpringBoot整合)
|
2月前
|
消息中间件 Java Spring
SpringBoot实现RabbitMQ的广播交换机(SpringAMQP 实现Fanout广播交换机)
SpringBoot实现RabbitMQ的广播交换机(SpringAMQP 实现Fanout广播交换机)
31 2
|
12天前
|
存储 Linux 网络安全
|
15天前
|
消息中间件 RocketMQ
消息队列 MQ产品使用合集之在开源延时消息插件方案中和原生延时消息方案中,同时设置参数是否会出现错乱
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
20天前
|
消息中间件
02.交换机RabbitMQ交换机
02.交换机RabbitMQ交换机
14 0
|
2月前
|
消息中间件 Java
SpringBoot实现RabbitMQ的通配符交换机(SpringAMQP 实现Topic交换机)
SpringBoot实现RabbitMQ的通配符交换机(SpringAMQP 实现Topic交换机)
22 1
|
2月前
|
消息中间件 Java Spring
SpringBoot实现RabbitMQ的定向交换机(SpringAMQP 实现Direct定向交换机)
SpringBoot实现RabbitMQ的定向交换机(SpringAMQP 实现Direct定向交换机)
30 1
|
2月前
|
消息中间件 微服务
RabbitMQ入门指南(十):延迟消息-死信交换机
RabbitMQ是一个高效、可靠的开源消息队列系统,广泛用于软件开发、数据传输、微服务等领域。本文主要介绍了死信交换机、死信交换机实现延迟消息等内容。
94 0

热门文章

最新文章