RabbitMQ高级特性之消息过期机制以及死信队列

﹏ヽ暗。殇╰゛Y 2022-08-29 10:02 316阅读 0赞

# RabbitMQ高级特性之消息过期机制 #

## 前言 ##

在下单场景中，如果一个用户在下单后迟迟没有付款，过来一段时间后，我们需要将这个订单标记为取消状态。  
那如何取实现这个自动取消订单功能呢？

#### 定期轮询数据库 ####

用户下单成功，将订单信息放入数据库，同时将支付状态放入数据库，用户付款更改数据库状态。定期轮询数据库支付状态，如果超过30分钟就将该订单取消。

*  优点：设计实现简单
 *  缺点：需要对数据库进行大量的IO操作，效率低下，数据量以及并发大的时候，会卡死数据库，延迟性比较高。

#### 使用JDK自带Timer ####

*  Timers没有持久化机制.
 *  Timers不灵活 (只可以设置开始时间和重复间隔，对等待支付貌似够用)
 *  Timers 不能利用线程池，一个timer一个线程
 *  Timers没有真正的管理计划

#### ScheduledExecutorService线程池 ####

*  可以多线程执行，一定程度上避免任务间互相影响，单个任务异常不影响其它任务。
 *  在高并发的情况下，不建议使用定时任务去做，因为太浪费服务器性能，不推荐。

#### 支持ttl机制的mq ####

在mq中的消息超过某个时间会被自动从队列里删除，可以很优雅的解决这种问题，还可以避免大量消息在mq中堆积导致不能消费，降低mq压力等等。

## TTL机制 ##

TTL，Time to Live 的简称，即过期时间。

RabbitMQ 可以对消息和队列两个维度来设置TTL。

任何消息中间件的容量和堆积能力都是有限的，如果有一些消息总是不被消费掉，那么需要有一种过期的机制来做兜底。  
此时ttl可以减少消息堆积，提供Mq运载能力。

*  1.  通过Queue属性设置，队列中所有消息都有相同的过期时间。
 *  1.  对消息自身进行单独设置，每条消息的TTL 可以不同。

首先来看对整个队列设置ttl

package ttl;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    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;
    
    
    /** * 工作队列模式 * 生产者发消息 * * * ttl机制 * 1 可以对队列设置过期时间，队列中的过期时间都一致 * 2 对小时设置过期时间，每个消息的过期时间不一致(缺点，会导致不同消息因为不先出队列的时间过长导致队列里其他消息过期了，还没有被消费。) * */
    public class Produce { 
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            //连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            connectionFactory.setVirtualHost("/");
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            //amqp 协议端口
            connectionFactory.setPort(5672);
            //连接
            Connection connection = connectionFactory.newConnection();
            //获取通道
            Channel channel = connection.createChannel();
    
            // 创建队列（实际上使用的是AMQP default这个direct类型的交换器）
            // 设置队列属性
             Map<String, Object> arguments = new HashMap<>();
            // 设置队列的TTL
             arguments.put("x-message-ttl", 30000);
            // 设置队列的空闲存活时间（如该队列根本没有消费者，一直没有使用，队列可以存活多久） 根据自己业务设置
             arguments.put("x-expires", 10000);
    
            //声明消息队列 名称
            // 是否可持久化
            // 排他 （队列中没消息时候，系统）
            // 是否自动删除消息队列
            // 属性map集合
             channel.queueDeclare("queue.ttl", false, false, false, arguments);
    
            //声明交换器
            // 交换器名称,
            // 交换器类型，
            // 是否持久化，
            // 是否自动删除的，
            // 属性map属性
            channel.exchangeDeclare("ttl.exchange", BuiltinExchangeType.DIRECT,true,false,false,null);
    
    
            //队列和交换器绑定 并指定路由键
            channel.queueBind("queue.ttl","ttl.exchange","ttl.key");
    
            //发送消息 交换器 路由key,属性，消息 -> amqp协议会将消息发送出去
            //发送消息 发送到交换器，指定路由key
            //channel.basicPublish("work.exchange.biz","test",null,"hello world 123".getBytes());
    
            //发送消息
            for (int i = 0; i < 20; i++) { 
                channel.basicPublish("ttl.exchange","ttl.key",null,("消息 -》 " + i ).getBytes());
    
            }
    
            channel.close();
            connection.close();
    
    
        }
    }

运行程序后，在控制台可看到，此时队列是ttl类型，以及有exp标志，然后我们发送了20条数据，过一段时间之后，20条消息会从队列里消失。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70]

但从队列里消失中我们如何将次消息(订单)标记为取消状态呢？此时就需要`死信队列`上场了之后介绍。

对单个消息设置ttl

