【RabbitMQ】死信（延迟队列）的使用

作者：不正经 | 2024-02-14 22:56:13

踩

一、介绍

一、介绍

1、什么是死信队列(延迟队列)

        死信，在官网中对应的单词为“Dead Letter”,它是 RabbitMQ 的一种消息机制。
        死信队列（Dead Letter Queue）和延迟队列（Delay Queue）是两种不同的队列类型，但在实际应用中它们可以结合使用。
        死信队列是当消息在队列中因为过期、被拒绝等原因无法正常处理时，会被重新发送到另一个交换机上，这个交换机就是死信交换机。死信队列可以用于实现重试机制、日志审计等特殊应用逻辑。

延迟队列则是一种特殊的队列类型，它允许将消息延迟指定的时间后才能被消费者消费。这种队列通常用于处理那些需要在特定时间点被处理的任务，例如定时任务、限时优惠等。在RabbitMQ中，可以通过设置消息的TTL（生存时间）来实现延迟队列的功能。当消息在队列中超过了TTL，它就会被移除并被发送到指定的死信交换机，进而被路由到死信队列中。

结合使用死信队列和延迟队列可以实现一些复杂的应用逻辑。

例如:

可以将某个需要延迟处理的消息发送到延迟队列中，并在消息过期之前将其存储在死信队列中。这样，当消息从延迟队列中移除时，它会被自动发送到死信队列中，然后由消费者消费并执行相应的操作。这种结合使用的方式可以提供更高的灵活性和可靠性，使得系统能够更好地应对各种异常情况。

2、应用场景

死信队列（Dead Letter Queue）和延迟队列（Delay Queue）在以下应用场景中表现优异：

这些场景都是通过结合使用死信队列和延迟队列来提高系统的可靠性、鲁棒性和灵活性。通过合理地设置死信队列和延迟队列的参数，可以实现各种复杂的业务逻辑和异常处理机制。

库存解锁服务：例如，当一个商品被锁定且无法被购买时，可以将其放入死信队列中，并在一定时间后重新发送到原始队列进行处理，从而解锁库存。
定时关单功能：例如，用户在商城下单成功并点击去支付后，如果在指定时间内未支付，系统可以自动将订单放入死信队列中，并在一定时间后重新发送到原始队列进行处理。
保证订单业务的消息数据不丢失：当消息消费发生异常时，可以将消息投入死信队列中，以便后续等到环境好了之后，再消费死信队列中的消息。
实现重试机制：当某个消息处理失败时，可以将它放入死信队列中，并在一段时间后重新发送到交换机进行重试。这样可以提高系统的鲁棒性，确保消息能够被正确处理。
处理定时任务和秒杀活动：使用延迟队列可以将消息延迟到特定的时间后进行处理，例如定时任务、秒杀活动等。这样可以确保在特定的时间点执行相应的操作。
日志处理和审计：将日志消息发送到死信队列中，可以在日志发生异常时进行记录和审计，以便分析和排查问题。

3、死信队列(延迟队列)的使用

使用死信队列（Dead Letter Queue）和延迟队列（Delay Queue）可以提高系统的可靠性和灵活性，特别是在处理异常情况、重试机制和定时任务等方面。

需要注意的是，在使用死信队列和延迟队列时，需要考虑系统的可用性和性能。如果大量的消息被放入死信队列中，可能会导致系统资源的过度消耗。因此，需要合理配置死信队列和延迟队列的大小和数量，以及监控和管理系统的性能和资源使用情况。

定义死信队列和延迟队列：在RabbitMQ中，可以在定义队列时指定队列类型为死信队列或延迟队列。例如，使用RabbitMQ的命令行工具或管理界面创建死信队列或延迟队列。
配置交换机和队列属性：在绑定交换机和队列时，可以设置一些属性来控制消息的路由和消费。例如，可以设置消息的TTL（生存时间）来实现延迟队列的功能，或者设置消息的优先级和延迟时间等属性。
发送消息到队列：使用RabbitMQ的生产者将消息发送到定义的队列中。如果需要将消息放入延迟队列中，可以在发送消息时设置相应的延迟时间。
处理消息：消费者从队列中获取消息并进行处理。如果消息无法正常处理，可以将它放入死信队列中。在处理消息时，可以根据需要设置重试机制，以便在消息处理失败时重新发送消息到队列中进行重试。
监控和管理：使用RabbitMQ的管理界面或命令行工具监控和管理死信队列和延迟队列的状态和消息。可以查看队列的大小、消息的延迟时间、消费者数量等信息，并根据需要进行调整和管理。

4、死信消息来源

消息 TTL 过期
队列满了，无法再次添加数据
消息被拒绝（reject 或 nack），并且 requeue =false

