RabbitMQ:伪延时队列

目录

  • 一、什么是延时队列
  • 二、RabbitMQ实现
  • 三、 延时队列的问题
  • 四、解决RabbitMQ的伪延时方案

ps:伪延时队列先卖个关子,我们先了解下延时队列。

一、什么是延时队列

 所谓延时队列是指消息push到队列后,监听的消费者不能第一时间获取消息,需要等到指定时间才能消费。
 一般在业务里面需要对某些消息做定时发送,不想走定时任务或者是用户下单之后多长时间自动失效类似的场景可以考虑通过延时队列实现。

二、RabbitMQ实现

MQ本身并不支持直接的延时队列实现,但是我们可以通过RabbitMQ的消息TTL和Dead Letter规则来实现

  1. Time TO Live (TTL):

    RabbitMQ可以针对Queue设置x-expires 或者 针对Message设置 x-message-ttl,来控制消息的生存时间

  2. Dead Letter 死信

    RabbitMQ官网这样定义死信消息:

  • . 消息被拒绝(basic.reject或basic.nack)并且requeue=false.
  • . 消息TTL过期
  • 队列达到最大长度(队列满了,无法再添加数据到mq中)
  1. Dead Letter Exchanges(DLX)死信交换机

    MQ默认的死信消息是丢弃的,但是我们可以通过设置以下两个属性让死信消息转发到我们指定的队列。

  • x-dead-letter-exchange:出现dead letter之后将dead letter重新发送到指定exchange
  • x-dead-letter-routing-key:出现dead letter之后将dead letter重新按照指定的routing-key发送
  1. 延时队列实现:

    了解了MQ队列的TTL和Dead Letter之后,我们就可以通过这两个特性来实现,首先我们通过设置消息或者队列的TTL来设置消息在指定时间后成为死信,再设置死信消息的路由转发规则到特定队列,消费者通过监听这个特定队列就能实现延时队列的效果。

  2. 代码实现

