Rabbitmq 死信队列

男娘i 2023-09-27 11:16 129阅读 0赞

死信指的是无法被消费的消息。这些消息因为一些如网络超时等原因，导致无法被消费，就成了死信消息。所以为了保证这些数据不丢失，就有了死信队列，专门对死信消息进行处理。

![e5c80bac247041cdbd81c7af40df0895.png][]

--------------------

*  工具类

import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    public class RabbitMQUtils {
    
        // 创建连接mq的链接工厂对象
        private static  ConnectionFactory connectionFactory;
    
        static {
    
            connectionFactory = new ConnectionFactory();
            // 设置rabbitmq服务器所在的ip port
            connectionFactory.setHost("192.168.1.100");
            connectionFactory.setPort(5672);
            // 设置rabbitmq所在的虚拟主机
            connectionFactory.setVirtualHost("/");
            // 设置rabbitmq所在服务的用户名 密码
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
    
        }
    
        // 定义提供链接对象的方法
        public static Connection getConnection(){
    
    
                try {
                    return connectionFactory.newConnection();
                } catch (Exception e) {
                    e.printStackTrace();
                }
    
    
            return null;
        }
    
        // 关闭通道和链接的方法
        public static void closeChannelAndConnection(Channel channel, Connection connection){
            try {
                if(channel!=null) channel.close();
                if(connection!=null) connection.close();
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }

--------------------

*  生产者

import cn.my.utils.RabbitMQUtils;
    import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    import java.io.IOException;
    
    public class DeadProduct {
        //正常交换机
        public static String NORMAL_EXCHANGE = "NORMAL_EXCHANGE01";
        //正常routingkey
        public static String NORMAL_ROUTINGKEY = "normalQueue01";
    
        public static void main(String[] args) throws IOException {
            Connection connection = RabbitMQUtils.getConnection();
            Channel channel = connection.createChannel();
    
            channel.exchangeDeclare(NORMAL_EXCHANGE, "direct");
    
            AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
    
            String message = "Test dead message";
    
            for (int i = 1; i < 11; i++) {
                channel.basicPublish(NORMAL_EXCHANGE, NORMAL_ROUTINGKEY, properties, message.getBytes());
            };
    
            System.out.println("发送消息");
        }
    
    }

--------------------

*  正常队列

import cn.my.utils.RabbitMQUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.HashMap;
    
    public class DeadConsumer01 {
        //正常交换机
        public static String NORMAL_EXCHANGE = "NORMAL_EXCHANGE01";
        //死信交换机
        public static String DEAD_EXCHANGE = "DEAD_EXCHANGE01";
        //正常队列
        public static String NORMAL_QUEUE = "NORMAL_QUEUE01";
        //死信队列
        public static String DEAD_QUEUE = "DEAD_QUEUE01";
        //正常 routingkey
        public static String NORMAL_ROUTINGKEY = "normalQueue01";
        //死信routingkey
        public static String DEAD_ROUTINGKEY = "deadQueue01";
    
    
        public static void main(String[] args) throws IOException {
            Connection connection = RabbitMQUtils.getConnection();
            Channel channel = connection.createChannel();
    
            //正常交换机
            channel.exchangeDeclare(NORMAL_QUEUE,"direct");
            //死信交换机
            channel.exchangeDeclare(DEAD_EXCHANGE,"direct");
    
            //声明死信队列
            channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
    
            //设置参数
            HashMap<String, Object> arguments = new HashMap<>();
            //设置死信交换机
            arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
            //设置死信RoutingKey
            arguments.put("x-dead-letter-routing-key", DEAD_ROUTINGKEY);
            //设置正常队列长度
            arguments.put("x-max-length", 6);
            //设置过期时间,10s（一般不在这里设置，而是在生产者端配置，这样子过期时间可以由生产者随意改动）
            //arguments.put("x-message-ttl", "10000");
    
            //声明普通队列
            channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
    
            //绑定普通交换机和队列
            channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,NORMAL_ROUTINGKEY);
            //绑定死信交换机和队列
            channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,DEAD_ROUTINGKEY);
    
            //消费者成功消费时候的回调方法
            DeliverCallback deliverCallback = (s, delivery) -> {
                System.out.println("  不想接受，拒绝！");
                channel.basicReject(delivery.getEnvelope().getDeliveryTag(),false);
            };
    
            //消费者取消消费的回调方法
            CancelCallback cancelCallback = s -> System.out.println("消费者取消消息的回调方法："+s);
    
            channel.basicConsume(NORMAL_QUEUE,false,deliverCallback,cancelCallback);
    
            System.out.println("正常队列准备消费消息......");
    
        }
    
    }

--------------------

*  死信队列

import cn.my.utils.RabbitMQUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class DeadConsumer02 {
        //死信队列
        public static String DEAD_QUEUE = "DEAD_QUEUE01";
    
        public static void main(String[] args) throws IOException {
            Connection connection = RabbitMQUtils.getConnection();
            Channel channel = connection.createChannel();
    
            DeliverCallback deliverCallback = new DeliverCallback() {
                @Override
                public void handle(String s, Delivery delivery) throws IOException {
                    System.out.println("死信队列消费：  " + new String(delivery.getBody()));
                }
            };
    
            CancelCallback cancelCallback = new CancelCallback() {
                @Override
                public void handle(String s) throws IOException {
                    System.out.println("消费者取消消费的回调方法 = " + s);
                }
            };
    
            channel.basicConsume(DEAD_QUEUE,true, deliverCallback, cancelCallback);
            System.out.println("死信队列等待消费···");
        }
    }

正常队列拒绝接受消息后，死信队列才会消费消息。

![aaa393342af74441aed42788b32c1132.jpeg][]

![ab0ab4b06be842efb26f9be511b917e3.jpeg][]

[e5c80bac247041cdbd81c7af40df0895.png]: https://img-blog.csdnimg.cn/e5c80bac247041cdbd81c7af40df0895.png
[aaa393342af74441aed42788b32c1132.jpeg]: https://img-blog.csdnimg.cn/aaa393342af74441aed42788b32c1132.jpeg
[ab0ab4b06be842efb26f9be511b917e3.jpeg]: https://img-blog.csdnimg.cn/ab0ab4b06be842efb26f9be511b917e3.jpeg