devbox
花生_TL007
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

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

作者：花生_TL007 | 2024-05-14 19:07:04

赞

踩

rabbmitmq 延迟队列 及死信

目录

一、介绍

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

2、应用场景

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

4、死信消息来源

二、案例实践

1、案例一

2、案例二（消息接收确认 ）

3、总结

一、介绍

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

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

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

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

例如:

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

2、应用场景

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

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

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

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

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

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

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

4、死信消息来源

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

二、案例实践

1、案例一

生产者

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

  1.  //    =============死信队列===============
  2.     // 定义QueueA Bean，返回QueueA实例
  3.     @Bean
  4.     public Queue QueueA() {
  5.         Map<String, Object> config = new HashMap<>();
  6.         //message在该队列queue的存活时间最大为5秒
  7.         config.put("x-message-ttl", 5000);
  8.         //x-dead-letter-exchange参数是设置该队列的死信交换器（DLX）
  9.         config.put("x-dead-letter-exchange", "deadExchange");
  10.         //x-dead-letter-routing-key参数是给这个DLX指定路由键
  11.         config.put("x-dead-letter-routing-key", "deadQueue");
  12.         // 返回QueueA实例
  13.         return new Queue("QueueA", true, true, false, config);
  14.     }
  15.  
  16.     // 定义DirectExchangeA Bean，返回DirectExchangeA实例
  17.     @Bean
  18.     public DirectExchange DirectExchangeA() {
  19.         // 返回DirectExchangeA实例
  20.         return new DirectExchange("DirectExchangeA");
  21.     }
  22.  
  23.     // 定义bindingA Bean，将QueueA绑定到DirectExchangeA，并设置路由键为A.A
  24.     @Bean
  25.     public Binding bindingA() {
  26.         // 将QueueA绑定到DirectExchangeA
  27.         return BindingBuilder
  28.                 // 设置路由键为A.A
  29.                 .bind(QueueA())
  30.                 .to(DirectExchangeA())
  31.                 .with("A.A");
  32.     }
  33.  
  34.     // 定义QueueB Bean，返回QueueB实例
  35.     @Bean
  36.     public Queue QueueB() {
  37.         // 返回QueueB实例
  38.         return new Queue("QueueB");
  39.     }
  40.  
  41.     // 定义DirectExchangeB Bean，返回DirectExchangeB实例
  42.     @Bean
  43.     public DirectExchange DirectExchangeB() {
  44.         // 返回DirectExchangeB实例
  45.         return new DirectExchange("DirectExchangeB");
  46.     }
  47.  
  48.     // 定义bindingB Bean，将QueueB绑定到DirectExchangeB，并设置路由键为B.B
  49.     @Bean
  50.     public Binding bindingB() {
  51.         // 将QueueB绑定到DirectExchangeB
  52.         return BindingBuilder
  53.                 // 设置路由键为B.B
  54.                 .bind(QueueB())
  55.                 .to(DirectExchangeB())
  56.                 .with("B.B");
  57.     }

消费者

在消费者里面添加QueueA和QueueB 

QueueA

  1. import lombok.extern.slf4j.Slf4j;
  2. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  3. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  4. import org.springframework.stereotype.Component;
  5.  
  6. @Component
  7. @SuppressWarnings("all")
  8. @Slf4j
  9. @RabbitListener(queues = "QueueA")
  10. public class ReceiverQA {
  11.  
  12.         // 接收directExchange01交换机中Queue02队列消息的方法
  13.         @RabbitHandler
  14.         public void Queue02(String msg) {
  15.             log.warn("QueueA，接收到信息:" + msg);
  16.         }
  17.  
  18. }

QueueB

  1. import lombok.extern.slf4j.Slf4j;
  2. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  3. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  4. import org.springframework.stereotype.Component;
  5.  
  6. @Component
  7. @SuppressWarnings("all")
  8. @Slf4j
  9. @RabbitListener(queues = "QueueB")
  10. public class ReceiverQB {
  11.  
  12.     // 接收directExchange01交换机中Queue02队列消息的方法
  13.     @RabbitHandler
  14.     public void Queue02(String msg) {
  15.         log.warn("QueueB，接收到信息:" + msg);
  16.     }
  17.  
  18. }

编写Controller层

  1.     @RequestMapping("send5")
  2.     public String send5() {
  3.         rabbitTemplate.convertAndSend("DirectExchangeA", "A.A", "Hello,A");
  4.         return "
    声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/花生_TL007/article/detail/569682
    推荐阅读
    相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。