RabbitMQ深入 —— 死信队列_rabbitmq死信队列

作者：小蓝xlanll | 2024-05-21 15:11:57

踩

rabbitmq死信队列

前言

前面荔枝梳理了RabbitMQ中的普通队列、交换机以及相关的知识，在这篇文章中荔枝将会梳理RabbitMQ的一个重要的队列 —— 死信队列，主要了解消息流转到死信队列的三种的方式以及相应的实现demo。希望能帮助到有需要的小伙伴~~~

文章目录

死信队列

1 基本概念

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

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

死信具有一定的延迟性，它可以作为延迟消息来处理。

死信出现的原因：

消息TTL过期
队列达到最大长度（队列满了，无法再添加数据到mq中）
消息被拒绝(basic.reject或basic.nack)并且requeue=false.I

2 设置消息时间TTL过期的死信队列

首先我们在消费者Consumer1中声明普通交换机、死信交换机、普通队列和死信队列之间的关系，同时在声明之后令Consumer1拒收消息，在RabbitMQ中观察消息生产者发出消息的流转情况。

设置死信队列的消费者1

在死信队列中我们设置了普通交换机、死信交换机、普通队列和死信队列。同时在正常队列中通过channel信道对象中的queueDeclare方法中的一个Map类型的参数，设置了死信交换机和普通交换机之间的关系，配置好TTL、RoutingKey并声明其死信交换机。


package com.crj.rabbitmq.deadQueue;
 
import com.crj.rabbitmq.utils.RabbitMqUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
 
import java.util.HashMap;
import java.util.Map;
 
/**
 * 死信队列
 * 消费者1：需要声明死信队列和普通队列
 */
public class Consumer {
    //普通交换机名称
    public static final String NORMAL_EXCHANGE = "normal";
    //死信交换机名称
    public static final String DEAD_EXCHANGE = "dead";
    //普通队列的名称
    public static final String NORMAL_QUEUE = "normalQueue";
    //死信队列的名称
    public static final String DEAD_QUEUE = "deadQueue";
 
    public static void main(String[] args) throws Exception {
        //声明通道
        Channel channel = RabbitMqUtil.getChannel();
        //声明普通交换机和死信交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);
        /**
         * 声明普通队列和死信队列
         */
        //创建一个hashmap对象来配置连接死信队列的参数
        Map<String, Object> arguments = new HashMap<>();
        //设置过期时间
        arguments.put("x-message-ttl",10000);
        //正常队列设置死信交换机
        arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        //设置死信RoutingKey
        arguments.put("x-dead-letter-routing-key","dead1");
        //声明普通队列
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
        //死信队列
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
 
        //绑定队列和交换机
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"normal");
        channel.queueBind(DEAD_QUEUE,DEAD_QUEUE,"dead");
 
        //接收消息
        DeliverCallback deliverCallback = (consumerTag, message)->{
            System.out.println("Consumer1接收到的信息："+new String(message.getBody(),"UTF-8"));
            System.out.println("接收队列："+DEAD_QUEUE+"接收键："+message.getEnvelope().getRoutingKey());
        };
        //消费者开始消费消息
        channel.basicConsume(DEAD_QUEUE,true,deliverCallback,(consumerTag)->{});
    }
}

需要注意的是，这里在正常队列中设置过期时间TTL一般不太常用，我们通常会在publish处设置消息的TTL，因此这里arguments对象有关 "x-message-ttl" 参数的配置可以注释掉。

实际处理消息的消费者2

在处理死信队列消息的消费者处，我们只需要设置消费者接收消息是来自死信队列即可。


package com.crj.rabbitmq.deadQueue;
 
import com.crj.rabbitmq.utils.RabbitMqUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
 
import java.util.HashMap;
import java.util.Map;
 
/**
 * 死信队列
 * 消费者1：需要声明死信队列和普通队列
 */
public class Consumer2 {
    //死信队列的名称
    public static final String DEAD_QUEUE = "deadQueue";
 
    public static void main(String[] args) throws Exception {
        //声明通道
        Channel channel = RabbitMqUtil.getChannel();
        System.out.println("等待接收消息");
        //接收消息
        DeliverCallback deliverCallback = (consumerTag, message)->{
            System.out.println("Consumer2接收到的信息："+new String(message.getBody(),"UTF-8"));
            System.out.println("接收队列："+DEAD_QUEUE+"接收键："+message.getEnvelope().getRoutingKey());
        };
        //消费者开始消费消息
        channel.basicConsume(DEAD_QUEUE,true,deliverCallback,(consumerTag)->{});
    }
}

