RabbitMQ - 死信队列，延时队列

电玩女神 2023-09-27 18:31 98阅读 0赞

## [Time-To-Live and Expiration — RabbitMQ][Time-To-Live and Expiration _ RabbitMQ] ##

## 一、死信队列 ##

## [Dead Letter Exchanges — RabbitMQ][Dead Letter Exchanges _ RabbitMQ] ##

#### 死信队列： ####

DLX 全称（Dead-Letter-Exchange）,称之为死信交换器，当消息变成一个死信之后，如果这个消息所在的队列存在**x-dead-letter-exchange**参数，那么它会被发送到x-dead-letter-exchange对应值的交换器上，这个交换器就称之为死信交换器，与这个死信交换器绑定的队列就是死信队列

![cd91ce6c4e4be7616ea4ecd2cf17e050.png][]

#### 死信消息： ####

*  消息被拒绝（Basic.Reject或Basic.Nack）并且设置 requeue 参数的值为 false
 *  消息过期（消息TTL过期。TTL：Time To Live的简称，即过期时间）
 *  队列达到最大的长度

#### 过期消息： ####

在 rabbitmq 中存在2种方法可设置消息的过期时间:

*  第一种通过**对队列进行设置**，这种设置后，该队列中所有的消息都存在相同的过期时间
 *  第二种通过**对消息本身进行设置**，那么每条消息的过期时间都不一样

如果**同时使用这2种方法，那么以过期时间小的那个数值为准。**当消息达到过期时间还没有被消费，那么那个消息就成为了一个 **死信** 消息

**队列设置**：在队列申明的时候使用\*\* x-message-ttl \*\*参数，单位为 毫秒；

*  队列中这个属性的设置要在第一次声明队列的时候设置才有效，如果队列一开始已存在且没有这个属性，则要删掉队列再重新声明才可以。
 *  队列的 TTL 只能被设置为某个固定的值，一旦设置后则不能更改，否则会抛出异常

**单个消息设置**：是设置消息属性的 expiration 参数的值，单位为 毫秒。

**说明：**

对于第一种设置队列属性的方法，一旦消息过期，就会从队列中抹去；而在第二种方法中，即使消息过期，也不会马上从队列中抹去，因为每条消息是否过期是在即将投递到消费者之前判定的

1. 生产者：  
　　声明队列的时候用属性指定其死信队列交换机名称。

测试：

package rabbitmq;
    
    import java.io.IOException;
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.TimeoutException;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    public class Producer {
    
        public static ConnectionFactory getConnectionFactory() {
            // 创建连接工程，下面给出的是默认的case
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.99.100");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            factory.setVirtualHost("/");
            return factory;
        }
    
        public static void main(String[] args) throws IOException, TimeoutException  {
            ConnectionFactory connectionFactory = getConnectionFactory();
            Connection newConnection = null;
            Channel createChannel = null;
            try {
                newConnection = connectionFactory.newConnection();
                createChannel = newConnection.createChannel();
                
                // 声明一个正常的direct类型的交换机
                createChannel.exchangeDeclare("order.exchange", BuiltinExchangeType.DIRECT);
                // 声明死信交换机为===order.dead.exchange
                String dlxName = "order.dead.exchange";
                createChannel.exchangeDeclare(dlxName, BuiltinExchangeType.DIRECT);
                // 声明队列并指定死信交换机为上面死信交换机
                Map<String, Object> arg = new HashMap<String, Object>();
                arg.put("x-dead-letter-exchange", dlxName);
                createChannel.queueDeclare("myQueue", true, false, false, arg);
                
                String message = "测试消息";
                createChannel.basicPublish("order.exchange", "routing_key_myQueue", null, message.getBytes());
                System.out.println("消息发送成功");
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (createChannel != null) {
                    createChannel.close();
                }
                if (newConnection != null) {
                    newConnection.close();
                }
            }
            
        }
    }

结果：

(1)生成两个Exchange

(2)队列myQueue的死信队列有属性

2. 消费者：  
　　一个消费者监听正常队列，一个消费者监听死信队列。(只是绑定的交换机不同)

消费者一：监听正常队列