核心代码如下

String message = ("消息 -》 " + i );
                //设置对消息过期 单位毫秒 ，2 持久化消息
                AMQP.BasicProperties build = new AMQP.BasicProperties().builder().expiration("30000").deliveryMode(2).build();
                channel.basicPublish("ttl.exchange","ttl.key",build,message.getBytes());

`那如果两种方法一起使用呢?`

则消息的TTL 以`两者之间较小数值`为准。通常来讲，消息在队列中的生存时间一旦超过设置的TTL 值时，就会变成“死信”(Dead Message)，消费者默认就无法再收到该消息。当然，“死信”也是可以被取出来消费的。

## 死信队列 ##

在定义业务队列时可以考虑指定一个 死信交换机，并绑定一个死信队列。当消息变成死信时，该消息就会被发送到该死信队列上，这样方便我们查看消息失败的原因。DLX，全称为Dead-Letter-Exchange，死信交换器。

消息在一个队列中变成死信（Dead Letter）之后，被重新发送到一个特殊的交换器（DLX）中，同时，绑定DLX的队列就称为“死信队列”。

以下几种情况导致消息变为死信：

1.  消息被拒绝（Basic.Reject/Basic.Nack），并且设置requeue参数为false； 2. 消息过期；
2.  队列达到最大长度。

对于RabbitMQ 来说，DLX 是一个非常有用的特性。它可以处理异常情况下，消息不能够被消费者正确消费(消费者调用了Basic.Nack 或者Basic.Reject)而被置入死信队列中的情况，后续分析程序可以通过消费这个死信队列中的内容来分析当时所遇到的异常情况，进而可以改善和优化系统。

package dlx;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    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 ProduceForQueueDlx { 
    
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
    
            //连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            connectionFactory.setVirtualHost("/");
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
    
            //amqp 协议端口
            connectionFactory.setPort(5672);
    
            //简历连接
            Connection connection = connectionFactory.newConnection();
    
            //获取通道
            Channel channel = connection.createChannel();
    
    
            // 创建队列（实际上使用的是AMQP default这个direct类型的交换器）
    
    
            //声明死信交换器
            channel.exchangeDeclare("ex.dlx",BuiltinExchangeType.DIRECT,true);
            //声明队列做死信队列
            channel.queueDeclare("queue.lx", true, false, false,null);
            //绑定死信交换器和死信队列
            channel.queueBind("queue.lx","ex.dlx","key.dlx");
    
    
    
    
            // 设置队列属性
             Map<String, Object> arguments = new HashMap<>();
            // 设置队列的TTL
            arguments.put("x-message-ttl", 5000);
            // 设置队列的空闲存活时间（如该队列根本没有消费者，一直没有使用，队列可以存活多久） 根据自己业务设置
             arguments.put("x-expires", 6 * 10000);
    
            //指定过期消息通过死信交换器发送到死信队列，死信交换器的名称
            //指定死信交换器
            arguments.put("x-dead-letter-exchange", "ex.dlx");
            //指定死信交换器的路由键
            arguments.put("x-dead-letter-routing-key", "key.dlx");
    
    
    
            //声明消息队列 名称
            // 是否可持久化
            // 排他 （队列中没消息时候，系统）
            // 是否自动删除消息队列
            // 属性map集合
            channel.queueDeclare("queue.biz", true, false, false, arguments);
    
            //声明交换器
            // 交换器名称,
            // 交换器类型，
            // 是否持久化，
            // 是否自动删除的，
            // 属性map属性
            channel.exchangeDeclare("ex.biz", BuiltinExchangeType.DIRECT,true,false,false,null);
    
    
            // 业务 队列和交换器绑定 并指定路由键
            channel.queueBind("queue.biz","ex.biz","biz.key");
    
    
    
    
            //发送消息 交换器 路由key,属性，消息 -> amqp协议会将消息发送出去
            //发送消息 发送到交换器，指定路由key
            //channel.basicPublish("work.exchange.biz","test",null,"hello world 123".getBytes());
    
            //发送消息
            for (int i = 0; i < 20; i++) { 
                channel.basicPublish("ex.biz","biz.key",null,("消息 -》 " + i ).getBytes());
    
            }
    
            channel.close();
            connection.close();
    
    
        }
    }

发送消息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 1]  
5s后 ，消息过期转移到死信队列中  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 2]

放到死信队列之后，我们再单独对死信队列监听或者拉取数据进行处理即可。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70]: /images/20220829/c51ed92135644fb9955c456914e94775.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 1]: /images/20220829/ecb47c60d2144fb8ba38b9b6762cc5b0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 2]: /images/20220829/45e098dc763a43bcbc8e3d88a223e5a4.png