springcloud:RabbitMQ死信队列与延迟交换机实现(四)

简介: 死信队列是消息队列中非常重要的概念,同时我们需要业务场景中都需要延迟发送的概念,比如12306中的30分钟后未支付订单取消。那么本期,我们就来讲解死信队列,以及如何通过延迟交换机来实现延迟发送的需求。

0.引言

死信队列是消息队列中非常重要的概念,同时我们需要业务场景中都需要延迟发送的概念,比如12306中的30分钟后未支付订单取消。那么本期,我们就来讲解死信队列,以及如何通过延迟交换机来实现延迟发送的需求。

1. 死信队列

1.2 什么是死信?

理解死信队列前,我们先讲解什么是死信,所谓死信就是没有被成功消费的消息,但并不是所有未成功消费的消息都是死信消息,死信消息的产生来源于以下三种途径:
(1)消息被消费者拒绝,参数requeue设置为false的消息
(2)过期的消息,过期消息分为两种:

    a. 发送消息时,设置了某一条消息的生存时间(message TTL),如果生存时间到了,消息还没有被消费,就会被标注为死信消息
    b. 设置了队列的消息生存时间,针对队列中所有的消息,如果生存时间到了,消息还没有被消费,就会被标注为死信消息

(3)当队列达到了最大长度后,再发送过来的消息就会直接变成死信消息

1.3 什么是死信队列?

直接来讲,用来盛装死信的队列就是死信队列,好像是一句废话,所以其重点在于理解死信的概念。

死信队列的作用:
(1)队列在已满的情况下,会将消息发送到死信队列中,这样消息就不会丢失了,回头再从死信队列里将消息取出来进行消费即可
(2)可以基于死信队列实现延迟消费的效果。具体的实现我们后续讲解

1.4 创建死信交换机、死信队列

死信交换机、死信队列其实都是普通的交换机、队列,只是专门声明出来用于存储死信消息的。我们只需要通过deadLetterExchange方法来声明死信交换机,然后用deadLetterRoutingKey方法来声明死信队列

如下代码所示,我们创建了test.queuetest.exchangedead.queuedead.exchange,并且在test.queue中将死信交换机和死信路由指定到了测试队列中

注意:涉及到修改队列、交换机属性的,如果该队列、交换机已经存在需要将其删除后才能生效,否则可能还会报错。

@Configuration
public class RabbitMqConfig {

    private static final String TEST_EXCHANGE = "test.exchange";
    private static final String TEST_QUEUE = "test.queue";
    private static final String TEST_ROUTING_KEY = "test.routing.key";

    private static final String DEAD_EXCHANGE = "dead.exchange";
    private static final String DEAD_QUEUE = "dead.queue";
    private static final String DEAD_ROUTING_KEY = "dead.routing.key";
    
    @Bean
    public Queue deadQueue(){
        return new Queue(DEAD_QUEUE);
    }

    @Bean
    public DirectExchange deadExchange(){
    // 设置演示,使用了直接交换机Direct,大家可以根据自己的业务情况声明为其他类型的交换机
        return new DirectExchange(DEAD_EXCHANGE);
    }

    @Bean
    public Binding deadBinding(Queue deadQueue,Exchange deadExchange){
        return BindingBuilder.bind(deadQueue).to(deadExchange).with(DEAD_ROUTING_KEY).noargs();
    }

    @Bean
    public Queue testQueue(){
        return QueueBuilder.durable(TEST_QUEUE).deadLetterExchange(DEAD_EXCHANGE).deadLetterRoutingKey(DEAD_ROUTING_KEY).build();
    }

    @Bean
    public DirectExchange testExchange(){
        return new DirectExchange(TEST_EXCHANGE);
    }

    @Bean
    public Binding testQueueBing(Queue testQueue, DirectExchange testExchange){
        return BindingBuilder.bind(testQueue).to(testExchange).with(TEST_ROUTING_KEY);

    }
}

1.5 实现死信消息

1.5.1 基于消费者进行reject或nack实现死信消息

@Component
public class QueueListener {

    @RabbitListener(queues = RabbitMqConfig.TEST_QUEUE)
    public void handler(MyMessage messageInfo, Message message, Channel channel) {
        try{
            System.out.println("接收的消息:"+messageInfo.toString());
            // requeue参数设置为false 设置死信消息
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
            // multiple和requeue设置为false 设置死信消息
            channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
            // 返回ack 确认接收到消息
//            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }catch (IOException e){
            try {
                channel.basicRecover();
            } catch (IOException ex) {
                ex.printStackTrace();
                log.error("消息处理失败:{}",e.getMessage());
            }
        }
    }
}