package rabbitmq;
    
    import java.io.IOException;
    import java.util.Date;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.TimeoutException;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    import com.rabbitmq.client.AMQP.BasicProperties;
    
    public class Consumer {
    
        public static ConnectionFactory getConnectionFactory() {
            // 创建连接工程，下面给出的是默认的case
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.99.100");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            factory.setVirtualHost("/");
            return factory;
        }
    
        public static void main(String[] args) throws IOException, TimeoutException {
            ConnectionFactory connectionFactory = getConnectionFactory();
            Connection newConnection = null;
            Channel createChannel = null;
            try {
                newConnection = connectionFactory.newConnection();
                createChannel = newConnection.createChannel();
    
                // 队列绑定交换机-channel.queueBind(队列名, 交换机名, 路由key[广播消息设置为空串])
                createChannel.queueBind("myQueue", "order.exchange", "routing_key_myQueue");
    
                createChannel.basicConsume("myQueue", false, "", new DefaultConsumer(createChannel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                            byte[] body) throws IOException {
    
                        System.out.println("consumerTag: " + consumerTag);
                        System.out.println("envelope: " + envelope);
                        System.out.println("properties: " + properties);
                        String string = new String(body, "UTF-8");
                        System.out.println("接收到消息： -》 " + string);
    
                        long deliveryTag = envelope.getDeliveryTag();
                        Channel channel = this.getChannel();
                        System.out.println("拒绝消息， 使之进入死信队列");
                        System.out.println("时间： " + new Date());
                        try {
                            TimeUnit.SECONDS.sleep(3);
                        } catch (InterruptedException e) {
                        }
                        
                        // basicReject第二个参数为false进入死信队列或丢弃
                        channel.basicReject(deliveryTag, false);
                    }
                });
    
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
            }
    
        }
    }

消费者二：监听死信队列

package rabbitmq;
    
    import java.io.IOException;
    import java.util.Date;
    import java.util.concurrent.TimeoutException;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    import com.rabbitmq.client.AMQP.BasicProperties;
    
    public class Consumer2 {
    
        public static ConnectionFactory getConnectionFactory() {
            // 创建连接工程，下面给出的是默认的case
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.99.100");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            factory.setVirtualHost("/");
            return factory;
        }
    
        public static void main(String[] args) throws IOException, TimeoutException {
            ConnectionFactory connectionFactory = getConnectionFactory();
            Connection newConnection = null;
            Channel createChannel = null;
            try {
                newConnection = connectionFactory.newConnection();
                createChannel = newConnection.createChannel();
    
                // 队列绑定交换机-channel.queueBind(队列名, 交换机名, 路由key[广播消息设置为空串])
                createChannel.queueBind("myQueue", "order.dead.exchange", "routing_key_myQueue");
    
                createChannel.basicConsume("myQueue", false, "", new DefaultConsumer(createChannel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                            byte[] body) throws IOException {
    
                        System.out.println("时间： " + new Date());
                        
                        System.out.println("consumerTag: " + consumerTag);
                        System.out.println("envelope: " + envelope);
                        System.out.println("properties: " + properties);
                        String string = new String(body, "UTF-8");
                        System.out.println("接收到消息： -》 " + string);
    
                        long deliveryTag = envelope.getDeliveryTag();
                        Channel channel = this.getChannel();
                        channel.basicAck(deliveryTag, true);
                        System.out.println("死信队列中处理完消息息");
                    }
                });
    
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
            }
    
        }
    }

结果： 消费者一先正常监听到，basicReject为false拒绝后进入死信队列；消费者二监听的死信队列收到消息。

消费者一打出的日志如下：

consumerTag: amq.ctag-0noHs24F0FsGe-dfwwqWNw
    envelope: Envelope(deliveryTag=1, redeliver=false, exchange=order.exchange, routingKey=routing_key_myQueue)
    properties: #contentHeader<basic>(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
    接收到消息： -》 测试消息
    拒绝消息， 使之进入死信队列
    时间： Sat Nov 07 12:18:44 CST 2020

消费者二打出的日志如下：

时间： Sat Nov 07 12:18:47 CST 2020
    consumerTag: amq.ctag-ajYMpMFkXHDiYWkD3XFJ7Q
    envelope: Envelope(deliveryTag=1, redeliver=false, exchange=order.dead.exchange, routingKey=routing_key_myQueue)
    properties: #contentHeader<basic>(content-type=null, content-encoding=null, headers={x-death=[{reason=rejected, count=1, exchange=order.exchange, time=Sat Nov 07 01:52:19 CST 2020, routing-keys=[routing_key_myQueue], queue=myQueue}]}, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
    接收到消息： -》 测试消息
    死信队列中处理完消息息