二、案例实践

1、案例一

生产者

在生产者的Config里面添加队列。


 //    =============死信队列===============
    // 定义QueueA Bean，返回QueueA实例
    @Bean
    public Queue QueueA() {
        Map<String, Object> config = new HashMap<>();
        //message在该队列queue的存活时间最大为5秒
        config.put("x-message-ttl", 5000);
        //x-dead-letter-exchange参数是设置该队列的死信交换器（DLX）
        config.put("x-dead-letter-exchange", "deadExchange");
        //x-dead-letter-routing-key参数是给这个DLX指定路由键
        config.put("x-dead-letter-routing-key", "deadQueue");
        // 返回QueueA实例
        return new Queue("QueueA", true, true, false, config);
    }
 
    // 定义DirectExchangeA Bean，返回DirectExchangeA实例
    @Bean
    public DirectExchange DirectExchangeA() {
        // 返回DirectExchangeA实例
        return new DirectExchange("DirectExchangeA");
    }
 
    // 定义bindingA Bean，将QueueA绑定到DirectExchangeA，并设置路由键为A.A
    @Bean
    public Binding bindingA() {
        // 将QueueA绑定到DirectExchangeA
        return BindingBuilder
                // 设置路由键为A.A
                .bind(QueueA())
                .to(DirectExchangeA())
                .with("A.A");
    }
 
    // 定义QueueB Bean，返回QueueB实例
    @Bean
    public Queue QueueB() {
        // 返回QueueB实例
        return new Queue("QueueB");
    }
 
    // 定义DirectExchangeB Bean，返回DirectExchangeB实例
    @Bean
    public DirectExchange DirectExchangeB() {
        // 返回DirectExchangeB实例
        return new DirectExchange("DirectExchangeB");
    }
 
    // 定义bindingB Bean，将QueueB绑定到DirectExchangeB，并设置路由键为B.B
    @Bean
    public Binding bindingB() {
        // 将QueueB绑定到DirectExchangeB
        return BindingBuilder
                // 设置路由键为B.B
                .bind(QueueB())
                .to(DirectExchangeB())
                .with("B.B");
    }

消费者

在消费者里面添加QueueA和QueueB

QueueA


import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
@Component
@SuppressWarnings("all")
@Slf4j
@RabbitListener(queues = "QueueA")
public class ReceiverQA {
 
        // 接收directExchange01交换机中Queue02队列消息的方法
        @RabbitHandler
        public void Queue02(String msg) {
            log.warn("QueueA，接收到信息:" + msg);
        }
 
}

QueueB


import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
@Component
@SuppressWarnings("all")
@Slf4j
@RabbitListener(queues = "QueueB")
public class ReceiverQB {
 
    // 接收directExchange01交换机中Queue02队列消息的方法
    @RabbitHandler
    public void Queue02(String msg) {
        log.warn("QueueB，接收到信息:" + msg);
    }
 
}

编写Controller层


    @RequestMapping("send5")
    public String send5() {
        rabbitTemplate.convertAndSend("DirectExchangeA", "A.A", "Hello,A");
        return "声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/不正经/article/detail/82405
推荐阅读
article深入详解使用 RabbitMQ 过程中涉及到的多个细节问题（面试可用）...
深入详解使用 RabbitMQ 过程中涉及到的多个细节问题，这些问题在面试时可能会被问到。深入详解使用 RabbitMQ...
                                    赞
踩
articledocker-compose.yml文件安装consul ftp redis rabbitmq ft...
docker-compose.yml文件安装consul redis rabbitmq ftp的配置文件_docker-...
                                    赞
踩
articleRabbitMq学习——Springboot整合rabbitmq之手动消息确认(ACK)_sprin...
一、前言前几天我研究了关于springboot整合简单消息队列，实现springboot推送消息至队列中，消费者成功消费...
                                    赞
踩
article虚拟机完成多节点网络配置，安装chronyc，OpenStack packages，mariadb ...
完成centos7安装，安装完成多节点网络配置，正确安装chronyc，可以同步主机时间。正确安装OpenStack p...
                                    赞
踩
articleRabbitMQ 安装...
消息队列RabbitMQ 安装                             下载erlang语言： erla...
                                    赞
踩
articleRabbitMQ插件详解：rabbitmq_web_stomp【RabbitMQ 六】...
本文着重介绍RabbitMQ中的rabbitmq_web_stomp插件，通过讲解STOMP协议和WebSocket的基...
                                    赞
踩
相关标签
RabbitMQ
生产者
消费者
消息分发
消息队列
cluster
rabbitmq
分布式
centos
linux
ruby