分享笔记RabbitMQ高级之消息限流与延时队列

tech2023-02-21  102

楔子

本篇是消息队列RabbitMQ的第五弹。

上篇本来打算讲述RabbitMQ的一些高级用法:

如何保证消息的可靠性? 消息队列如何进行限流? 如何设置延时队列进行延时消费?

最终因为篇幅缘故,上篇只讲了如何保证消息的可靠性?,本篇将会把剩下两个讲完,本篇也可能是RabbitMQ系列的最后一篇了~

我所讲的知识点在工作中基本上也够用了,希望大家好好消化。

旧坑填上之后可能会慢慢开新坑了,同时因为现在到九月中旬这段时间我有一场考试需要筹备,所以文章更新可能会比较慢,但是一周一更算是最低限度把,希望大家多多担待。

祝有好收获,先赞后看,快乐无限。

本文代码: 码云地址 GitHub地址

🔍消息队列如何限流?

消息队列限流是指在服务器面临巨额流量时,为了进行自保,进行的一种救急措施。

因为巨大的流量代表着非常多的消息,这些消息如果多到服务器处理不过来就会造成服务器瘫痪,影响用户体验,造成不良影响。

所以要进行一次降级操作,把处理不了的流量隔绝在系统之外,避免它们打垮系统。

基本上任何一个消息队列都有限流的功能,今天我们就来看看在RabbitMQ之中进行限流具体应该怎么做?

RabbitMQ提供了一种QOS(服务质量保证)功能,即在非自动确认消息的前提下,如果一定数目的消息还未被消费确认,则不进行新消息的消费。

spring: rabbitmq: addresses: 127.0.0.1 host: 5672 username: guest password: guest virtual-host: / # 手动确认消息 listener: simple: acknowledge-mode: manual prefetch: 2

我们只需要配置一下rabbitmq.listener.simple下的prefetch属性即可,为了演示方便我这里配置为两条,语义即为:如果队列中有两条以上未签收的消息,则不进行新的消息消费。

我往我的队列中发送三条信息,并不进行签收,来看看效果:

消息限流演示01

发送完显示我们系统中有三条Ready消息,这代表这三条消息还在队列中没有消费端去消费。

这时我打开消费端进行消费但依旧不进行签收,接着来看效果:

消息限流演示02

unacked=2,ready=1,这就代表有两条消息在服务端消费了但是没有签收,还有一条消息还在队列中没有往服务端推送,因为我们设置了prefetch=2,所以现在队列的最大同时在消费的消息数量为2,通过此种方式,我们就完成了消费限流。

Tip : 这种方式下消息一定要进行手动签收,因为之前的文章中我们讲过,自动签收是消息一达到消费端就进行签收了,可能我们的业务逻辑还没运行就已经进行签收了,所以自动签收状态下开启限流几乎没有作用。 2. 📑RabbitMQ控制台

上一节我的截图中,大家可以发现居然出现了可视化的界面,以往在我的截图中一般都是DOS命令操作台界面,但其实RabbitMQ是自带了可视化界面的插件的,我们只需要开启即可。

在我们的RabbitMQ中输入如下命令:rabbitmq-plugins.bat enable rabbitmq_management

就可以开启可视化页面了,紧接着访问:http://localhost:15672/

可视化页面01

默认用户名和密码都是 guest,直接登录即可。

可视化页面02

很方便的控制台,大家可以自己试一下~ 3. 📔TTL消息/队列

TTL是Time To Live的缩写,也就是生存时间的意思,RabbitMQ支持消息的过期时间,在消息发送时可以进行指定,也支持队列的过期时间,从消息入队列开始计算,只要超过了队列的超时时间配置,那么消息会自动的清除。

设置队列的话就是整个队列的消息到时都会过期,设置消息的话就是单条消息到时自动过期。

// TTL队列示例 @Bean public Queue ttlQueue() { Map<String, Object> arguments = new HashMap<>(); // 设置3s过期 arguments.put("x-message-ttl",3000); return new Queue("topicQueue1",false,false,false, arguments); }

上面的代码就是演示如何创建一个TTL队列,需要放入参数才行,队列构造中的其他参数我为了方便直接填了false。