注意：

进入死信队列之后，headers 加了一些死信相关的信息，包括原队列以及进入死信的原因。

补充：在队列进入死信队列之前也可以修改其routingKey,而且只有在指定x-dead-letter-exchange的前提下才能修改下面属性，否则会报错

(1)修改生产者声明队列的方式，如下：

// 声明一个正常的direct类型的交换机
                createChannel.exchangeDeclare("order.exchange", BuiltinExchangeType.DIRECT);
                // 声明死信交换机为===order.dead.exchange
                String dlxName = "order.dead.exchange";
                createChannel.exchangeDeclare(dlxName, BuiltinExchangeType.DIRECT);
                // 声明队列并指定死信交换机为上面死信交换机
                Map<String, Object> arg = new HashMap<String, Object>();
                arg.put("x-dead-letter-exchange", dlxName);
                // 修改进入死信队列的routingkey，如果不修改会使用默认的routingKey
                arg.put("x-dead-letter-routing-key", "routing_key_myQueue_dead");
                createChannel.queueDeclare("myQueue", true, false, false, arg);

(2)修改监听死信队列的消费者二：

// 队列绑定交换机-channel.queueBind(队列名, 交换机名, 路由key[广播消息设置为空串])
                createChannel.queueBind("myQueue", "order.dead.exchange", "routing_key_myQueue_dead");

结果，收到消费者二收到的信息如下：

时间： Sat Nov 07 12:27:08 CST 2020
    consumerTag: amq.ctag-THqpEdYH_-iNeCIccgpuaw
    envelope: Envelope(deliveryTag=1, redeliver=false, exchange=order.dead.exchange, routingKey=routing_key_myQueue_dead)
    properties: #contentHeader<basic>(content-type=null, content-encoding=null, headers={x-death=[{reason=rejected, count=1, exchange=order.exchange, time=Sat Nov 07 02:00:41 CST 2020, routing-keys=[routing_key_myQueue], queue=myQueue}]}, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
    接收到消息： -》 测试消息
    死信队列中处理完消息

## 二、延时队列 ##

延迟队列，即消息进入队列后不会立即被消费，只有到达指定时间后，才会被消费

RabbitMQ本身没提供延时队列，我们可以利用消息的生存时间和死信队列实现延时

典型的应用场景就是订单30分钟内未支付就关闭订单，还有一种场景，账单24小时未确认，就发送提醒消息

![e8453d2662afd1996b8f84d6d78dfa00.png][]

[延时队列插件安装][Link 1]

#### 2.1.1、yml配置 ####

spring:
        rabbitmq:
            host: 192.168.99.12
            port: 5672
            username: guest
            password: guest
            # 发送确认
            publisher-confirms: true
            # 路由失败回调
            publisher-returns: true
            template:
                # 必须设置成true 消息路由失败通知监听者，false 将消息丢弃
                mandatory: true
            #消费端
            listener:
                simple:
                    # 每次从RabbitMQ获取的消息数量
                    prefetch: 1
                    default-requeue-rejected: false
                    # 每个队列启动的消费者数量
                    concurrency: 1
                    # 每个队列最大的消费者数量
                    max-concurrency: 1
                    # 签收模式为手动签收-那么需要在代码中手动ACK
                    acknowledge-mode: manual
    #邮件队列
    email:
        queue:
            name: demo.email
    
    #邮件交换器名称
    exchange:
        name: demoTopicExchange
    
    #死信队列
    dead:
        letter:
            queue:
                name: demo.dead.letter
            exchange:
                name: demoDeadLetterTopicExchange
    
    #延时队列
    delay:
        queue:
            name: demo.delay
        exchange:
            name: demoDelayTopicExchange

#### 2.1.2、延时队列配置 ####

