RabbitMQ解决消息积压的方法
目录
减少发送mq的消息体内容
增加消费者数量
批量消费消息
临时队列转移
监控和预警机制
分阶段实施
最后还有一个方法就是开启队列的懒加载
这篇文章总结一下自己知道的解决消息积压得方法。
减少发送mq的消息体内容
像我们没有必要知道一个的中间状态,只需知道一个最终状态就可以了。
发送的消息体只用包含:id和状态等关键信息,不用发送一个完整的对象内容。
消费者收到消息之后,通过id调用原服务再将完整的消息对象内容查询出来即可,最后再进行消费处理。
增加消费者数量
采用动态增加消费者的数量
@Configuration
public class RabbitMQConfig {@Beanpublic SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);// 设置并发消费者数量factory.setConcurrentConsumers(5); // 初始消费者数量factory.setMaxConcurrentConsumers(20); // 最大消费者数量// 动态调整消费者数量factory.setConsumerTagStrategy(queue -> "consumer-" + UUID.randomUUID());return factory;}
}
@Service
public class ConsumerManagerService {@Autowiredprivate RabbitListenerEndpointRegistry registry;public void adjustConsumerCount(String queueName, int count) {MessageListenerContainer container = registry.getListenerContainer(queueName);if (container instanceof SimpleMessageListenerContainer) {SimpleMessageListenerContainer simpleContainer = (SimpleMessageListenerContainer) container;simpleContainer.setConcurrentConsumers(count);}}
}
批量消费消息
@Service
public class BatchMessageConsumer {@RabbitListener(queues = "myQueue", containerFactory = "batchFactory")public void processBatch(List<Message> messages, Channel channel) {try {// 批量处理消息List<MessageDTO> dtos = messages.stream().map(this::convertToDTO).collect(Collectors.toList());// 批量保存到数据库batchSaveToDatabase(dtos);// 获取最后一条消息的deliveryTaglong lastDeliveryTag = messages.get(messages.size() - 1).getMessageProperties().getDeliveryTag();// 批量确认channel.basicAck(lastDeliveryTag, true);} catch (Exception e) {// 批量拒绝handleBatchError(messages, channel);}}
}// 配置批量消费
@Configuration
public class BatchConsumerConfig {@Beanpublic SimpleRabbitListenerContainerFactory batchFactory(ConnectionFactory connectionFactory) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);// 启用批量消费factory.setBatchListener(true);// 批量大小factory.setBatchSize(100);// 批量超时时间factory.setReceiveTimeout(1000L);return factory;}
}
临时队列转移
@Service
public class MessageTransferService {@Autowiredprivate RabbitTemplate rabbitTemplate;public void transferMessages(String sourceQueue, String tempQueue, int batchSize) {while (true) {// 从源队列批量获取消息List<Message> messages = new ArrayList<>();for (int i = 0; i < batchSize; i++) {Message message = rabbitTemplate.receive(sourceQueue);if (message == null) break;messages.add(message);}if (messages.isEmpty()) break;// 转移到临时队列messages.forEach(msg -> rabbitTemplate.send(tempQueue, msg));}}
}// 临时队列的消费者
@Component
public class TempQueueConsumer {@RabbitListener(queues = "#{tempQueue.name}")public void processMessage(Message message) {// 使用更高效的处理方式fastProcessMessage(message);}@Beanpublic Queue tempQueue() {return new Queue("temp-queue-" + UUID.randomUUID(), false, false, true);}
}
监控和预警机制
@Service
@Slf4j
public class QueueMonitorService {@Autowiredprivate RabbitTemplate rabbitTemplate;@Scheduled(fixedRate = 60000) // 每分钟执行一次public void monitorQueueSize() {String queueName = "myQueue";// 获取队列信息Properties properties = rabbitTemplate.execute(channel -> channel.queueDeclarePassive(queueName));// 获取消息数量int messageCount = properties.getMessageCount();// 检查消息堆积if (messageCount > threshold) {// 发送告警sendAlert(queueName, messageCount);// 动态调整消费者adjustConsumers(queueName, messageCount);}}private void adjustConsumers(String queueName, int messageCount) {// 根据消息数量动态调整消费者数量int newConsumerCount = calculateConsumerCount(messageCount);consumerManagerService.adjustConsumerCount(queueName, newConsumerCount);}
}
分阶段实施
@Service
public class MessageHandlingStrategy {public void handleMessageBacklog() {// 1. 首先增加消费者数量adjustConsumerCount();// 2. 如果仍然堆积,启用批量处理if (isStillBacklogged()) {enableBatchProcessing();}// 3. 如果问题持续,使用临时队列if (isEmergency()) {transferToTemporaryQueue();}}
}
最后还有一个方法就是开启队列的懒加载
相关文章:
RabbitMQ解决消息积压的方法
目录 减少发送mq的消息体内容 增加消费者数量 批量消费消息 临时队列转移 监控和预警机制 分阶段实施 最后还有一个方法就是开启队列的懒加载 这篇文章总结一下自己知道的解决消息积压得方法。 减少发送mq的消息体内容 像我们没有必要知道一个的中间状态,只需…...
Android 网络层相关介绍
关注 Android 默认支持的网络管理行为,默认支持的网络服务功能。 功能术语 术语缩写全称释义DHCPv6Dynamic Host Configuration Protocol for IPv6动态主机配置协议的第六版,用于在IPv6网络中动态分配IP地址和其他网络配置参数。DNS Domain Name System域名系统。LLALink-Loc…...
2025年第三届“华数杯”国际赛B题解题思路与代码(Matlab版)
问题1:产业关联性分析 在 question1.m 文件中,我们分析了中国主要产业之间的相互关系。以下是代码的详细解读: % 问题1:分析中国主要产业之间的相互关系function question1()% 清空工作区和命令窗口clear;clc;% 设置中文显示set…...

小米路由器IPv6 功能使用指南
本文不限于多层路由使用IPv6 的情况,提供解决IPv6 无法获取的更硬核的方法,需要有ssh 工具。(无安卓设备,测试环境win、mac、ios) 首先明确一点,就是如果想让你的设备得到GUA 地址,即访问 6.i…...

k8s dashboard离线部署步骤
确定k8s版本,以1.23为例。 部署metrics-server服务,最好用v0.5.2。 用v0.6.0,可能会报以下错误: nodekubemaster:~/Desktop/metric$ kubectl top nodes Error from server (ServiceUnavailable): the server is currently unabl…...

Wireshark抓包教程(2024最新版个人笔记)
改内容是个人的学习笔记 Wireshark抓包教程(2024最新版)_哔哩哔哩_bilibili 该课程笔记1-16 wireshark基础 什么是抓包工具:用来抓取数据包的一个软件 wireshark的功能:用来网络故障排查;用来学习网络技术 wireshark下…...
稀疏矩阵:BM25;稠密矩阵:RoBERTa - wwm - ext顺序
稀疏矩阵:BM25;稠密矩阵:RoBERTa - wwm - ext顺序 先后顺序 先BM25后RoBERTa - wwm - ext: 流程说明:首先可以使用BM25进行初步的检索。由于BM25是基于词频等统计信息的检索模型,它能够快速地从大规模文档集合中筛选出可能包含相关信息的文档子集。例如,在一个包含大量新…...
C# 结构体(Struct)
C# 结构体(Struct) 引言 在C#编程语言中,结构体(Struct)是一种值类型,它允许用户自定义数据类型。结构体可以包含多个成员,如字段、属性、构造函数和方法。与类(Class)相似,但结构体在内存管理、性能和继承方面有其独特的特点。本文将详细介绍C#结构体的概念、用法…...

Homestyler 和 Tripo AI 如何利用人工智能驱动的 3D 建模改变定制室内设计
让设计梦想照进现实 在Homestyler,我们致力于为每一个梦想设计师提供灵感的源泉,而非挫折。无论是初学者打造第一套公寓,或是专业设计师展示作品集,我们的直观工具都能让您轻松以惊人的3D形式呈现空间。 挑战:实现定制设计的新纪元 我们知道,将个人物品如传家宝椅子、…...

Python的pandas库基础知识(超详细教学)
目录 一、配置环境 二、序列和数据表 2.1 初始化 2.2 获取数值 2.3 获取索引 2.4 索引取内容 2.5 索引改变取值 2.6 字典生成序列 2.7 计算取值出现次数 2.8 数据表 2.9 数据表添加新变量 2.10 获取列名 2.11 根据列名获取数据 2.12 输出固定行 2.13 输出多行…...

【数据库】一、数据库系统概述
文章目录 一、数据库系统概述1 基本概念2 现实世界的信息化过程3 数据库系统内部体系结构4 数据库系统外部体系结构5 数据管理方式 一、数据库系统概述 1 基本概念 数据:描述事物的符号记录 数据库(DB):长期存储在计算机内的、…...
大数据智能选课系统
1.产品介绍 产品名称:大数据智能选课系统 一、产品概述 随着信息技术的快速发展,大数据技术在教育领域的应用越来越广泛。针对当前高校选课过程中的繁琐操作、资源分配不均等问题,我们研发了一款基于大数据智能分析的选课系统。本系统旨在…...

esp32开发笔记之一:esp32开发环境搭建vscode+ubuntu
最近想用esp32做一个物联网项目,踩坑N个终于有点心得,写下来避免和我一样的小白踩无谓的坑。 写在前面: 第一,大家一定要用linux系统作为编译工具,速度上是windows无法比的,不要因为不熟悉linux而选择win…...

赛灵思(Xilinx)公司Artix-7系列FPGA
苦难从不值得歌颂,在苦难中萃取的坚韧才值得珍视; 痛苦同样不必美化,从痛苦中开掘出希望才是壮举。 没有人是绝对意义的主角, 但每个人又都是自己生活剧本里的英雄。滑雪,是姿态优雅的“贴地飞行”,也有着成…...

Trie树算法
Trie树,也称为前缀树或字典树,是一种特殊的树型数据结构。它用于存储一组字符串,使得查找、插入和删除字符串的操作非常高效。类似这种, 模板: 这是用数组来模拟上图中的树的结构,逻辑上和上图结构一致。 …...
NLTK分词以及处理方法
在自然语言处理(NLP)的领域中,文本的处理是一个基础且核心的环节,特别是在大规模数据分析和文本挖掘中。无论是聊天机器人、情感分析,还是机器翻译,分词都是必不可少的步骤之一。分词的目的是将长篇的文本拆解为较小的单位(如单词或句子),这些单位是后续分析和处理的基…...
vue3树形组件+封装+应用
文章目录 概要应用场景代码注释综合评价注意事项功能拓展代码说明概要 创建一个基于Vue 3的树形结构组件,用于展示具有层级关系的数据,并提供了节点展开/折叠、点击等交互功能。以下是对其应用场景、代码注释以及综合评价和注意事项的详细说明。 应用场景 这个组件适用于需…...

kotlin项目无法访问Java类的问题
使用IntelliJ创建一个Kotlin项目,然后在src/main/kotlin中创建一个java接口:Animal.java,然后在Main.kt中打印这个java接口,如下: fun main() {println(Animal::class.java) }代码在编辑器中并没有报错,但…...

计算机网络 (30)多协议标签交换MPLS
前言 多协议标签交换(Multi-Protocol Label Switching,MPLS)是一种在开放的通信网上利用标签引导数据高速、高效传输的新技术。 一、基本概念 MPLS是一种第三代网络架构技术,旨在提供高速、可靠的IP骨干网络交换。它通过将IP地址映…...

qt-C++笔记之自定义继承类初始化时涉及到parents的初始化
qt-C笔记之自定义继承类初始化时涉及到parents的初始化 code review! 参考笔记 1.qt-C笔记之父类窗口、父类控件、对象树的关系 2.qt-C笔记之继承自 QWidget和继承自QObject 并通过 getWidget() 显示窗口或控件时的区别和原理 3.qt-C笔记之自定义类继承自 QObject 与 QWidget …...

Lombok 的 @Data 注解失效,未生成 getter/setter 方法引发的HTTP 406 错误
HTTP 状态码 406 (Not Acceptable) 和 500 (Internal Server Error) 是两类完全不同的错误,它们的含义、原因和解决方法都有显著区别。以下是详细对比: 1. HTTP 406 (Not Acceptable) 含义: 客户端请求的内容类型与服务器支持的内容类型不匹…...
Java 语言特性(面试系列1)
一、面向对象编程 1. 封装(Encapsulation) 定义:将数据(属性)和操作数据的方法绑定在一起,通过访问控制符(private、protected、public)隐藏内部实现细节。示例: public …...

零基础设计模式——行为型模式 - 责任链模式
第四部分:行为型模式 - 责任链模式 (Chain of Responsibility Pattern) 欢迎来到行为型模式的学习!行为型模式关注对象之间的职责分配、算法封装和对象间的交互。我们将学习的第一个行为型模式是责任链模式。 核心思想:使多个对象都有机会处…...

成都鼎讯硬核科技!雷达目标与干扰模拟器,以卓越性能制胜电磁频谱战
在现代战争中,电磁频谱已成为继陆、海、空、天之后的 “第五维战场”,雷达作为电磁频谱领域的关键装备,其干扰与抗干扰能力的较量,直接影响着战争的胜负走向。由成都鼎讯科技匠心打造的雷达目标与干扰模拟器,凭借数字射…...

select、poll、epoll 与 Reactor 模式
在高并发网络编程领域,高效处理大量连接和 I/O 事件是系统性能的关键。select、poll、epoll 作为 I/O 多路复用技术的代表,以及基于它们实现的 Reactor 模式,为开发者提供了强大的工具。本文将深入探讨这些技术的底层原理、优缺点。 一、I…...
python报错No module named ‘tensorflow.keras‘
是由于不同版本的tensorflow下的keras所在的路径不同,结合所安装的tensorflow的目录结构修改from语句即可。 原语句: from tensorflow.keras.layers import Conv1D, MaxPooling1D, LSTM, Dense 修改后: from tensorflow.python.keras.lay…...

Linux nano命令的基本使用
参考资料 GNU nanoを使いこなすnano基础 目录 一. 简介二. 文件打开2.1 普通方式打开文件2.2 只读方式打开文件 三. 文件查看3.1 打开文件时,显示行号3.2 翻页查看 四. 文件编辑4.1 Ctrl K 复制 和 Ctrl U 粘贴4.2 Alt/Esc U 撤回 五. 文件保存与退出5.1 Ctrl …...

(一)单例模式
一、前言 单例模式属于六大创建型模式,即在软件设计过程中,主要关注创建对象的结果,并不关心创建对象的过程及细节。创建型设计模式将类对象的实例化过程进行抽象化接口设计,从而隐藏了类对象的实例是如何被创建的,封装了软件系统使用的具体对象类型。 六大创建型模式包括…...

Visual Studio Code 扩展
Visual Studio Code 扩展 change-case 大小写转换EmmyLua for VSCode 调试插件Bookmarks 书签 change-case 大小写转换 https://marketplace.visualstudio.com/items?itemNamewmaurer.change-case 选中单词后,命令 changeCase.commands 可预览转换效果 EmmyLua…...

Matlab实现任意伪彩色图像可视化显示
Matlab实现任意伪彩色图像可视化显示 1、灰度原始图像2、RGB彩色原始图像 在科研研究中,如何展示好看的实验结果图像非常重要!!! 1、灰度原始图像 灰度图像每个像素点只有一个数值,代表该点的亮度(或…...