Java初识RabbitMQ一死信队列

落日映苍穹つ 2023-07-05 05:28 144阅读 0赞

推荐：[消息中间件汇总][Link 1]

## Java初识RabbitMQ一死信队列 ##

**什么是死信队列**

死信队列`DLX`（dead-letter-exchange），利用`DLX`，当消息在一个队列中变成死信 (dead message) 之后，它能被重新投递到另一个exchange上，这个exchange就是`DLX`，似乎叫死信交换机更加贴切，当死信投递到这个exchange后，我们也可以用一个queue来绑定该exchange，该exchange就可以根据路由规则把这个死信路由到这个queue了，同样可以创建消费端去进行消费，以便对死信进行相应的处理。

**消息变成死信的情况**

1.  消息被拒绝(`basic.reject` / `basic.nack`)，并且`requeue = false`（不重回队列）。
2.  消息`TTL`过期。
3.  队列达到最大长度。

**死信处理过程**

`DLX`也是一个正常的exchange，和一般的exchange没有区别，不过，需要在队列上将它设置成`DLX`，在`arguments`中添加`key`为`x-dead-letter-exchange`，`value`为该交换机名称的`key-value`即可，代码如下：

// 设置死信队列
            Map<String , Object> arguments = new HashMap<String , Object>();
            arguments.put("x-dead-letter-exchange" , dlxExchange);
            channel.exchangeDeclare(dlxExchange , "topic" , true , false , null);
            channel.queueDeclare(dlxQueueName , true , false , false , null);
            channel.queueBind(dlxQueueName , dlxExchange , dlxRoutingKey);
    
            // 创建Queue
            channel.queueDeclare(queueName , true , false , false , arguments);

运行上面代码之后，可以看到名称为`test`的队列就有了`DLX`。  
![在这里插入图片描述][20200213152502902.png]  
当该队列中有死信时，RabbitMQ服务器就会自动的将这个消息重新投递到设置的exchange上去，进而被路由到另一个队列中。

我们这里只测试消息TTL过期的情况，并且是设置消息本身的TTL，而不是队列的TTL。

**生产端**

package com.kaven.rabbitmq.api.dlx;
    
    import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Producer { 
        // 自己服务器的IP
        private static String ip = "IP";
        // RabbitMQ启动的默认端口，也是应用程序进行连接RabbitMQ的端口
        private static int port = 5672;
        // RabbitMQ有一个 "/" 的虚拟主机
        private static String virtualHost = "/";
    
        // default exchange
        private static String exchange = "";
        // default exchange 的路由规则： routingKey（test） 将匹配同名的 queue(test)
        private static String routingKey = "test";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            // 1 创建ConnectionFactory
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost(ip);
            connectionFactory.setPort(port);
            connectionFactory.setVirtualHost(virtualHost);
    
            // 2 创建Connection
            Connection connection = connectionFactory.newConnection();
    
            // 3 创建Channel
            Channel channel = connection.createChannel();
    
            // 4 发送消息
            for (int i = 0; i < 5; i++) { 
                AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                        .deliveryMode(2)
                        .contentEncoding("UTF-8")
                        .expiration("10000")
                        .build();
                String msg = "RabbitMQ: dlx message" + i;
                channel.basicPublish(exchange , routingKey , properties , msg.getBytes());
            }
    
            // 5 关闭连接
            channel.close();
            connection.close();
        }
    }

**消费端**

package com.kaven.rabbitmq.api.dlx;
    
    
    import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    
    import java.io.IOException;
    
    public class MyConsumer extends DefaultConsumer { 
    
        public Channel channel;
    
        public MyConsumer(Channel channel) { 
            super(channel);
            this.channel = channel;
        }
    
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope,
                                   AMQP.BasicProperties properties, byte[] body) throws IOException { 
            System.out.println("------------ consumer message -----------");
            System.out.println("body：" + new String(body));
        }
    }

package com.kaven.rabbitmq.api.dlx;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.TimeoutException;
    
    public class Consumer { 
        // 自己服务器的IP
        private static String ip = "IP";
        // RabbitMQ启动的默认端口，也是应用程序进行连接RabbitMQ的端口
        private static int port = 5672;
        // RabbitMQ有一个 "/" 的虚拟主机
        private static String virtualHost = "/";
    
        // default exchange
        private static String exchange = "";
        // 队列名
        private static String queueName = "test";
    
        // 死信队列的设置
        private static String dlxExchange = "dlx.exchange";
        private static String dlxQueueName = "dlx.queue";
        private static String dlxRoutingKey = "#";
    
        public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { 
            // 1 创建ConnectionFactory
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost(ip);
            connectionFactory.setPort(port);
            connectionFactory.setVirtualHost(virtualHost);
    
            // 2 创建Connection
            Connection connection = connectionFactory.newConnection();
    
            // 3 创建Channel
            Channel channel = connection.createChannel();
    
            // 4 设置死信队列
            Map<String , Object> arguments = new HashMap<String , Object>();
            arguments.put("x-dead-letter-exchange" , dlxExchange);
            channel.exchangeDeclare(dlxExchange , "topic" , true , false , null);
            channel.queueDeclare(dlxQueueName , true , false , false , null);
            channel.queueBind(dlxQueueName , dlxExchange , dlxRoutingKey);
    
            // 5 创建Queue
            channel.queueDeclare(queueName , true , false , false , arguments);
    
            // 6 消费消息
    // channel.basicConsume(queueName , true , new MyConsumer(channel));
        }
    }

// 死信队列的设置
        private static String dlxExchange = "dlx.exchange";
        private static String dlxQueueName = "dlx.queue";
        private static String dlxRoutingKey = "#";

这是为死信队列设置的相关参数，交换机名称、队列名称、交换机与队列绑定的`routingKey`，这里的交换机是`topic`类型的，`routingKey="#"`是当消息变成死信时，该消息就会被投递到该交换机（`dlx.exchange`）上，根据这个`routingKey`，可以让任何在该交换机（`dlx.exchange`）上的消息都可以路由到该队列（`dlx.queue`）中。

定义死信队列的交换机（`dlx.exchange`）、队列（`dlx.queue`），并且绑定该交换机和队列，并且设置`routingKey`（`dlxRoutingKey = "#"`）。

// 创建Queue
            channel.queueDeclare(queueName , true , false , false , arguments);

定义队列时，为队列添加一个`DLX`（在`arguments`参数中体现）。  
我们这里也把消费端的消费消息给关闭了，方便测试消息过期后是否会被投递到死信队列中。

**测试**

启动生产端和消费端，看看`RabbitMQ Management`，之所以有`10`条消息，是因为我测试了两次。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70]  
`10`秒种内，消息还没过期，还在`test`中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 1]  
`10`秒种后，消息就过期了，消息便到`dlx.queue`中了，而`test`中的消息就没有了，很明显符合预期。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 2]

[Link 1]: https://kaven.blog.csdn.net/article/details/104223534
[20200213152502902.png]: https://img-blog.csdnimg.cn/20200213152502902.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200213150725181.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200213150823519.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/2020021315094328.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz,size_16,color_FFFFFF,t_70