RabbitMq TTL+死信队列 延迟消息问题记录
延迟队列存储的对象是对应的延迟消息,所谓的延迟消息是指当消息被发送以后,并不想让消费者立刻拿到消息,而是等待特定时间后,消费者才能拿到这个消息进行消费
利用RabbitMq
的TTL
和死信队列 来实现延时消费。
如果设置的是队列统一过期时间放到死信队列,没有什么问题。
(资料图)
如果是延时时间设置到每条消息上的。而不是给队列的。
实现方式为消息存活时间为动态用户页面可配置的。
这就导致了一个问题:
先用一条消息的存活时间是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界面新建交换机选择类型出现红框标注即表示成功
代码实现
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; }}
关键词: RabbitMQ
精心推荐
- 31个省区市公布2023高考分数线|每日短讯
- 汇总!全国31个省份高考分数线公布
- 观天下!黄子韬的老婆叫什么名_黄子韬的老婆
- 热门车讯或20万元起,领克02 PHEV将于成都车展上市!|天天快播报
- 京山轻机:三协人工智能产业园目前尚未竣工 天天新视野
- 环球热资讯!沿河:“五度”助推退役军人工作高质量发展
- 嘉禾“洗澡肉” 名俗味不俗
- 古筝弹奏入门基础曲子有什么_古筝弹奏入门-看热讯
- 环球今头条!中暑都有哪些症状?如何防范中暑?专家解答
- 孟轲乡全力以赴参与创卫攻坚战
- 我的青春恋爱物语果然有问题小说全卷百度云 我的青春恋爱物语果然有问题小说百度云
- 江西第七届全国高校招生咨询会举行 220所高校为考生提供“一站式”服务
- 在河北加快打造全国现代商贸物流重要基地中展现廊坊作为|焦点简讯
- 想清楚这六个问题,选专业就不会选错!
X 关闭
X 关闭
产业
-
不用跑北京 在家门口也能挂上顶...
日前,我省首个神经疾病会诊中心——首都医科大学宣武医院河北医院...
-
“十四五”期间 河北省将优化快...
从省邮政管理局获悉,十四五期间,我省将优化快递空间布局,着力构...
-
张家口市宣化区:光伏发电站赋能...
3月19日拍摄的张家口市宣化区春光乡曹庄子村光伏发电站。张家口市宣...
-
“张同学”商标被多方抢注 涉及...
“张同学”商标被多方抢注,官方曾点名批评恶意抢注“丁真” ...
-
山东济南“防诈奶奶团”花式反诈...
中新网济南12月15日电 (李明芮)“老有所为 无私奉献 志愿服...
-
广州新增1例境外输入关联无症状...
广州卫健委今日通报,2021年12月15日,在对入境转运专班工作人...
-
西安报告初筛阳性病例转为确诊病例
12月15日10:20,经陕西西安市级专家组会诊,西安市报告新冠病毒...
-
广东东莞新增本土确诊病例2例 ...
(抗击新冠肺炎)广东东莞新增本土确诊病例2例 全市全员核酸检测...
-
中缅边境临沧:民警深夜出击捣毁...
中新网临沧12月15日电 (胡波 邱珺珲)记者15日从云南临沧边境...
-
“土家鼓王”彭承金:致力传承土...
中新网恩施12月15日电 题:“土家鼓王”彭承金:致力传承土家...