/**
     * rabbitmq 配置
     *
     * @author DUCHONG
     * @since 2020-08-23 14:05
     **/
    @Configuration
    @Slf4j
    public class RabbitmqConfig {
    
    
        @Value("${email.queue.name}")
        private String emailQueue;
        @Value("${exchange.name}")
        private String topicExchange;
        @Value("${dead.letter.queue.name}")
        private String deadLetterQueue;
        @Value("${dead.letter.exchange.name}")
        private String deadLetterExchange;
        @Value("${delay.queue.name}")
        private String delayQueue;
        @Value("${delay.exchange.name}")
        private String delayExchange;
    
        @Bean
        public Queue emailQueue() {
    
            Map<String, Object> arguments = new HashMap<>(2);
            // 绑定死信交换机
            arguments.put("x-dead-letter-exchange", deadLetterExchange);
            // 绑定死信的路由key
            arguments.put("x-dead-letter-routing-key", deadLetterQueue+".#");
    
            return new Queue(emailQueue,true,false,false,arguments);
        }
    
    
        @Bean
        TopicExchange emailExchange() {
            return new TopicExchange(topicExchange);
        }
    
    
        @Bean
        Binding bindingEmailQueue() {
            return BindingBuilder.bind(emailQueue()).to(emailExchange()).with(emailQueue+".#");
        }
    
    
        //私信队列和交换器
        @Bean
        public Queue deadLetterQueue() {
            return new Queue(deadLetterQueue);
        }
    
        @Bean
        TopicExchange deadLetterExchange() {
            return new TopicExchange(deadLetterExchange);
        }
    
        @Bean
        Binding bindingDeadLetterQueue() {
            return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with(deadLetterQueue+".#");
        }
        //延时队列
        @Bean
        public Queue delayQueue() {
            return new Queue(delayQueue);
        }
    
        @Bean
        CustomExchange delayExchange() {
            Map<String, Object> args = new HashMap<>();
            args.put("x-delayed-type", "topic");
            //参数二为类型：必须是x-delayed-message
            return new CustomExchange(delayExchange, "x-delayed-message", true, false, args);
    
        }
    
        @Bean
        Binding bindingDelayQueue() {
            return BindingBuilder.bind(delayQueue()).to(delayExchange()).with(delayQueue+".#").noargs();
        }
    }

### 2.2、消息发送方 ###

30分钟时间太久了，这里延时2分钟来看效果

@Configuration
    @EnableScheduling
    @Slf4j
    public class ScheduleController {
    
        @Autowired
        RabbitTemplate rabbitTemplate;
    
        @Value("${exchange.name}")
        private String topicExchange;
    
        @Value("${delay.exchange.name}")
        private String delayTopicExchange;
    
        @Scheduled(cron = "0 0/1 * * * ?")
        public void sendEmailMessage() {
    
            String msg = RandomStringUtils.randomAlphanumeric(8);
            JSONObject email=new JSONObject();
            email.put("content",msg);
            email.put("to","duchong@qq.com");
            CorrelationData correlationData=new CorrelationData(UUID.randomUUID().toString());
            rabbitTemplate.convertAndSend(topicExchange,"demo.email.x",email.toJSONString(),correlationData);
            log.info("---发送 email 消息---{}---messageId---{}",email,correlationData.getId());
        }
    
    
        @Scheduled(cron = "0 0/1 * * * ?")
        public void sendDelayOrderMessage() throws Exception{
    
            //订单号 id实际是保存订单后返回的，这里用uuid代替
            String orderId = UUID.randomUUID().toString();
            // 模拟订单信息
            JSONObject order=new JSONObject();
            order.put("orderId",orderId);
            order.put("goodsName","vip充值");
            order.put("orderAmount","99.00");
            CorrelationData correlationData=new CorrelationData(orderId);
            MessageProperties messageProperties = new MessageProperties();
            messageProperties.setMessageId(orderId);
            //30分钟时间太长，这里延时120s消费
            messageProperties.setHeader("x-delay", 120000);
            Message message = new Message(order.toJSONString().getBytes(CharEncoding.UTF_8), messageProperties);
    
            rabbitTemplate.convertAndSend(delayTopicExchange,"demo.delay.x",message,correlationData);
    
            log.info("---发送 order 消息---{}---orderId---{}",order,correlationData.getId());
            //睡一会，为了看延迟效果
            TimeUnit.MINUTES.sleep(10);
        }
    }

### 2.3、消息消费方 ###

