RabbitMQ:死信队列

慈云数据 1年前 (2024-03-15) 技术支持 52 0

✨ RabbitMQ:死信队列

  • 1.死信队列
    • 1.1死信队列基本介绍
    • 1.2消息成为死信的三种情况
    • 1.3死信队列结构图
    • 1.4死信的处理方式
    • 2.TTL消息过期时间
      • 2.1基本介绍
      • 2.2生产者
      • 2.3消费者1
      • 2.4消费者2
      • 2.5设置TTL的两种方式
        • 2.5.1队列设置TTL
        • 2.5.2消息设置TTL
        • 2.5.3区别

          📃个人主页:不断前进的皮卡丘

          🌞博客描述:梦想也许遥不可及,但重要的是追梦的过程,用博客记录自己的成长,记录自己一步一步向上攀登的印记

          🔥个人专栏:消息中间件

          1.死信队列

          1.1死信队列基本介绍

          • 队列中不能被消费的消息称为死信队列
          • 有时候因为特殊原因,可能导致队列中的某些信息无法被消费,而队列中这些不能被消费的消息在后期没有进行处理,就会变成死信队列,死信队列中的消息称为死信。
          • 应用场景:未来保证订单业务的消息数据不丢失,我们需要使用到RabbitMQ的死信队列机制,当消息消费发生异常的时候,我们就把消息投入到死信队列中,比如说用户买东西,下单成功后去支付,但是没有在指定时间支付的时候就会自动失效。
          • 死信队列,英文缩写:DLX 。DeadLetter Exchange(死信交换机),当消息成为Dead message后,可以被重新发送到另一个交换机,这个交换机就是DLX。
          • 当消息在一个队列中变成死信后,它能被重新发布到另一个Exchange中,这个Exchange就是DLX
          • 在这里插入图片描述

            1.2消息成为死信的三种情况

            1. 队列消息数量到达限制;比如队列最大只能存储10条消息,而发了11条消息,根据先进先出,最先发的消息会进入死信队列。
            2. 消费者拒接消费消息,basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false。
            3. 原队列存在消息过期设置,消息到达超时时间未被消费。

            1.3死信队列结构图

            通常情况下,消费者是能正常消费消息的,但是出现上面说的三种情况之一,就无法正常消费信息,消息就会进入死信交换机,死信交换机会和死信队列进行绑定,最后由其他消费者来消费死信消息。

            在这里插入图片描述

            1.4死信的处理方式

            死信的产生既然不可避免,那么就需要从实际的业务角度和场景出发,对这些死信进行后续的处理,常见的处理方式大致有下面几种,

            ① 丢弃,如果不是很重要,可以选择丢弃

            ② 记录死信入库,然后做后续的业务分析或处理

            ③ 通过死信队列,由负责监听死信的应用程序进行处理

            综合来看,更常用的做法是第三种,即通过死信队列,将产生的死信通过程序的配置路由到指定的死信队列,然后应用监听死信队列,对接收到的死信做后续的处理。

            队列绑定死信交换机:

            给队列设置参数:x-dead-letter-exchange 和x-dead-letter-routing-key

            在这里插入图片描述

            2.TTL消息过期时间

            2.1基本介绍

            当消息到达存活时间后,还没有被消费,就会被自动清除。RabbitMQ可以对消息或者队列设置过期时间,队列中的消息过期是成为死信队列的三种原因之一。

            在这里插入图片描述

            2.2生产者

            public class Producer {
                //正常交换机
                public static final String NORMAL_EXCHANGE = "normal_exchange";
                //正常队列
                public static final String NORMAL_QUEUE = "normal_queue";
                public static void main(String[] args) {
                    try {
                        Channel channel = ConnectUtil.getChannel();
                        //声明交换机
                        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
                        //声明队列
                        //channel.queueDeclare(NORMAL_QUEUE, true, false, false, null);
                        //把正常交换机和正常队列进行绑定
                        //channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "tom");
                        //设置过期时间
                        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
                        //发送消息
                        for (int i = 0; i  
            

            2.3消费者1

            public class Consumer1 {
                //定义交换机(正常交换机,死信交换机)
                public static final String NORMAL_EXCHANGE = "normal_exchange";
                public static final String DEAD_EXCHANGE = "dead_exchange";
                //定义队列(正常队列,死信队列)
                public static final String NORMAL_QUEUE = "normal_queue";
                public static final String DEAD_QUEUE = "dead_queue";
                public static void main(String[] args) {
                    try {
                        //创建信道对象
                        Channel channel = ConnectUtil.getChannel();
                        //声明交换机
                        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
                        channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);
                        //设置正常队列和死信队列进行绑定,key固定不可以改变
                       Map map = new HashMap();
                        map.put("x-dead-letter-exchange",DEAD_EXCHANGE);
                        map.put("x-dead-letter-routing-key", "jack");
                        //声明正常队列
                        channel.queueDeclare(NORMAL_QUEUE,false,false,false,map);
                        //正常交换机绑定正常队列
                        channel.queueBind(NORMAL_QUEUE,NORMAL_QUEUE,"tom");
                        //声明死信队列
                        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
                        //死信交换机绑定死信队列
                        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"jack");
                        //消费消息
                        DefaultConsumer consumer = new DefaultConsumer(channel) {
                            /**
                             * 消费回调函数,当收到消息以后,会自动执行这个方法
                             * @param consumerTag 消费者标识
                             * @param envelope    消息包的内容(比如交换机,路由key,消息id等)
                             * @param properties   属性信息
                             * @param body         消息数据
                             * @throws IOException
                             */
                            @Override
                            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                                System.out.println("消息者1接受到的消息:" + new String(body, "UTF-8"));
                            }
                        };
                        //监听消息(队列名称,是否自动确认消息,消费对象)
                        channel.basicConsume(NORMAL_QUEUE, true, consumer);
                    } catch (IOException e) {
                        e.printStackTrace();
                    } catch (TimeoutException e) {
                        e.printStackTrace();
                    }
                }
            }
            

            2.4消费者2

            public class Consumer2 {
                //定义交换机(死信交换机)
                public static final String DEAD_EXCHANGE = "dead_exchange";
                //定义队列(死信队列)
                public static final String DEAD_QUEUE = "dead_queue";
                public static void main(String[] args) {
                    try {
                        //创建信道对象
                        Channel channel = ConnectUtil.getChannel();
                        //声明死信队列
                        channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
                        //死信交换机绑定死信队列
                        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "jack");
                        //消费消息
                        DefaultConsumer consumer = new DefaultConsumer(channel) {
                            /**
                             * 消费回调函数,当收到消息以后,会自动执行这个方法
                             * @param consumerTag 消费者标识
                             * @param envelope    消息包的内容(比如交换机,路由key,消息id等)
                             * @param properties   属性信息
                             * @param body         消息数据
                             * @throws IOException
                             */
                            @Override
                            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                                System.out.println("消息者2接受到的消息:" + new String(body, "UTF-8"));
                            }
                        };
                        //监听消息(队列名称,是否自动确认消息,消费对象)
                        channel.basicConsume(DEAD_QUEUE, true, consumer);
                    } catch (IOException e) {
                        e.printStackTrace();
                    } catch (TimeoutException e) {
                        e.printStackTrace();
                    }
                }
            }
            

            2.5设置TTL的两种方式

            2.5.1队列设置TTL

            在创建队列的时候设置队列的x-message-ttl属性,例如:

              Map map = new HashMap();
            //设置队列有效期为10秒
            map.put("x-message-ttl",10000);
            channel.queueDeclare(queueName,durable,exclusive,autoDelete,map);
            

            2.5.2消息设置TTL

            对每条消息设置TTL

              AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
             channel.basicPublish(exchangeName,routingKey,mandatory,properties,"msg body".getBytes());
            

            2.5.3区别

            如果设置了队列的TTL属性,那么一旦消息过期,就会被队列丢弃。

            如果是消息设置了TTL属性,那么即使消息过期,也不一定会被马上丢弃,因为消息是否过期是在即将投递到消费者之前判定的,如果当前队列有严重的消息积压情况,那么已经过期的消息也许还能存活较长时间。

            如果我们没有设置TTL,就表示消息永远不会过期,如果TTL设置为0,则表示除非此时可以直接投递到消费者,否则该消息会被丢弃。

微信扫一扫加客服

微信扫一扫加客服