(四)RabbitMQ高级特性(消费端限流、利用限流实现不公平分发、消息存活时间、优先级队列
Lison <dreamlison@163.com>, v1.0.0, 2023.06.23
RabbitMQ高级特性(消费端限流、利用限流实现不公平分发、消息存活时间、优先级队列
文章目录
- RabbitMQ高级特性(消费端限流、利用限流实现不公平分发、消息存活时间、优先级队列
- 消费端限流
- 利用限流实现不公平分发
- 消息存活时间
- 优先级队列
消费端限流

之前我们讲过MQ可以对请求进行“削峰填谷”,即通过消费端限流的方式限制消息的拉取速度,达到保护消费端的目的。
1、生产者批量发送消息
@Test
public void testSendBatch() {// 发送十条消息for (int i = 0; i < 10; i++) {rabbitTemplate.convertAndSend("my_topic_exchange", "my_routing", "send message..."+i);}
}
2、消费端配置限流机制
spring:rabbitmq:host: 127.0.0.1port: 5672username: adminpassword: 1233456virtual-host: /listener:simple:# 限流机制必须开启手动签收acknowledge-mode: manual# 消费端最多拉取5条消息消费,签收后不满5条才会继续拉取消息。prefetch: 5
3、消费者监听队列
@Component
public class QosConsumer{@RabbitListener(queues = "my_queue")public void listenMessage(Message message, Channel channel) throws IOException, InterruptedException {// 1.获取消息System.out.println(new String(message.getBody()));// 2.模拟业务处理Thread.sleep(3000);// 3.签收消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);}
}
利用限流实现不公平分发
在RabbitMQ中,多个消费者监听同一条队列,则队列默认采用的轮询分发。但是在某种场景下这种策略并不是很好,例如消费者1处 理任务的速度非常快,而其他消费者处理速度却很慢。此时如果采用公平分发,则消费者1有很大一部分时间处于空闲状态。此时可以 采用不公平分发,即谁处理的快,谁处理的消息多
1、生产者批量发送消息
@Test
public void testSendBatch() {// 发送十条消息for (int i = 0; i < 10; i++) {rabbitTemplate.convertAndSend("my_topic_exchange", "my_routing", "send message..."+i);}
}
端配置不公平分发
spring:rabbitmq:host: 127.0.0.1port: 5672username: adminpassword: 1233456virtual-host: /listener:simple:# 限流机制必须开启手动签收acknowledge-mode: manual# 消费端最多拉取1条消息消费,这样谁处理的快谁拉取下一条消息,实现了不公平分发prefetch: 1
编写两个消费者
@Component
public class UnfairConsumer {// 消费者1@RabbitListener(queues = "my_queue")public void listenMessage1(Message message, Channel channel) throws Exception{//1.获取消息System.out.println("消费者1:"+new String(message.getBody(),"UTF-8"));//2. 处理业务逻辑Thread.sleep(500); // 消费者1处理快//3. 手动签收channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);}// 消费者2@RabbitListener(queues = "my_queue")public void listenMessage2(Message message, Channel channel) throws Exception{//1.获取消息System.out.println("消费者2:"+new String(message.getBody(),"UTF-8"));//2. 处理业务逻辑Thread.sleep(3000);// 消费者2处理慢//3. 手动签收channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);}
}
消息存活时间
RabbitMQ可以设置消息的存活时间(Time To Live,简称TTL), 当消息到达存活时间后还没有被消费,会被移出队列。RabbitMQ 可以对队列的所有消息设置存活时间,也可以对某条消息设置存活时间。
设置队列所有消息存活时间
1、在创建队列时设置其存活时间:
@Configuration
public class RabbitConfig2 {private final String EXCHANGE_NAME="my_topic_exchange2";private final String QUEUE_NAME="my_queue2";// 1.创建交换机@Bean("bootExchange2")public Exchange getExchange2(){return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();}// 2.创建队列@Bean("bootQueue2")public Queue getMessageQueue2(){return QueueBuilder.durable(QUEUE_NAME).ttl(10000) //队列的每条消息存活10s.build();}// 3.将队列绑定到交换机@Beanpublic Binding bindMessageQueue2(@Qualifier("bootExchange2") Exchange exchange,@Qualifier("bootQueue2") Queue queue){return BindingBuilder.bind(queue).to(exchange).with("my_routing").noargs();}
}
2、生产者批量生产消息,测试存活时间
@Test
public void testSendBatch2() throws InterruptedException {// 发送十条消息for (int i = 0; i < 10; i++) {rabbitTemplate.convertAndSend("my_topic_exchange2", "my_routing", "send message..."+i);}
}
设置单条消息存活时间
@Test
public void testSendMessage() {//设置消息属性MessageProperties messageProperties = new MessageProperties();//设置存活时间messageProperties.setExpiration("10000");// 创建消息对象Message message = new Message("send message...".getBytes(StandardCharsets.UTF_8), messageProperties);// 发送消息rabbitTemplate.convertAndSend("my_topic_exchange", "my_routing", message);
}
注意:
1 如果设置了单条消息的存活时间,也设置了队列的存活时间,以时间短的为准。
2 消息过期后,并不会马上移除消息,只有消息消费到队列顶端时,才会移除该消息。
@Test
public void testSendMessage2() {
for (int i = 0; i < 10; i++) {
if (i == 5) {
// 1.创建消息属性
MessageProperties messageProperties = new MessageProperties();
// 2.设置存活时间
messageProperties.setExpiration(“10000”);
// 3.创建消息对象
Message message = new Message((“send message…” +i).getBytes(),messageProperties);
// 4.发送消息
rabbitTemplate.convertAndSend(“my_topic_exchange”, “my_routing”, message);
} else {
rabbitTemplate.convertAndSend(“my_topic_exchange”, “my_routing”, “sendmessage…” + i);
}
}
}
在以上案例中,i=5的消息才有过期时间,10s后消息并没有 马上被移除,但该消息已经不会被消费了,当它到达队列顶 端时会被移除。
优先级队列
假设在电商系统中有一个订单催付的场景,即客户在一段时间内未付款会给用户推送一条短信提醒,但是系统中分为大型商家和小型商家。比如像苹果,小米这样大商家一年能给我们创造很大的利润,所以在订单量大时,他们的订单必须得到优先处理,此时就需要为不同的消息设置不同的优先级,此时我们要使用优先级队列。
1、创建队列和交换机
@Configuration
public class RabbitConfig3 {private final String EXCHANGE_NAME="priority_exchange";private final String QUEUE_NAME="priority_queue";// 1.创建交换机@Bean(EXCHANGE_NAME)public Exchange priorityExchange(){return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();}// 2.创建队列@Bean(QUEUE_NAME)public Queue priorityQueue(){return QueueBuilder.durable(QUEUE_NAME)//设置队列的最大优先级,最大可以设置到255,官网推荐不要超过10,,如果设置太高比较浪费资源.maxPriority(10).build();}// 3.将队列绑定到交换机@Beanpublic Binding bindPriority(@Qualifier(EXCHANGE_NAME) Exchange exchange, @Qualifier(QUEUE_NAME) Queue queue){return BindingBuilder.bind(queue).to(exchange).with("my_routing").noargs();}
}
2、编写生产者
@Test
public void testPriority() {for (int i = 0; i < 10; i++) {if (i == 5) {// i为5时消息的优先级较高MessageProperties messageProperties = new MessageProperties();messageProperties.setPriority(9);Message message = new Message(("send message..." +i).getBytes(StandardCharsets.UTF_8), messageProperties);rabbitTemplate.convertAndSend("priority_exchange", "my_routing", message);} else {rabbitTemplate.convertAndSend("priority_exchange", "my_routing", "send message..." + i);}}
}
3、编写消费者
@Component
public class PriorityConsumer {@RabbitListener(queues = "priority_queue")public void listenMessage(Message message, Channel channel) throws Exception{//获取消息System.out.println(new String(message.getBody()));//手动签收channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);}
}
相关文章:
(四)RabbitMQ高级特性(消费端限流、利用限流实现不公平分发、消息存活时间、优先级队列
Lison <dreamlison163.com>, v1.0.0, 2023.06.23 RabbitMQ高级特性(消费端限流、利用限流实现不公平分发、消息存活时间、优先级队列 文章目录 RabbitMQ高级特性(消费端限流、利用限流实现不公平分发、消息存活时间、优先级队列消费端限流利用限流…...
Vue如何配置eslint
eslint官网: eslint.bootcss.com eslicate如何配置 1、选择新的配置: 2、选择三个必选项 3、再选择Css预处理器 4、之后选择处理器 5、选择是提交的时候就进行保存模式 6、放到独立的配置文件上去 7、最后一句是将自己的数据存为预设 8、配合console不要出现的规则…...
Elasticsearch查询文档
GET查询索引单个文档 GET /索引/_doc/ID GET /ffbf/_doc/123返回结果如下,查到了有数据"found" : true表示 {"_index" : "ffbf","_type" : "_doc","_id" : "123","_version" : 2...
面向对象编程:多态性的理论与实践
文章目录 1. 修饰词和访问权限2. 多态的概念3. 多态的使用现象4. 多态的问题与解决5. 多态的意义 在面向对象编程中,多态是一个重要的概念,它允许不同的对象以不同的方式响应相同的消息。本文将深入探讨多态的概念及其应用,以及在Java中如何实…...
linux:filezilla root密码登陆
问题: 如题 参考: 亚马逊服务器FileZilla登录失败解决办法_亚马逊云 ssh链接秘钥认证不了 ubuntu拒绝root用户ssh远程登录解决办法 总结: vi /etc/ssh/sshd_config,修改配置: PermitRootLogin yes PasswordAuthenticat…...
在nginx上部署nuxt项目
先安装Node.js 我安的18.17.0。 安装完成后,可以使用cmd,winr然cmd进入,测试是否安装成功。安装在哪个盘都可以测试。 测试 输入node -v 和 npm -v,(中间有空格)出现下图版本提示就是完成了NodeJS的安装…...
嵌入式linux通用spi驱动之spidev使用总结
Linux内核集成了spidev驱动,提供了SPI设备的用户空间API。支持用于半双工通信的read和write访问接口以及用于全双工通信和I/O配置的ioctl接口。使用时,只需将SPI从设备的compatible属性值添加到spidev区动的spidev dt ids[]数组中,即可将该SP…...
【Nodejs】Puppeteer\爬虫实践
puppeteer 文档:puppeteer.js中文文档|puppeteerjs中文网|puppeteer爬虫教程 Puppeteer本身依赖6.4以上的Node,但是为了异步超级好用的async/await,推荐使用7.6版本以上的Node。另外headless Chrome本身对服务器依赖的库的版本要求比较高,c…...
Windows Active Directory密码同步
大多数 IT 环境中,员工需要记住其默认 Windows Active Directory (AD) 帐户以外的帐户的单独凭据,最重要的是,每个密码还受不同的密码策略和到期日期的约束,为不同的帐户使用单独的密码会增加用户忘记密码和…...
安科瑞能源物联网以能源供应、能源管理、设备管理、能耗分析的能源流向为主线-安科瑞黄安南
摘要:随着科学技术的发展,我国的物联网技术有了很大进展。为了提升电力抄表服务的稳定性,保障电力抄表数据的可靠性,本文提出并实现了基于物联网的智能电力抄表服务平台,结合云计算、大数据等技术,提供电力…...
FPGA设计时序分析一、时序路径
目录 一、前言 二、时序路径 2.1 时序路径构成 2.2 时序路径分类 2.3 数据捕获 2.4 Fast corner/Slow corner 2.5 Vivado时序报告 三、参考资料 一、前言 时序路径字面容易简单地理解为时钟路径,事实时钟存在的意义是为了数据的处理、传输,因此严…...
spring复习:(52)注解方式下,ConfigurationClassPostProcessor是怎么被添加到容器的?
进入AnnotationConfigApplicationContext的构造方法: 进入AnnotatedBeanDefinitionReader的构造方法: 进入this(registry, getOrCreateEnvironment(registry));代码如下: 进入AnnotationConfigUtils.registerAnnotationConfigProcessors方…...
全国大学生数据统计与分析竞赛2021年【本科组】-B题:用户消费行为价值分析
目录 摘 要 1 任务背景与重述 1.1 任务背景 1.2 任务重述 2 任务分析 3 数据假设 4 任务求解 4.1 任务一:数据预处理 4.1.1 数据清洗 4.1.2 数据集成 4.1.3 数据变换 4.2 任务二:对用户城市分布情况与分布情况可视化分析 4.2.1 城市分布情况可视化分析 4…...
力扣1667. 修复表中的名字
表: Users ------------------------- | Column Name | Type | ------------------------- | user_id | int | | name | varchar | ------------------------- 在 SQL 中,user_id 是该表的主键。 该表包含用户的 ID 和名字。…...
【设计模式】详解观察者模式
文章目录 1、简介2、观察者模式简单实现抽象主题(Subject)具体主题(ConcreteSubject)抽象观察者(Observer)具体观察者(ConcrereObserver)测试: 观察者设计模式优缺点观察…...
用html+javascript打造公文一键排版系统8:附件及标题排版
最近工作有点忙,所 以没能及时完善公文一键排版系统,现在只好熬夜更新一下。 有时公文有包括附件,招照公文排版规范: 附件应当另面编排,并在版记之前,与公文正文一起装订。“附件”二字及附件顺序号用3号黑…...
微服务体系<1>
我们的微服务架构 我们的微服务架构和单体架构的区别 什么是微服务架构 微服务就是吧我们传统的单体服务分成 订单模块 库存模块 账户模块单体模块 是本地调用 从订单模块 调用到库存模块 再到账户模块 这三个模块都是调用的同一个数据库 这就是我们的单体架构微服务 就是…...
M5ATOMS3基础02传感器MPU6886
M5ATOMS3基础01按键 简洁版本 MPU6886是一款6轴IMU单元,具有3轴重力加速度计和3轴陀螺仪。它采用16位ADC,内置可编程数字滤波器和片上温度传感器,并通过I2C接口(地址为0x68)与上位机通信。MPU6886支持低功耗模式&#…...
vue 快速自定义分页el-pagination
vue 快速自定义分页el-pagination template <div style"text-align: center"><el-paginationbackground:current-page"pageObj.currentPage":page-size"pageObj.page":page-sizes"pageObj.pageSize"layout"total,prev,…...
0-虚拟机补充知识
虚拟机克隆 如果想要构建服务器集群,没有必要一台一台的去进行安装,只要通过克隆就可以。 快速获得多台服务器主要有两种方式,分别为:直接拷贝操作和vmware的克隆操作 直接拷贝 将之前安装虚拟机的所有文件进行拷贝࿰…...
使用VSCode开发Django指南
使用VSCode开发Django指南 一、概述 Django 是一个高级 Python 框架,专为快速、安全和可扩展的 Web 开发而设计。Django 包含对 URL 路由、页面模板和数据处理的丰富支持。 本文将创建一个简单的 Django 应用,其中包含三个使用通用基本模板的页面。在此…...
R语言AI模型部署方案:精准离线运行详解
R语言AI模型部署方案:精准离线运行详解 一、项目概述 本文将构建一个完整的R语言AI部署解决方案,实现鸢尾花分类模型的训练、保存、离线部署和预测功能。核心特点: 100%离线运行能力自包含环境依赖生产级错误处理跨平台兼容性模型版本管理# 文件结构说明 Iris_AI_Deployme…...
JavaScript 中的 ES|QL:利用 Apache Arrow 工具
作者:来自 Elastic Jeffrey Rengifo 学习如何将 ES|QL 与 JavaScript 的 Apache Arrow 客户端工具一起使用。 想获得 Elastic 认证吗?了解下一期 Elasticsearch Engineer 培训的时间吧! Elasticsearch 拥有众多新功能,助你为自己…...
23-Oracle 23 ai 区块链表(Blockchain Table)
小伙伴有没有在金融强合规的领域中遇见,必须要保持数据不可变,管理员都无法修改和留痕的要求。比如医疗的电子病历中,影像检查检验结果不可篡改行的,药品追溯过程中数据只可插入无法删除的特性需求;登录日志、修改日志…...
java调用dll出现unsatisfiedLinkError以及JNA和JNI的区别
UnsatisfiedLinkError 在对接硬件设备中,我们会遇到使用 java 调用 dll文件 的情况,此时大概率出现UnsatisfiedLinkError链接错误,原因可能有如下几种 类名错误包名错误方法名参数错误使用 JNI 协议调用,结果 dll 未实现 JNI 协…...
tree 树组件大数据卡顿问题优化
问题背景 项目中有用到树组件用来做文件目录,但是由于这个树组件的节点越来越多,导致页面在滚动这个树组件的时候浏览器就很容易卡死。这种问题基本上都是因为dom节点太多,导致的浏览器卡顿,这里很明显就需要用到虚拟列表的技术&…...
Java + Spring Boot + Mybatis 实现批量插入
在 Java 中使用 Spring Boot 和 MyBatis 实现批量插入可以通过以下步骤完成。这里提供两种常用方法:使用 MyBatis 的 <foreach> 标签和批处理模式(ExecutorType.BATCH)。 方法一:使用 XML 的 <foreach> 标签ÿ…...
深度学习之模型压缩三驾马车:模型剪枝、模型量化、知识蒸馏
一、引言 在深度学习中,我们训练出的神经网络往往非常庞大(比如像 ResNet、YOLOv8、Vision Transformer),虽然精度很高,但“太重”了,运行起来很慢,占用内存大,不适合部署到手机、摄…...
nnUNet V2修改网络——暴力替换网络为UNet++
更换前,要用nnUNet V2跑通所用数据集,证明nnUNet V2、数据集、运行环境等没有问题 阅读nnU-Net V2 的 U-Net结构,初步了解要修改的网络,知己知彼,修改起来才能游刃有余。 U-Net存在两个局限,一是网络的最佳深度因应用场景而异,这取决于任务的难度和可用于训练的标注数…...
渗透实战PortSwigger靶场:lab13存储型DOM XSS详解
进来是需要留言的,先用做简单的 html 标签测试 发现面的</h1>不见了 数据包中找到了一个loadCommentsWithVulnerableEscapeHtml.js 他是把用户输入的<>进行 html 编码,输入的<>当成字符串处理回显到页面中,看来只是把用户输…...
