RabbitMq TTL+死信队列 延迟消息问题记录

2023-02-21 20:19:03 来源: 腾讯云

打印 放大 缩小

延迟队列存储的对象是对应的延迟消息,所谓的延迟消息是指当消息被发送以后,并不想让消费者立刻拿到消息,而是等待特定时间后,消费者才能拿到这个消息进行消费

利用RabbitMqTTL和死信队列 来实现延时消费。

如果设置的是队列统一过期时间放到死信队列,没有什么问题。


(资料图片仅供参考)

如果是延时时间设置到每条消息上的。而不是给队列的。

实现方式为消息存活时间为动态用户页面可配置的。

这就导致了一个问题:

先用一条消息的存活时间是1天。后面又进了一条消息存活时间是1小时。

结果一小时到了,发现这条消息并没有被转发到消费延时过期消息的队列。

原因是尽管ttl是设给每条消息的。但是本质上,所有延时消息都还在一个队列里,对它过期时间的检测也是从头部开始的。

它不会检测每一条消息是否过期。而是顺序检测。

如果first in的消息过期时间很长,会导致它阻塞后进的消息。

不仅无法实现真正的过期时间。还会导致,一个大的过期时间的先进的消息,会堆积一堆后进的过期时间短的消息。

问题解决

这个时候可以使用rabbitMq的一个插件:rabbitmq_delayed_message_exchange

一段时间以来,人们一直在寻找用RabbitMQ实现延迟消息的传递方法,到目前为止,公认的解决方案是混合使用TTL和DLX。而rabbitmq_delayed_message_exchange插件就是基于此来实现的,RabbitMQ延迟消息插件新增了一种新的交换器类型,消息通过这种交换器路由就可以实现延迟发送

插件安装

需要根据自己的rabbitMq选择对应的版本。我rabbitMq的版本是RabbitMQ 3.11.0,对应的插件版本就是:3.11.1

基于Linux

--1、cd到rabbitmq默认安装位置cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.7.18/plugins--2、通过ftp工具将插件上传到此目录下--3、开启插件rabbitmq-plugins enable rabbitmq_delayed_message_exchange--4、重启MQ服务systemctl restart rabbitmq-server

基于Docker

--1、通过ftp工具将插件上传到Linux服务器的根目录下--2、拷贝到docker中rabbitmq插件目录下,rabbitmq_delayed_message_exchange-3.9.0.ez(下载包的全名)docker cp /rabbitmq_delayed_message_exchange-3.9.0.ez 容器ID:/plugins--3、进入容器docker exec -it 容器id /bin/bash--4、查看插件是否存在(确保2中的操作已经将插件拷贝过来了)cd pluginsls |grep delay--5、开启插件rabbitmq-plugins enable rabbitmq_delayed_message_exchange--6、退出容器exit--7、重启MQ服务docker restart 容器ID

安装成功

web界面新建交换机选择类型出现红框标注即表示成功

image.png

代码实现

1:springBoot配置

@Configurationpublic class DelayRabbitmqConfig {     /**     * 声明延迟队列     * @return     */    @Bean    public Queue delayQueue(){        return new Queue(QueueConstant.DelayQueue,                true,false,false);    }     /**     * 声明延迟自定义交换机类型     * @return     */    @Bean    public CustomExchange delayCustomExchange(){        HashMap args = new HashMap<>();//        设置 x-delayed-type 为 direct,当然也可以是 topic 等 发送消息时设置消息头 headers 的 x-delay 属性,即延迟时间,如果不设置消息将会立即投递        args.put("x-delayed-type","direct");        return new CustomExchange(ExchangeConstant.DelayCustomerExchange,                "x-delayed-message",true,false,args);    }     /**     * 绑定延迟交换机和队列     * @return     */    @Bean    public Binding delayQueueAndCustomExchange(){        return BindingBuilder.bind(delayQueue())                .to(delayCustomExchange()).with(RoutingKeyConstant.DelayCustomerRoutingKey).noargs();    }}

springMvc配置

引入依赖:    xmlns:util="http://www.springframework.org/schema/util"    http://www.springframework.org/schema/util    http://www.springframework.org/schema/util/spring-util-4.0.xsd                                                                                                                

代码实现

//消息发送final MessagePostProcessor messagePostProcessor = new MyMessagePostProcessor(Integer.valueOf(ttl.toString()));DisTimingPushDto disTimingPushDto = new DisTimingPushDto();disTimingPushDto.setOrderId(dispense.getOrderId());disTimingPushDto.setPushTime(disDispense.getPushTime());rabbitTemplate.convertAndSend(MsgQueueEnum.TIMING_PUSH.getExchangeName(), MsgQueueEnum.TIMING_PUSH.getQueueName(), disTimingPushDto, messagePostProcessor);//每条消息时间配置import org.springframework.amqp.AmqpException;import org.springframework.amqp.core.Message;import org.springframework.amqp.core.MessagePostProcessor;/** * 延迟消息处理器 Processor * @author king * @date 2022年12月28日 11:14 */public class MyMessagePostProcessor implements MessagePostProcessor {    /**     * 消息延迟时间,单位:毫秒     */    private final Integer TTL;    public MyMessagePostProcessor(final Integer ttl) {        this.TTL = ttl;    }    @Override    public Message postProcessMessage(Message message) throws AmqpException {        message.getMessageProperties().setDelay(TTL);        return message;    }}
编辑: sj004

相关阅读

最近更新

环球讯息:抢抓“风口” 新能源产业布局呈现新动向
加强节能标准更新升级和应用实施
速讯:让时代再次澎湃丨汇龙府营销中心暨样板体验馆盛大开放
广告起诉状代写律师怎么收费
全球即时看!原创组图丨昆明圆通山樱花节:花开春到 人潮涌动
当前热文:货币政策精准有力 传递稳增长信号
环球通讯!和讯个股快报:2023年03月20日 吉林高速(601518),该股K线呈现“好友反攻”形态
全球看热讯:这里开出的火车都有独特标记 每天要写上百个“太太”
环球快消息!应急管理部全面启动2023年度化工园区安全整治提升专家指导服务
全球关注:《中国农村人居环境发展报告(2022)》发布
最新消息:“新疆第一春”托克逊县已是杏花飘香
每日讯息!湖北建始举办首届乡村振兴电商直播大赛
今日热讯:全面注册制相关业务首次全网测试进展如何?记者探访
世界观点:第七届碳捕集利用与封存国际论坛在京召开
世界短讯!共促消费公平!新经济诚信品牌榜单正式揭晓,这些企业上榜
纺织品服装出口总额连续3年超3000亿美元 中国纺织行业两化融合提速
每日速递:家庭托育迈出关键一步 “带娃难题”破冰之举也有忧
每日快看:压实资本市场“看门人”责任
【天天速看料】广西2023年前两月实际使用外资额同比增长76.12%
【世界新视野】四川铸强新型工业化主引擎
观速讯丨消费者满意度稳中有降
美欧银行业连遭冲击 中国财经人士警示风险
每日播报!五一出境游签证预约激增,想要办理有哪些门道?这些国家最受关注
今日讯!小银杏带动大产业
世界关注:著写城市封面 越秀国际金融中心与湾区共同腾飞发展
世界球精选!安眠药网上商城
每日资讯:朱民:世界经济正在走向滞胀进入新的“三高一低”
世界最新:李迅雷:恢复和扩大消费在实际操作中面临四大困难
天天快播:18万的小法拉利,我劝你别买!
【环球新要闻】买基金不赚钱?可能是理财习惯有问题 专家支招基民如何理性投资