1.5.2 基于生存时间实现

(1)发送消息时设置生存时间

    @GetMapping("sendTestQueueWithExpiration")
    public String sendTestQueueWithExpiration(){
        MyMessage message = new MyMessage(1L,"物流提醒","到达装货区域,注意上传凭证",new Date());
        rabbitTemplate.convertAndSend(RabbitMqConfig.TEST_EXCHANGE,RabbitMqConfig.TEST_ROUTING_KEY, message,msg -> {
            msg.getMessageProperties().setExpiration("5000");
            return msg;
        });
        return "发送成功";
    }

(2)队列设置生存时间

    @Bean
    public Queue testQueue(){
        return QueueBuilder.durable(TEST_QUEUE)
                .deadLetterExchange(DEAD_EXCHANGE)
                .deadLetterRoutingKey(DEAD_ROUTING_KEY)
                // 10s 过期
                .ttl(10000)
                .build();
    }

1.5.3 基于队列max_length实现

    @Bean
    public Queue testQueue(){
        return QueueBuilder.durable(TEST_QUEUE)
                .deadLetterExchange(DEAD_EXCHANGE)
                .deadLetterRoutingKey(DEAD_ROUTING_KEY)
                // 容量最大100条
                .maxLength(100)
                .build();
    }

1.6 基于死信队列实现消息延迟发送

上述我们说过死信队列还可以消息延迟发送,其思路就是:
(1)消息发送时设置消息的生存时间,其生存时间就是我们想要延迟的时间
(2)消息者监控死信队列进行消费

正常队列的消息因为没有消费者消费,同时又指定了生存时间,到达时间后消息转发到死信队列中,消费者监听了死信队列从而将其消费掉。

基于死信队列实现消息延迟发送的问题

如果有两个消息,一个是5s生存时间,一个是10s生存时间,当我们先发送了10s生存时间的消息到queue中时,因为rabbitmq只会监控队列最外侧的消息的生存时间,也就是监控10s生存时间的消息,而5s生存时间的消息只会在最外侧的10s消息到期后才会监控,也就导致我实际需要5s生存的消息,实际需要10s才监听到了。

所以呢,基于死信队列实现的延迟消息,只使用于延迟时间一致的消息。

为了适配更多的延迟场景,已经更加简单的实现延迟消息,我们引入了延迟交换机

2. 延迟交换机

延迟交换机并不是rabbitmq自带的功能,而是要通过安装延迟交换机插件delayed_message_exchange来实现

其插件的安装我们之间已经讲解过,不再累叙,可以参考如下博文
springcloud:安装rabbitmq并配置延迟队列插件

通过延迟交换机实现的延迟消息,其重点主要在交换机上,队列就是普通队列,消息发送到交换机上后,会记录消息的延迟时间,到达时间后才会发送到队列中,这样消费者通过监控队列,就能在指定时间获取到消息

因此延迟交换机与普通交换机的实现,只在创建交换机时,其他的操作与普通交换机无异,因此使用起来也很方便

创建延迟交换机,通过x-delayed-type属性声明交换机类型,可以是direct也可以是topic,具体支持4中交换机类型,如果不清楚的可以参考之前的博文

@Configuration
public class RabbitMqDelayConfig {

    public static final String DELAY_EXCHANGE = "delay.exchange";

    public static final String DELAY_QUEUE = "delay.queue";

    public static final String DELAY_ROUTING_KEY = "delay.routing.key";

    @Bean
    public Exchange delayExchange(){
        Map<String, Object> arguments = new HashMap<>(1);
        arguments.put("x-delayed-type","direct");
        return new CustomExchange(DELAY_EXCHANGE,"x-delayed-message",true,false,arguments);
    }

    @Bean
    public Queue delayQueue(){
        return new Queue(DELAY_QUEUE);
    }
    
    @Bean
    public Binding delayBinding(Queue delayQueue, Exchange delayExchange){
        return BindingBuilder.bind(delayQueue).to(delayExchange).with(DELAY_ROUTING_KEY).noargs();
    }


}

发送消息时指定延迟时间,单位毫秒

rabbitTemplate.convertAndSend(DelayedConfig.DELAYED_EXCHANGE, "delayed.abc", "xxxx", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setDelay(30000);
                return message;
            }
        });

我们还可以将该方法封装为工具类方法,方便之后调用

