RabbitMQ的死信队列和延迟队列

RabbitMQ的死信队列和延迟队列

• 死信队列
• • 什么是死信队列
  • 什么条件才会转移到死信队列呢
  • 怎么配置死信队列
  • 死信队列有什么用呢？
  • 延迟队列
  • • 什么是延迟队列呢
    • 惰性队列
    • • 什么是惰性队列
      • 怎么设置
      • 有什么优点/缺点
      • 总结

        死信队列

        什么是死信队列

        简单来说，就是普通队列中的消息符合某个条件时，会交由另一个交换机转移到另一个队列，这个队列就是死信队列，负责转移的交换机就是死信交换机。

        什么条件才会转移到死信队列呢

        • 队列消息长度到达限制
        • 消费者拒接消费信息
        • 消息超时未被消费，分为两种，一种是消息自身设置的超时时间，另一种则是队列的超时时间。

          怎么配置死信队列

          package com.xiaow.fileserver.config;
          import org.springframework.amqp.core.*;
          import org.springframework.amqp.rabbit.connection.ConnectionFactory;
          import org.springframework.amqp.rabbit.core.RabbitAdmin;
          import org.springframework.context.annotation.Bean;
          import org.springframework.context.annotation.Configuration;
          @Configuration
          public class RabbitMQConfig {
              @Bean
              public Queue normQueue() {
                  return QueueBuilder.durable("normal_queue")
          //              设置死信交换机
                          .deadLetterExchange("DeadExchange")
          //              设置死信队列路由键
                          .deadLetterRoutingKey("timeout_routing")
                          .build();
              }
              //  正常交换机
              @Bean
              DirectExchange normalExchange() {
                  //  return new DirectExchange("TestDirectExchange",true,true);
                  return new DirectExchange("NormalExchange", true, false);
              }
              //绑定常规队列
              @Bean
              Binding bindingNormal() {
                  return BindingBuilder.bind(normQueue()).to(normalExchange()).with("normal_routing");
              }
              //  死信队列
              @Bean
              public Queue dQueue() {
                  return QueueBuilder.durable("d_queue")
                          .build();
              }
              //  死信交换机
              @Bean
              DirectExchange DeadExchange() {
                  //  return new DirectExchange("TestDirectExchange",true,true);
                  return new DirectExchange("DeadExchange", true, false);
              }
              //  死信队列与死信交换机绑定
              @Bean
              Binding bindingD() {
                  return BindingBuilder.bind(dQueue()).to(DeadExchange()).with("timeout_routing");
              }
          }
          

          死信队列有什么用呢？

          以淘宝下订单为例，用户下订单时我们可以发送一个消息到队列中，并设置超时时间，当超过了超时时间用户仍未付款，则删除该订单。

          模拟一下，这里我们写一下发送消息的接口和接收消息的Recevier

          发送消息的接口

           @GetMapping("testmq")
              public Result testmq(){
                  String orderId = String.valueOf(UUID.randomUUID());
                  String messageData = "下订单!";
                  String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
                  Map map=new HashMap();
                  map.put("orderId",orderId);
                  map.put("messageData",messageData);
                  map.put("createTime",createTime);
                  rabbitTemplate.convertAndSend("NormalExchange", "normal_routing", map,message -> {
                      MessageProperties messageProperties = message.getMessageProperties();
                      // 设置过期时间
                      messageProperties.setExpiration("5000");
                      return message;
                  });
                  return Result.succ("ok");
              }
          

          接收消息的Recevier

          @Component
          @RabbitListener(queues = "d_queue")
          public class DReceiver {
              @RabbitHandler
              public void process(Map testMessage) {
                  System.out.println("订单号:" + testMessage.get("orderId") + ", 未付款,自动取消订单");
              }
          }
          

          然后你就可以看到

          

          延迟队列

          什么是延迟队列呢

          顾名思义，也就是说消息到了队列之后不会立刻被处理，会等到指定时间之后才进行处理。

          是不是有点眼熟？

          咱们刚才利用死信队列做的事不就是这样吗！

          当消息到达5秒之后，才进行删除订单操作。

          死信队列结合过期时间也就实现了延迟队列。

          惰性队列

          什么是惰性队列

          遇到消息堆积问题，会导致内存占用过大，因此有了惰性队列。

          • 接收到消息后直接存入磁盘而非内存
          • 消费者要消费消息时才会从磁盘中读取并加载到内存
          • 支持数百万条的消息存储

            怎么设置

              @Bean
                public Queue lazyQueue() {
                    return QueueBuilder.durable("lazyQueue")
                            .lazy()
                            .build();
                }
            

            有什么优点/缺点

            存储到磁盘，第一个优点就是信息存储量更大了。但磁盘和内存之间的速度差异导致其效率有所降低。

            总结

            简单记录一下，欢迎大家指正。