生产者发送消息:ttlQueue存放过期时间的队列,deadLetterQueue死信转发队列,seconds是过期时间

 public static void sendTTLMsg(String ttlQueue, String deadLetterQueue, Object msg, Integer seconds) {
        MqSender.getInstance().setHost(RABBIT_MQ_HOST);
        // 获取到连接以及MQ通道
        Connection connection;
        try {
            connection = MqSender.getInstance().newConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
            // 配置
            Map<String, Object> args = new HashMap<String, Object>();
            args.put("x-dead-letter-exchange", "");
            args.put("x-dead-letter-routing-key", deadLetterQueue);
            channel.queueDeclare(deadLetterQueue, true, false, false, null);
            channel.queueDeclare(ttlQueue, true, false, false, args);
            // 发送消息
            channel.basicPublish("", ttlQueue, new AMQP.BasicProperties.Builder().expiration(String.valueOf(seconds)).build(), MAPPER.writeValueAsBytes(msg));
            channel.close();
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

消费者通过监听deadLetterQueue来实现延时消息监听

三、 延时队列的问题

通过我们测试发现,这种方式实现的延时队列,在队列设置TTL的情况下是可以正常的,但是如果根据消息设置了不同的TTL,就会有问题,因为MQ本质上还是消息队列中间件,队列是遵循先进先出的,如果有两个消息先后入队,但是后入队的消息TTL小于前面的消息,它必须等待之前的消息被消费完后才能挪到队列头部,这样不同延时消息就会出现问题。

通过RabbitMQ官网的文档也介绍了这个问题:

 Only when expired messages reach the head of a queue will they actually be discarded (or dead-lettered)

所以我才称之为MQ的伪延时队列,这种延时队列在消息TTL不同的情况下并不能实现真正的延时消费。

四、解决RabbitMQ的伪延时方案

既然RabbitMQ无法支持不同TTL消息的延时消费,那么如果我们要实现这种功能,有什么方案呢,在实际业务开发中,我们有这样的解决方案:

首先我们会创建多级延时消费队列(比如两分钟,三十分钟,一天三种,具体可以根据业务量和访问量还有时间精确度来划分,这里的两分钟、三十分钟是指队列统一的TTL),push消费队列的时候,会根据需要延时的时间,丢到不同的消费队列,比如小于三十分钟的我们push到两分钟队列,三十分钟到一天的放入三十分钟队列,超过一天的放入一天队列,在死信队列的监听器做同样的判断,如果是小于等于当前时间消息的,立马消费,否则按照上述规则继续循环到不同的延时队列

这种方案解决了多级延时消费的问题,并且能够较大程度地避免了消息的重复循环,降低MQ的压力,但是缺点也比较明显,因为最低是两分钟的延时,理论上来说最多会有两分钟的误差,如果对时间要求性比较高的,可以适当调低最低一级别的延时TTL,比如一分钟或者三十秒

类似代码如下:cts是需要消费掉的时间戳

 long now = System.currentTimeMillis();
        long cts = Long.valueOf(feedComment.getCts());
        if (cts - now <= 30 * 60 * 1000) {
            MqSender.sendTTLMsg(MqConstants.FEED_COMMENT_DELAY_QUEUE_2MIN, MqConstants.FEED_COMMENT_AUTO_POST_QUEUE, feedComment, 2 * 60);
        } else if (cts - now <= 24 * 60 * 60 * 1000) {
            MqSender.sendTTLMsg(MqConstants.FEED_COMMENT_DELAY_QUEUE_30MIN, MqConstants.FEED_COMMENT_AUTO_POST_QUEUE, feedComment, 30 * 60);
        } else {
            MqSender.sendTTLMsg(MqConstants.FEED_COMMENT_DELAY_QUEUE_24HOUR, MqConstants.FEED_COMMENT_AUTO_POST_QUEUE, feedComment, 24 * 60 * 60);
        }

原文地址:https://www.cnblogs.com/Kelin-/p/9485984.html

时间: 2024-10-27 13:17:30

RabbitMQ:伪延时队列的相关文章

基于rabbitMQ 消息延时队列方案 模拟电商超时未支付订单处理场景

前言 传统处理超时订单 采取定时任务轮训数据库订单,并且批量处理.其弊端也是显而易见的:对服务器.数据库性会有很大的要求,并且当处理大量订单起来会很力不从心,而且实时性也不是特别好 当然传统的手法还可以再优化一下,即存入订单的时候就算出订单的过期时间插入数据库,设置定时任务查询数据库的时候就只需要查询过期了的订单,然后再做其他的业务操作 jdk延迟队列 DelayQueue 采取jdk自带的延迟队列能很好的优化传统的处理方案,但是该方案的弊.端也是非常致命的,所有的消息数据都是存于内存之中,一旦

RabbitMq 实现延时队列-Springboot版本

rabbitmq本身没有实现延时队列,但是可以通过死信队列机制,自己实现延时队列: 原理:当队列中的消息超时成为死信后,会把消息死信重新发送到配置好的交换机中,然后分发到真实的消费队列: 步骤: 1.创建带有时限的队列 dealLineQueue; 2.创建死信Faout交换机dealLineExchange; 3.创建消费队列realQueue,并和dealLineExchange绑定 4.配置dealLineQueue 的过期时间,消息过期后的死信交换机,重发的routing-key: 以下

RabbitMQ实现延时队列(死信队列)

基于队列和基于消息的TTL TTL是time to live 的简称,顾名思义指的是消息的存活时间.rabbitMq可以从两种维度设置消息过期时间,分别是队列和消息本身. 队列消息过期时间-Per-Queue Message TTL: 通过设置队列的x-message-ttl参数来设置指定队列上消息的存活时间,其值是一个非负整数,单位为微秒.不同队列的过期时间互相之间没有影响,即使是对于同一条消息.队列中的消息存在队列中的时间超过过期时间则成为死信. 死信交换机DLX 队列中的消息在以下三种情况

Rabbitmq的延时队列的使用

配置: spring: rabbitmq: addresses: 192.168.108.128:5672 connection-timeout: 15000 username: guest password: guest publisher-confirms: true publisher-returns: true 依赖: <!--rabbitmq --> <dependency> <groupId>org.springframework.boot</grou

springboot-rabbitmq:实现延时队列

延时队列应用于什么场景 延时队列顾名思义,即放置在该队列里面的消息是不需要立即消费的,而是等待一段时间之后取出消费.那么,为什么需要延迟消费呢?我们来看以下的场景 网上商城下订单后30分钟后没有完成支付,取消订单(如:淘宝.去哪儿网)    系统创建了预约之后,需要在预约时间到达前一小时提醒被预约的双方参会    系统中的业务失败之后,需要重试 这些场景都非常常见,我们可以思考,比如第二个需求,系统创建了预约之后,需要在预约时间到达前一小时提醒被预约的双方参会.那么一天之中肯定是会有很多个预约的

spring boot Rabbitmq集成,延时消息队列实现

本篇主要记录Spring boot 集成Rabbitmq,分为两部分, 第一部分为创建普通消息队列, 第二部分为延时消息队列实现: spring boot提供对mq消息队列支持amqp相关包,引入即可: [html] view plain copy <!-- rabbit mq --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-

springboot使用RabbitMQ实现延时任务

延时队列顾名思义,即放置在该队列里面的消息是不需要立即消费的,而是等待一段时间之后取出消费.那么,为什么需要延迟消费呢?我们来看以下的场景 订单业务: 在电商/点餐中,都有下单后 30 分钟内没有付款,就自动取消订单.短信通知: 下单成功后 60s 之后给用户发送短信通知.失败重试: 业务操作失败后,间隔一定的时间进行失败重试. 本文基于springboot,使用rabbitmq_delayed_message_exchange插件实现延时队列(RabbitMQ及其插件环境安装点此),具体实践如

Rabbitmq 实现延时任务

1.需要用到插件 rabbitmq_delayed_message_exchange 来实现,插件下载地址:https://www.rabbitmq.com/community-plugins.html 2.下载后把插件放到 plugins 里面,然后到 sbin里面打开cmd,执行 rabbitmq-plugins enable rabbitmq_delayed_message_exchange 命令 3.插件装好后,重新启动mq,然后集成mq. 首先,导包 <dependency> <

Go RabbitMQ 死信消息队列(二)

实现原理: /** (1)创建一个正常的队列 Q1,目的是处理业务逻辑,比如发送订单消息等 ,对应交换器和绑定键  分别为  E1 和  Bingkey1 (2)创建一个延时消息队列 Q2,设定队列的延时时间为10s,对应的交换器和绑定键分别为 E2和Bingkey2;并在该队列创建时候,设定队列的  (a)超时时间 (b) 超时后跳转的 路由E1和绑定Bingkey1,即超时后跳到     队列Q1上 (3) 将消息先发送到 队列Q2 上,然后等着队列超时,执行逻辑 * 主要测试一个死信队列,