十五、死信队列

1、死信的概念

先从概念解释上搞清楚这个定义，死信，顾名思义就是无法被消费的消息，字面意思可以这样理解，一般来说，producer 将消息投递到 broker 或者直接到 queue 里了，consumer 从 queue 取出消息进行消费，但某些时候由于特定的原因导致 queue 中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。

应用场景：为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息消费发生异常时，将消息投入死信队列中。还有比如说：用户在商城下单成功并点击去支付后在指定时间未支付时自动失效。

2、死信的来源

• 消息 TTL 过期：TTL 是 Time To Live 的缩写，也就是生存时间。

• 队列达到最大长度：队列满了，无法再添加数据到 mq 中。

• 消息被拒绝：(basic.reject 或 basic.nack) 并且 requeue=false。

  3、死信实战

  

  死信之 TTL

  （1）消费者1

  ①定义普通交换机、死信交换机与正常队列、死信队列

  ②声明死信交换机、普通交换机与普通队列

  ③设置正常队列和死信交换机的联系，使用argument参数

  ④绑定：普通交换机和正常队列、死信交换机和死信队列

  ⑤正常接收消息

  

  

  

  public class Consumer01 {
      //普通交换机名称
      private static final String NORMAL_EXCHANGE = "normal_exchange";
      //死信交换机名称
      private static final String DEAD_EXCHANGE = "dead_exchange";
      public static void main(String[] args) throws Exception {
          Channel channel = RabbitMqUtils.getChannel();
          //声明死信和普通交换机 类型为 direct
          channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
          channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
          //声明死信队列
          String deadQueue = "dead-queue";
          channel.queueDeclare(deadQueue, false, false, false, null);
          //死信队列绑定：队列、交换机、路由键（routingKey）
          channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
          //正常队列绑定死信队列信息
          Map params = new HashMap<>();
          //正常队列设置死信交换机 参数 key 是固定值
          params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
          //正常队列设置死信 routing-key 参数 key 是固定值
          params.put("x-dead-letter-routing-key", "lisi");
          //正常队列
          String normalQueue = "normal-queue";
          channel.queueDeclare(normalQueue, false, false, false, params);
          channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");
          System.out.println("等待接收消息........... ");
          DeliverCallback deliverCallback = (consumerTag, delivery) -> {
              String message = new String(delivery.getBody(), "UTF-8");
              System.out.println("Consumer01 接收到消息" + message);
          };
          channel.basicConsume(normalQueue, true, deliverCallback, consumerTag -> {
          });
      }
  }
  

  （2）生产者：

  ①定义普通交换机的名称

  ②设置消息的TTL，死信消息的设置

  ③正常发送消息

  

  public class Producer {
      private static final String NORMAL_EXCHANGE = "normal_exchange";
      public static void main(String[] argv) throws Exception {
          Channel channel = RabbitMqUtils.getChannel();
          channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
          //设置消息的 TTL 时间 10s
          AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
          //该信息是用作演示队列个数限制
          for (int i = 1; i < 11; i++) {
              String message = "info" + i;
              channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", properties, message.getBytes());
              System.out.println("生产者发送消息:" + message);
          }
      }
  }
  

  （3）测试效果1：

  先启动消费者1，再启动生产者

  

  注意：正常交换机的Features里面多了 DLX、DLK

  

  

  （4）正常队列和死信队列中消息数的变化对比

  

  当正常队列的消息TTL到了之后，还没有被接受，正常队列的消息会被转发到死信交换机，再到死信队列里面，这个过程很快的

  

  （5）消费者2

  接收死信队列的消息即可

  

  public class Consumer02 {
      //死信交换机名称
      private static final String DEAD_EXCHANGE = "dead_exchange";
      public static void main(String[] args) throws Exception {
          Channel channel = RabbitMqUtils.getChannel();
          //声明交换机
          channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
          //声明队列
          String deadQueue = "dead-queue";
          channel.queueDeclare(deadQueue, false, false, false, null);
          channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
          System.out.println("等待接收死信消息........... ");
          DeliverCallback deliverCallback = (consumerTag, delivery) -> {
              String message = new String(delivery.getBody(), "UTF-8");
              System.out.println("Consumer02 接收到消息" + message);
          };
          channel.basicConsume(deadQueue, true, deliverCallback, consumerTag -> {
          });
      }
  }
  

  （6）测试效果2：

  启动消费者2

  

  

  死信之最大长度

  ①生产者和消费者删除掉TTL的代码

  ②在消费者1中设置正常队列的长度限制

  

  //设置正常队列的长度限制，例如发10个，4个则为死信
  params.put("x-max-length",6);
  

  注：先把原来生成的队列删除，因为条件参数发生变化了，已不再是同一个了，否则会报错

  测试效果：

  

  

  

  死信之消息被拒

  先注释掉最大长度的代码，然后，将消息都接收完，才可以继续发消息，还有删掉之前创建的最长时间TTL和最大长度的队列，以免对测试造成影响。

  

  public class Consumer01 {
      //普通交换机名称
      private static final String NORMAL_EXCHANGE = "normal_exchange";
      //死信交换机名称
      private static final String DEAD_EXCHANGE = "dead_exchange";
      public static void main(String[] args) throws Exception {
          Channel channel = RabbitMqUtils.getChannel();
          //声明死信和普通交换机 类型为 direct
          channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
          channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
          //声明死信队列
          String deadQueue = "dead-queue";
          channel.queueDeclare(deadQueue, false, false, false, null);
          //死信队列绑定：队列、交换机、路由键（routingKey）
          channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
          
          //正常队列绑定死信队列信息
          Map params = new HashMap<>();
          //正常队列设置死信交换机 参数 key 是固定值
          params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
          //正常队列设置死信 routing-key 参数 key 是固定值
          params.put("x-dead-letter-routing-key", "lisi");
  //        //设置正常队列的长度限制，例如发10个，4个则为死信
  //        params.put("x-max-length",6);
          
          //正常队列
          String normalQueue = "normal-queue";
          channel.queueDeclare(normalQueue, false, false, false, params);
          channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");
          System.out.println("等待接收消息........... ");
          
          DeliverCallback deliverCallback = (consumerTag, delivery) -> {
              String message = new String(delivery.getBody(), "UTF-8");
              if (message.equals("info5")) {
                  System.out.println("Consumer01 接收到消息" + message + "并拒绝签收该消息");
                  //requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中
                  channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);
              } else {
                  System.out.println("Consumer01 接收到消息" + message);
                  channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
              }
          };
          //开启手动应答
          channel.basicConsume(normalQueue, false, deliverCallback, consumerTag -> {
          });
      }
  }
  

  测试效果：

  

  

  

  消息队列-RabbitMQ：死信队列 到此完结，笔者归纳、创作不易，大佬们给个3连再起飞吧

   
   手机免费网站建设  建站工具包  微信小程序模板  网站设计哪家公司好  专业网页设计培训  网站建设公司哪家做得好