rabbitmq的死信队列

这篇具有很好参考价值的文章主要介绍了rabbitmq的死信队列。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

目录

成为死信的条件 

消息TTL过期 

 队列达到最大长度

 消息被拒

延迟队列

 延迟队列使用场景

 消息设置 TTL

队列设置 TTL

 两者区别


 

producer 将消息投递到 broker 或者直接到 queue 里了, consumer queue 取出消息
进行消费,但某些时候由于特定的 原因导致 queue 中的某些消息无法被消费 ,这样的消息如果没有后续的处理,就变成了死信,有死信自然就有了死信队列。

成为死信的条件 

  1. 超过消息的存活时间(TTL):可以为消息设置一个存活时间,在该时间段之后,如果消息还未被消费或者被重新投递到其他队列,该消息将成为死信。

  2. 消息被拒绝(Reject):当消息被消费者拒绝接收时,可以选择将该消息重新投递到另一个队列或将其标记为死信。

  3. 消息达到最大重试次数:可以通过在消费者端设置重试次数限制,当消息达到一定的重试次数而仍然无法被消费时,该消息将成为死信。

  4. 队列满溢(Queue Overflow):当一个队列的消息数量已经超过队列的最大容量限制时,新消息无法进入队列,而被视为死信。

 

消息TTL过期 

生产者 

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;

public class Producer {
      private static final String NORMAL_EXCHANGE = "normal_exchange";

      public static void main(String[] argv) throws Exception {
      try (Channel channel = RabbitMqUtils.getChannel()) {
      channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        //设置消息的 TTL 时间
        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);
        }
      }
   }
}

 消费者1

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.util.HashMap;
import java.util.Map;

public class Consumer01 {
       //普通交换机名称
       private static final String NORMAL_EXCHANGE = "normal_exchange";
       //死信交换机名称
       private static final String DEAD_EXCHANGE = "dead_exchange";
       public static void main(String[] argv) 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<String, Object> 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

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

public class Consumer02 {
      private static final String DEAD_EXCHANGE = "dead_exchange";
      public static void main(String[] argv) 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 -> {
      });
      }
}

关闭消费者1,模拟出故障

rabbitmq的死信队列,rabbitmq,java-rabbitmq,rabbitmq,java

正常队列

rabbitmq的死信队列,rabbitmq,java-rabbitmq,rabbitmq,java 消息超时进入死信队列

rabbitmq的死信队列,rabbitmq,java-rabbitmq,rabbitmq,java

 死信队列接收信息

rabbitmq的死信队列,rabbitmq,java-rabbitmq,rabbitmq,java

 队列达到最大长度

生产者

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;

public class Producer {
     private static final String NORMAL_EXCHANGE = "normal_exchange";
     public static void main(String[] argv) throws Exception {
         try (Channel channel = RabbitMqUtils.getChannel()) {
         channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
         //该信息是用作演示队列个数限制
         for (int i = 1; i <11 ; i++) {
         String message="info"+i;
         channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null, message.getBytes());
         System.out.println("生产者发送消息:"+message);
         }
         }
     }
}

 消费者1

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.util.HashMap;
import java.util.Map;

public class Consumer01 {
       //普通交换机名称
       private static final String NORMAL_EXCHANGE = "normal_exchange";
       //死信交换机名称
       private static final String DEAD_EXCHANGE = "dead_exchange";

       public static void main(String[] argv) 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<String, Object> params = new HashMap<>();
           //正常队列设置死信交换机 参数 key 是固定值
           params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
           //正常队列设置死信 routing-key 参数 key 是固定值
           params.put("x-dead-letter-routing-key", "lisi");
           // 设置正常队列长度的限制
           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");
              System.out.println("Consumer01 接收到消息"+message);
           };
           channel.basicConsume(normalQueue, true, deliverCallback, consumerTag -> {
           });
       }
}

 消费者2

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

public class Consumer02 {
      private static final String DEAD_EXCHANGE = "dead_exchange";
      public static void main(String[] argv) 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 -> {
            });
      }
}

关闭消费者1,模拟接收不到信息

rabbitmq的死信队列,rabbitmq,java-rabbitmq,rabbitmq,java

 死信队列消费了四个消息

