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 …...

深入理解JavaScript设计模式之单例模式
目录 什么是单例模式为什么需要单例模式常见应用场景包括 单例模式实现透明单例模式实现不透明单例模式用代理实现单例模式javaScript中的单例模式使用命名空间使用闭包封装私有变量 惰性单例通用的惰性单例 结语 什么是单例模式 单例模式(Singleton Pattern&#…...

Cinnamon修改面板小工具图标
Cinnamon开始菜单-CSDN博客 设置模块都是做好的,比GNOME简单得多! 在 applet.js 里增加 const Settings imports.ui.settings;this.settings new Settings.AppletSettings(this, HTYMenusonichy, instance_id); this.settings.bind(menu-icon, menu…...
【论文笔记】若干矿井粉尘检测算法概述
总的来说,传统机器学习、传统机器学习与深度学习的结合、LSTM等算法所需要的数据集来源于矿井传感器测量的粉尘浓度,通过建立回归模型来预测未来矿井的粉尘浓度。传统机器学习算法性能易受数据中极端值的影响。YOLO等计算机视觉算法所需要的数据集来源于…...
Java 加密常用的各种算法及其选择
在数字化时代,数据安全至关重要,Java 作为广泛应用的编程语言,提供了丰富的加密算法来保障数据的保密性、完整性和真实性。了解这些常用加密算法及其适用场景,有助于开发者在不同的业务需求中做出正确的选择。 一、对称加密算法…...

用docker来安装部署freeswitch记录
今天刚才测试一个callcenter的项目,所以尝试安装freeswitch 1、使用轩辕镜像 - 中国开发者首选的专业 Docker 镜像加速服务平台 编辑下面/etc/docker/daemon.json文件为 {"registry-mirrors": ["https://docker.xuanyuan.me"] }同时可以进入轩…...

SpringTask-03.入门案例
一.入门案例 启动类: package com.sky;import lombok.extern.slf4j.Slf4j; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.cache.annotation.EnableCach…...
Element Plus 表单(el-form)中关于正整数输入的校验规则
目录 1 单个正整数输入1.1 模板1.2 校验规则 2 两个正整数输入(联动)2.1 模板2.2 校验规则2.3 CSS 1 单个正整数输入 1.1 模板 <el-formref"formRef":model"formData":rules"formRules"label-width"150px"…...
Python Einops库:深度学习中的张量操作革命
Einops(爱因斯坦操作库)就像给张量操作戴上了一副"语义眼镜"——让你用人类能理解的方式告诉计算机如何操作多维数组。这个基于爱因斯坦求和约定的库,用类似自然语言的表达式替代了晦涩的API调用,彻底改变了深度学习工程…...

脑机新手指南(七):OpenBCI_GUI:从环境搭建到数据可视化(上)
一、OpenBCI_GUI 项目概述 (一)项目背景与目标 OpenBCI 是一个开源的脑电信号采集硬件平台,其配套的 OpenBCI_GUI 则是专为该硬件设计的图形化界面工具。对于研究人员、开发者和学生而言,首次接触 OpenBCI 设备时,往…...

Vue ③-生命周期 || 脚手架
生命周期 思考:什么时候可以发送初始化渲染请求?(越早越好) 什么时候可以开始操作dom?(至少dom得渲染出来) Vue生命周期: 一个Vue实例从 创建 到 销毁 的整个过程。 生命周期四个…...