public void sendTtl() { String message = "Hello 我是作者和耳朵,欢迎关注我。" + LocalDateTime.now().toString(); System.out.println("Message content : " + message); // 设置过期3s MessageProperties props = MessagePropertiesBuilder.newInstance() .setExpiration("3000").build(); rabbitTemplate.send(Producer.QUEUE_NAME,new Message(message.getBytes(StandardCharsets.UTF_8),props)); System.out.println("消息发送完毕。"); }

设置消息的TTL也是设置参数即可。

以上就是RabbitMQ中关于TTL的知识点。 4. 📌DLX死信队列 http://www.xytmgc.com DLX死信队列虽然叫队列,但其实指的是Exchange,或者说指的Exchange和它所属的Queue,他俩一块构成了死信队列。

当一条消息:

消费被拒绝(basic.reject/basic.nack)并且requeue=false TTL过期 要进入的队列达到最大长度

这三种情况,就可以判定一条消息死了,这种消息如果我们没有做处理,它就会被自动删除。

但其实我们可以在队列上加上一个参数,使当队列中发现了死亡的消息之后会将它自动转发到某个Exchange,由指定的Exchange来处理这些死亡的消息。

这个处理死亡消息的Exchange和之前我们讲述的Exchange没什么区别,依然可以绑定队列然后进行消息消费。

// DLX队列示例 @Bean public Queue dlxQueue() { Map<String, Object> arguments = new HashMap<>(); // 指定消息死亡后发送到ExchangeName="dlx.exchange"的交换机去 arguments.put("x-dead-letter-exchange","dlx.exchange"); return new Queue("topicQueue1", false, false, false, arguments); }

如上代码,就是设置了一个队列中的消息死亡后的去处,就等于消息死亡后给它不把它删掉而是做一次转发,发到其他Exchange去。

那这样搞有什么用呢?这就取决于业务需求了,不过下一节会用到它,接着往下看~ 5. 💡延时队列

RabbitMQ的基因中没有延时队列这回事,它不能直接指定一个队列类型为延时队列,然后去延时处理,但是经过上面两节的铺垫,我们可以将TTL+DLX相结合,这就能组成一个延时队列。

设想一个场景,下完订单之后15分钟未付款我们就要将订单关闭,这就是一个很经典的演示消费的场景,如果拿RabbitMQ来做,我们就需要结合TTL+DLX了。

先把订单消息设置好15分钟过期时间,然后过期后队列将消息转发给我们设置好的DLX-Exchange,DLX-Exchange再将分发给它绑定的队列,我们的消费者再消费这个队列中的消息,就做到了延时十五分钟消费。http://www.bjhfyjc.com

真是super~~~简单呢 后记

收尾了收尾了,RabbitMQ结束了,虽然有些东西没有讲比如:镜像队列,因为我没用过而且一般轮不到自己来做这个,所以就懒了一下就不写这个了,RabbitMQ毕竟不是一个天生的分布式消息队列,弄镜像什么的还是有点麻烦的。

陆陆续续似乎写了快一个月呢,东西有点多也有些繁杂,要不下期写一篇文章专门回顾一下,再画个思维导图什么的,给大家梳理一下,再抽几个小册六折码。

最后再给优狐打个广告,最近掘金在GitHub上面建立了一个开源计划 - open-source,旨在收录各种好玩的好用的开源库,如果大家有想要自荐或者分享的开源库都可以参与进去,为这个开源计划做一份贡献,同时这个开源库的Start也在稳步增长中,参与进去也可以增加自己项目的曝光度,一举两得。

同时这个开源库还有一个兄弟项目 - open-source-translation,旨在招募技术文章翻译志愿者进行技术文章的翻译工作, 争做最棒开源翻译,翻译业界高质量文稿,为技术人的成长献一份力。

最近这段时间事情挺多,优狐令我八月底之前升级到三级,所以各位读者的赞对我很重要,希望大家能够高抬贵手,帮我一哈~

好了,以上就是本期的全部内容,感谢你能看到这里,欢迎对本文点赞收藏与评论,👍你们的每个点赞都是我创作的最大动力。

我是耳朵,一个一直想做知识输出的伪文艺程序员,我们下期见。

最新回复(0)