rabbitmq的死信队列,rabbitmq,java-rabbitmq,rabbitmq,java

 分析:生产者产生10个消息,正常队列只能接受6个消息,多的消息便被转移到死信队列去了

rabbitmq的死信队列,rabbitmq,java-rabbitmq,rabbitmq,java

 消息被拒

生产者

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;

public class Producer {
     private static final String NORMAL_EXCHANGE = "normal_exchange";
     public static void main(String[] argv) throws Exception {
         try (Channel channel = RabbitMqUtils.getChannel()) {
         channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
         //该信息是用作演示队列个数限制
         for (int i = 1; i <11 ; i++) {
         String message="info"+i;
         channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null, message.getBytes());
         System.out.println("生产者发送消息:"+message);
         }
         }
     }
}

消费者1


import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.util.HashMap;
import java.util.Map;

public class Consumer01 {
     //普通交换机名称
     private static final String NORMAL_EXCHANGE = "normal_exchange";
     //死信交换机名称
     private static final String DEAD_EXCHANGE = "dead_exchange";
     public static void main(String[] argv) 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<String, Object> 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");
             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);
             }
         };
         boolean autoAck = false;
         channel.basicConsume(normalQueue, autoAck, deliverCallback, consumerTag -> {
         });
     }
}

 消费者2

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

public class Consumer02 {
      private static final String DEAD_EXCHANGE = "dead_exchange";
      public static void main(String[] argv) 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 -> {
            });
      }
}

 结果

消费者1,拒绝接受消息info5,info5进入死信队列

rabbitmq的死信队列,rabbitmq,java-rabbitmq,rabbitmq,java

 消费者2,死信队列接受到info5

rabbitmq的死信队列,rabbitmq,java-rabbitmq,rabbitmq,java

延迟队列

延时队列 , 队列内部是有序的,最重要的特性就体现在它的延时属性上,延时队列中的元素是希望
在指定时间到了以后或之前取出和处理,简单来说,延时队列就是用来存放需要在指定时间被处理的 元素的队列。

 延迟队列使用场景

1. 订单在十分钟之内未支付则自动取消
2. 新创建的店铺,如果在十天内都没有上传过商品,则自动发送消息提醒。
3. 用户注册成功后,如果三天内没有登陆则进行短信提醒。
4. 用户发起退款,如果三天内没有得到处理则通知相关运营人员。
5. 预定会议后,需要在预定的时间点前十分钟通知各个与会人员参加会议

 消息设置 TTL

 

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;

public class Producer {
      private static final String NORMAL_EXCHANGE = "normal_exchange";

      public static void main(String[] argv) throws Exception {
      try (Channel channel = RabbitMqUtils.getChannel()) {
      channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        //设置消息的 TTL 时间
        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);
        }
      }
   }
}

队列设置 TTL



import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.util.HashMap;
import java.util.Map;

public class Consumer01 {
     //普通交换机名称
     private static final String NORMAL_EXCHANGE = "normal_exchange";
     //死信交换机名称
     private static final String DEAD_EXCHANGE = "dead_exchange";
     public static void main(String[] argv) 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<String, Object> params = new HashMap<>();
         //正常队列设置死信交换机 参数 key 是固定值
         params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
         //正常队列设置死信 routing-key 参数 key 是固定值
         params.put("x-dead-letter-routing-key", "lisi");
         // 设置 TTL 值为 5000 毫秒(5 秒)
         params.put("x-message-ttl", 5000);
         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);
             }
         };
         boolean autoAck = false;
         channel.basicConsume(normalQueue, autoAck, deliverCallback, consumerTag -> {
         });
     }
}

 两者区别

如果设置了队列的 TTL 属性,那么一旦消息过期,就会被队列丢弃 ( 如果配置了死信队列被丢到死信队列中) ,而第二种方式,消息即使过期,也不一定会被马上丢弃,因为 消息是否过期是在即将投递到消费者 之前判定的 ,如果当前队列有严重的消息积压情况,则已过期的消息也许还能存活较长时间;另外,还需 要注意的一点是,如果不设置 TTL ,表示消息永远不会过期,如果将 TTL 设置为 0 ,则表示除非此时可以 直接投递该消息到消费者,否则该消息将会被丢弃。

 文章来源地址https://www.toymoban.com/news/detail-657260.html