@Component
    @Slf4j
    public class MessageHandler {
    
    
        /**
         * 邮件发送
         * @param message
         * @param channel
         * @param headers
         * @throws IOException
         */
        @RabbitListener(queues ="demo.email")
        @RabbitHandler
        public void handleEmailMessage(Message message, Channel channel, @Headers Map<String,Object> headers) throws IOException {
    
            try {
    
                String msg=new String(message.getBody(), CharEncoding.UTF_8);
                JSONObject jsonObject = JSON.parseObject(msg);
                jsonObject.put("messageId",headers.get("spring_returned_message_correlation"));
                log.info("---接受到消息---{}",jsonObject);
    			//主动异常
    			int m=1/0;
                //手动签收
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            }
            catch (Exception e) {
                log.info("handleEmailMessage捕获到异常,拒绝重新入队---消息ID---{}", headers.get("spring_returned_message_correlation"));
                //异常，ture 重新入队,或者false,进入死信队列
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
    
            }
        }
    
        /**
         * 死信消费者，自动签收开启状态下，超过重试次数，或者手动签收，reject或者Nack
         * @param message
         */
        @RabbitListener(queues = "demo.dead.letter")
        public void handleDeadLetterMessage(Message message, Channel channel,@Headers Map<String,Object> headers) throws IOException {
    
            //可以考虑数据库记录，每次进来查数量，达到一定的数量，进行预警，人工介入处理
            log.info("接收到死信消息:---{}---消息ID---{}", new String(message.getBody()),headers.get("spring_returned_message_correlation"));
    		//回复ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }
    
        /**
         * 延时队列消费
         * @param message
         * @param channel
         * @param headers
         * @throws IOException
         */
        @RabbitListener(queues ="demo.delay")
        @RabbitHandler
        public void handleOrderDelayMessage(Message message, Channel channel, @Headers Map<String,Object> headers) throws IOException {
    
            try {
    
                String msg=new String(message.getBody(), CharEncoding.UTF_8);
                JSONObject jsonObject = JSON.parseObject(msg);
                log.info("---接受到订单消息---orderId---{}",message.getMessageProperties().getMessageId());
                log.info("---订单信息---order---{}",jsonObject);
                //业务逻辑，根据订单id获取订单信息，如果还未支付，设置关闭状态，如果已支付，不做任何处理
                //手动签收
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            }
            catch (Exception e) {
                log.info("handleOrderDelayMessage捕获到异常,重新入队---orderId---{}", headers.get("spring_returned_message_correlation"));
                //异常，ture 重新入队,或者false,进入死信队列
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);
    
            }
        }
    
    }

### 2.4、结果 ###

![64616fa56419a4c77a8cdd08f2e97572.png][]

运行结果显示，同一个订单号的消息，发送过后2分钟，消费者才接受到，符合预期

[https://www.cnblogs.com/geekdc/p/13550620.html][https_www.cnblogs.com_geekdc_p_13550620.html]

[消息队列RabbitMQ（五）：死信队列与延迟队列][RabbitMQ]

[rabbitmq的延迟队列和死信队列\_死信队列和延时队列的区别\_zhuwenaptx的博客-CSDN博客][rabbitmq_zhuwenaptx_-CSDN]

[RabbitMQ的死信队列和延时队列 - 简书][RabbitMQ_ -]

[RabbitMQ死信队列与延迟队列\_51CTO博客\_rabbitmq延迟队列][RabbitMQ_51CTO_rabbitmq]

[Time-To-Live and Expiration _ RabbitMQ]: https://www.rabbitmq.com/ttl.html
[Dead Letter Exchanges _ RabbitMQ]: https://www.rabbitmq.com/dlx.html
[cd91ce6c4e4be7616ea4ecd2cf17e050.png]: https://img-blog.csdnimg.cn/img_convert/cd91ce6c4e4be7616ea4ecd2cf17e050.png
[e8453d2662afd1996b8f84d6d78dfa00.png]: https://img-blog.csdnimg.cn/img_convert/e8453d2662afd1996b8f84d6d78dfa00.png
[Link 1]: https://www.cnblogs.com/geekdc/p/13549613.html
[64616fa56419a4c77a8cdd08f2e97572.png]: https://img-blog.csdnimg.cn/img_convert/64616fa56419a4c77a8cdd08f2e97572.png
[https_www.cnblogs.com_geekdc_p_13550620.html]: https://www.cnblogs.com/geekdc/p/13550620.html
[RabbitMQ]: https://www.bbsmax.com/A/VGzl1WMxJb/
[rabbitmq_zhuwenaptx_-CSDN]: https://blog.csdn.net/zhuwenaptx/article/details/121074035
[RabbitMQ_ -]: https://www.jianshu.com/p/256d2eaf1786
[RabbitMQ_51CTO_rabbitmq]: https://blog.51cto.com/u_12826294/3091092