生产者

在这里我们借助AMQP. BasicProperties对象的build方法来设置相应的死信TTL。


package com.crj.rabbitmq.deadQueue;
 
import com.crj.rabbitmq.utils.RabbitMqUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
 
public class Publish {
    public static final String NORMAL_EXCHANGE = "normal";
    public static final String NORMAL_QUEUE = "normalQueue";
 
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtil.getChannel();
        //在Consumer已经声明过交换机了，所以在这里不能声明
        //死信消息，设置TTL
        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
 
        for (int i = 0; i < 11; i++) {
            String message = "info"+i;
            channel.basicPublish(NORMAL_EXCHANGE,"normal",properties,message.getBytes());
        }
    }
}

未运行Consumer2前我们看到普通队列在我们设置的TTL：10s之后将消息流转到死信队列中。

最后启动Consumer2后确实也收到了死信队列中的消息

3 队列达到最大长度发生死信

在这一部分中我们需要注释掉之前在生产者中设置的消息的TTL，同时在消费者1中开启正常队列的最大消息堆积容量。

arguments.put("x-max-length",6);

这样子我们就可以模拟队列达到最大长度后产生死信的情况了。

4 消息被拒引发死信

要想开启消费者拒收消息的功能，首先需要在消息接收的basicConsumer方法中关闭自动应答，同时自行设置手动应答的逻辑。在下面接收消息的回调函数中，在basicAck中设置应答，在basicReject实现消息拒收。


package com.crj.rabbitmq.deadQueue;
 
import com.crj.rabbitmq.utils.RabbitMqUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
 
import java.util.HashMap;
import java.util.Map;
 
/**
 * 死信队列
 * 消费者1：需要声明死信队列和普通队列
 */
public class Consumer {
    //普通交换机名称
    public static final String NORMAL_EXCHANGE = "normal";
    //死信交换机名称
    public static final String DEAD_EXCHANGE = "dead";
    //普通队列的名称
    public static final String NORMAL_QUEUE = "normalQueue";
    //死信队列的名称
    public static final String DEAD_QUEUE = "deadQueue";
 
    public static void main(String[] args) throws Exception {
        //声明通道
        Channel channel = RabbitMqUtil.getChannel();
        //声明普通交换机和死信交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);
        /**
         * 声明普通队列和死信队列
         */
        //创建一个hashmap对象来配置连接死信队列的参数
        Map<String, Object> arguments = new HashMap<>();
 
        //正常队列设置死信交换机
        arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        //设置死信RoutingKey
        arguments.put("x-dead-letter-routing-key","dead1");
 
        //声明普通队列
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
        //死信队列
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
 
        //绑定队列和交换机
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"normal");
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"dead1");
 
        System.out.println("等待接收消息》》》》》》》》》》》");
        //接收消息
        DeliverCallback deliverCallback = (consumerTag, message)->{
            String msg = new String(message.getBody(),"UTF-8");
            if (msg.equals("info5")){
                System.out.println("Consumer1接收的消息是："+msg+"：此消息是被拒绝的");
                //这里第二个参数设置了是否要将拒收的消息塞回原队列
                channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
            }else {
                System.out.println("Consumer1接收到的信息："+new String(message.getBody(),"UTF-8"));
                //成功应答，这里设置不批量操作
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            }
        };
        //开启手动应答
        //消费者开始消费消息
        channel.basicConsume(DEAD_QUEUE,false,deliverCallback,(consumerTag)->{});
    }
}

总结

时间过期、消息被拒、队列容量限制这三个机制会引发消息被转发死信队列，那么死信队列除了在这三种情况下继续保存消息之外，还有什么作用呢？下一篇文章荔枝会梳理延时队列，相信看完下一篇文章大家能有所收获~

今朝已然成为过去，明日依然向往未来！我是荔枝，在技术成长之路上与您相伴~~~

如果博文对您有帮助的话，可以给荔枝一键三连嘿，您的支持和鼓励是荔枝最大的动力！

如果博文内容有误，也欢迎各位大佬在下方评论区批评指正！！！

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小蓝xlanll/article/detail/603232