到了这里,关于rabbitmq的死信队列的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 【RabbitMQ笔记10】消息队列RabbitMQ之死信队列的介绍

    这篇文章,主要介绍消息队列RabbitMQ之死信队列。 目录 一、RabbitMQ死信队列 1.1、什么是死信队列 1.2、设置过期时间TTL 1.3、配置死信交换机和死信队列(代码配置) (1)设置队列过期时间 (2)设置单条消息过期时间 (3)队列设置死信交换机 (4)配置的基本思路 1.4、配置

    2024年02月16日
    浏览(82)
  • 【RabbitMQ学习日记】——死信队列与延迟队列

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

    2024年02月06日
    浏览(56)
  • 【RabbitMQ教程】第五章 —— RabbitMQ - 死信队列

                                                                       💧 【 R a b b i t M Q 教 程 】 第 五 章 — — R a b b i t M Q − 死 信 队 列 color{#FF1493}{【RabbitMQ教程】第五章 —— RabbitMQ - 死信队列} 【 R a b b i t M Q 教 程 】 第 五 章 — — R a

    2024年02月09日
    浏览(40)
  • RabbitMQ:死信队列

    📃个人主页:不断前进的皮卡丘 🌞博客描述:梦想也许遥不可及,但重要的是追梦的过程,用博客记录自己的成长,记录自己一步一步向上攀登的印记 🔥个人专栏:消息中间件 队列中不能被消费的消息称为死信队列 有时候因为特殊原因,可能导致队列中的某些信息无法被消费

    2024年02月02日
    浏览(44)
  • RabbitMQ 死信队列实现

    死信队列使用注解实现 报错: 可以使用注解的方式来绑定 死信队列,但是还是会报上面的错误,继续修改 参数试试 java - How to set x-dead-letter-exchange in Rabbit? - Stack Overflow 但是使用注解绑定的话好像又不生效了,问题原因,tmd将死信参数绑到交换机上了,c 修改代码  至于问题

    2024年02月02日
    浏览(34)
  • rabbitmq的死信队列

    目录 成为死信的条件  消息TTL过期   队列达到最大长度  消息被拒 延迟队列  延迟队列使用场景  消息设置 TTL 队列设置 TTL  两者区别   producer 将消息投递到 broker 或者直接到 queue 里了, consumer 从 queue 取出消息 进行消费,但某些时候由于特定的 原因导致 queue 中的某些消

    2024年02月12日
    浏览(36)
  • RabbitMQ: 死信队列

    其实就是一个普通的队列,绑定号私信交换机,不给ttl,给上匹配的路由,等待交换机发送消息。 1.在消费者里的RabbitMQConfig配置类里,创建队列,给它加参数 第四个参数,就是放入这个队列,的一些属性参数 也就是这两个位置 对应Java代码里好像少个参数,排他性,是指,

    2024年02月09日
    浏览(44)
  • RabbitMQ进阶——死信队列

    在消息队列中,执行异步任务时,通常是将消息生产者发布的消息存储在队列中,由消费者从队列中获取并处理这些消息。但是,在某些情况下,消息可能无法正常地被处理和消耗,例如:格式错误、设备故障等,这些未成功处理的消息就被称为“死信”。 为了避免这些未成

    2024年04月13日
    浏览(32)
  • rabbitMQ引入死信队列

            指的是,从队列当中取出来的消息,到达消费方后,因为某些原因导致消息并没有被正常消费掉,这些没有被后续处理的消息就是“死信”,而保存死信的队列,就是死信队列。         为了保证订单业务的消息数据不丢失,需要使用死信队列机制,在消息消费发生

    2024年02月02日
    浏览(45)
  • RabbitMQ——死信队列

    死信队列(Dead Letter Queue,DLQ)是 RabbitMQ 中的一种重要特性,用于处理无法被消费的消息,防止消息丢失。 死信的来源 在消息队列中,当消息满足一定条件而无法被正常消费时,这些消息会被发送到死信队列。满足条件的情况包括但不限于: 消息被拒绝( basic.reject 或 bas

    2024年03月14日
    浏览(55)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包