/**
     * 发送 延迟队列
     * @param exchange 交换机
     * @param routeKey 路由
     * @param message 消息
     * @param delaySecond 延迟秒数
     */
    public void send(String exchange, String routeKey, Object message, int delaySecond){  
        rabbitTemplate.convertAndSend(exchange,routeKey,message,msg -> {
            // 消息持久化 
            msg.getMessageProperties().setDelay(delaySecond * 1000);
            return msg;
        });
    }

3. 应用场景

延迟消息的应用场景丰富,除了我们开篇所说的30分钟未支付自动取消订单,还比如到货后72小时未签收自动签收

基本上所有需要延迟触发的业务场景都可以用rabbitmq延迟队列来实现。

4. 练习题

对于刚接触rabbitmq的同学,这里我提供一个练习题给大家,也让大家在实操中加强对于rabbitmq的理解:

需求:订单到货后72小时未签收,自动签收
讲解:我们这里要实现订单到货后的自动签收功能,订单到货后会触发发送自动签收消息的方法,订单已签收的状态status为2,到货状态为1,如果72小时前已经签收了即status被更新为2了,那么需要取消自动签收(不执行自动签收,即忽略自动签收消息)

做出来的同学可以私聊我索要答案源码

相关实践学习
快速体验阿里云云消息队列RocketMQ版
本实验将带您快速体验使用云消息队列RocketMQ版Serverless系列实例进行获取接入点、创建Topic、创建订阅组、收发消息、查看消息轨迹和仪表盘。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
消息中间件 NoSQL 大数据
RocketMQ实战—5.消息重复+乱序+延迟的处理
本文围绕RocketMQ的使用与优化展开,分析了优惠券重复发放的原因及解决方案。首先,通过案例说明了优惠券系统因消息重复、数据库宕机或消费失败等原因导致重复发券的问题,并提出引入幂等性机制(如业务判断法、Redis状态判断法)来保证数据唯一性。其次,探讨了死信队列在处理消费失败时的作用,以及如何通过重试和死信队列解决消息处理异常。接着,分析了订单库同步中消息乱序的原因,提出了基于顺序消息机制的代码实现方案,确保消息按序处理。此外,介绍了利用Tag和属性过滤数据提升效率的方法,以及延迟消息机制优化定时退款扫描的功能。最后,总结了RocketMQ生产实践中的经验.
RocketMQ实战—5.消息重复+乱序+延迟的处理
什么是死信交换机 ? 如何为队列绑定死信交换机 ?
死 信交换机和正常的交换机没有什么不同 , 如果一个包含死信的队列配置了dead-letter-exchange属性,指定了一个交换机,那么队列中的死信就会投递到这个交换机中,而这个交换机称为死信交换机 为队列绑定死信交换机 , 只需要设置队列属性 dead-letter-exchange即可
|
消息中间件 存储 缓存
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
2271 8
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
|
消息中间件 存储 监控
RabbitMQ 队列之战:Classic 和 Quorum 的性能洞察
RabbitMQ 是一个功能强大的消息代理,用于分布式应用程序间的通信。它通过队列临时存储消息,支持异步通信和解耦。经典队列适合高吞吐量和低延迟场景,而仲裁队列则提供高可用性和容错能力,适用于关键任务系统。选择哪种队列取决于性能、持久性和容错性的需求。
1309 6
|
消息中间件
rabbitmq,&队列
rabbitmq,&队列
|
消息中间件 JSON Java
玩转RabbitMQ声明队列交换机、消息转换器
玩转RabbitMQ声明队列交换机、消息转换器
631 0
|
消息中间件 数据管理 Serverless
阿里云消息队列 Apache RocketMQ 创新论文入选顶会 ACM FSE 2025
阿里云消息团队基于 Apache RocketMQ 构建 Serverless 消息系统,适配多种主流消息协议(如 RabbitMQ、MQTT 和 Kafka),成功解决了传统中间件在可伸缩性、成本及元数据管理等方面的难题,并据此实现 ApsaraMQ 全系列产品 Serverless 化,助力企业提效降本。
|
消息中间件 Java Kafka
消息队列比较:Spring 微服务中的 Kafka 与 RabbitMQ
本文深入解析了 Kafka 和 RabbitMQ 两大主流消息队列在 Spring 微服务中的应用与对比。内容涵盖消息队列的基本原理、Kafka 与 RabbitMQ 的核心概念、各自优势及典型用例,并结合 Spring 生态的集成方式,帮助开发者根据实际需求选择合适的消息中间件,提升系统解耦、可扩展性与可靠性。
781 1
消息队列比较:Spring 微服务中的 Kafka 与 RabbitMQ
|
消息中间件 JSON Java
开发者如何使用轻量消息队列MNS
【10月更文挑战第19天】开发者如何使用轻量消息队列MNS
1182 107