RocketMQ与kafka如何解决消息积压问题？

前言

  消息积压问题简单来说，就是MQ存在了大量没法快速消费完的数据，造成消息积压的原因主要在于“进入的多，消费的少”，或者生产的速度过快，而消费速度赶不上，基于这一问题，我们主要介绍如何通过前期的开发设置去避免出现消息积压的问题。主要介绍两款产品RocketMQ和Kafka的解决方式，以及其差异，本质上的差异就是RocketMQ与Kafka之间的存储结构差异带来的，基本的处理思路还是怎么控制生产流量，并增加消费者的消费速度，以及Broker的扩容。

1.RocketMQ如何解决消息积压问题？

  首先，消息积压可能出现在生产者、Broker或者消费者这三个环节中的任何一个。所以解决积压问题应该从这三个方面入手。比如，生产者发送速度太快，Broker处理不过来，或者消费者消费能力不足，都会导致积压。那RocketMQ有哪些机制来处理这些情况呢？
  其中，RocketMQ很多的设置理念都是来自Kafka，RocketMQ同样也有分区的概念。
记得RocketMQ有分区的概念，也就是Topic分成多个MessageQueue，这样可以并行处理。如果消费者数量不够，导致处理速度慢，可能需要增加消费者实例，或者调整消费者的线程数，提高并发处理能力。不过消费者的数量不能超过MQ的数量，否则会有空闲的消费者，所以可能需要先扩容。

  所以，RocketMQ解决消息积压问题通常需要从生产者、Broker、消费者 三个环节协同优化，并结合监控、扩容、流量控制等手段。以下是具体的解决方案：

1.1 消费者端优化

(1) 提升消费能力

  • 增加消费者实例：消费者组的实例数（Consumer Instance）应等于或小于订阅的Topic的队列数（MessageQueue）。若队列数不足，需先扩容Topic的队列。
# 修改 Topic 的队列数（需提前规划或动态支持）
mqadmin updateTopic -n localhost:9876 -t YourTopic -c DefaultCluster -w 32
  • 提高并发线程数：调整消费者的 consumeThreadMin 和 consumeThreadMax，增加并发消费线程。
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("GROUP");
consumer.setConsumeThreadMin(20);
consumer.setConsumeThreadMax(64);

(2) 批量消费

  • 若业务允许，开启批量消费模式，一次拉取多条消息处理。java代码处理数据如下所示。
consumer.setConsumeMessageBatchMaxSize(32); // 每次最多消费32条

(3) 异步消费
避免在消费者代码中执行耗时操作（如同步数据库写入），改用异步处理或写入缓冲队列。

1.2. Broker 端优化

(1) 扩容 Broker 和队列
增加 Broker 节点，提升 Topic 的队列数（MessageQueue），分散消息存储和消费压力。

# 动态创建新队列（需Broker支持）
mqadmin updateTopic -n localhost:9876 -t YourTopic -c DefaultCluster -w 64

(2) 调整刷盘策略
异步刷盘（ASYNC_FLUSH）相比同步刷盘（SYNC_FLUSH）可大幅提高 Broker 吞吐量，但需容忍宕机时少量数据丢失。

# Broker配置文件：flushDiskType=ASYNC_FLUSH

(3) 开启Slave读权限
若集群部署，允许消费者从 Slave 节点读取消息，分担负载。

# Broker配置文件：brokerPermission=2（Slave可读）

1.3. 生产端限流

若积压由生产速度过快导致，可通过以下方式限流：

  • 降低生产者发送速率：在代码中控制发送频率或批量大小。

  • RocketMQ 流控：利用 Broker 的 sendMessageThreadPoolNums 参数限制生产线程数。

1.4. 消息积压应急处理

(1) 跳过非关键消息
若允许部分消息丢失，可重置消费位点（Offset）到最新位置，跳过积压消息。

mqadmin resetOffsetByTime -n localhost:9876 -g GROUP -t YourTopic -s now

(2) 临时消费者组

  • 创建临时消费者组，并行消费积压消息，处理完成后下线。
    (3) 消息转发
  • 将积压消息转发到新 Topic，启动额外消费者处理。

1.5. 监控与预警

1.监控指标

  • 消息堆积量（MSG_BACKLOG）。
  • 消费 TPS（CONSUME_TPS）与生产 TPS（PRODUCE_TPS）的差值。
  • 消费延迟（CONSUME_LAG）。
    1.工具
  • RocketMQ Dashboard。
  • Prometheus + Grafana 集成监控。

1.6. 预防措施

  • 合理设计队列数：根据业务峰值提前规划 Topic 的队列数。
  • 消费者熔断机制：在消费异常时暂停消费，避免雪崩。
  • 消息过期策略：设置消息存活时间（TTL），自动清理过期消息。

小结

解决消息积压的核心思路是：

  • 提升消费能力（扩容消费者、优化代码）。
  • 分散压力（扩容Broker和队列）。
  • 限流生产。
  • 应急处理（重置Offset或临时扩容）。
  • 通过监控系统提前预警，结合业务场景选择最优方案。

2.Kafka如何解决消息积压问题？

  Kafka 解决消息积压问题的核心思路是提升消费能力、优化生产与存储、应急处理，需结合Kafka的分区机制、消费者组模型和水平扩展特性。

2.1. 消费者端优化

(1) 增加消费者实例

  • Kafka 的分区（Partition）是并行消费的最小单位，消费者组的实例数 ≤ 分区数。若消费能力不足：
    1）扩容分区（需提前规划，分区数只能增加不能减少）：
kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic YourTopic --partitions 32

2）增加消费者实例：启动新消费者实例加入同一消费者组，自动触发分区重平衡（Rebalance）。

(2) 提高消费吞吐量

  • 调整消费者参数：
# 单次拉取最大数据量（默认1MB）
fetch.max.bytes=10485760  # 10MB
# 单次拉取最大消息数
max.poll.records=1000
# 消费者处理消息的超时时间（避免因处理慢导致Rebalance）
max.poll.interval.ms=300000
# 自动提交Offset间隔（确保处理完再提交）
enable.auto.commit=false  # 改为手动提交
  • 异步批量处理：使用多线程或异步框架（如 Reactor、Vert.x）加速消息处理。

(3) 优化消费逻辑
避免同步阻塞操作（如调用外部 API），改用异步非阻塞处理。
使用本地缓存或批处理减少数据库/网络请求（如攒批写入数据库）。

2.2 Broker端优化

(1) 扩容 Broker 和分区

  • 增加 Broker 节点，提升集群整体吞吐量。
  • 提前规划分区数，确保分区足够支持消费者水平扩展。
    (2) 调整 Broker 参数
  • 提高吞吐配置：
# Broker 处理请求的线程数
num.network.threads=8
num.io.threads=16
# 刷盘策略（吞吐优先）
log.flush.interval.messages=100000  # 异步刷盘
# 日志段保留时间（避免磁盘爆满）
log.retention.hours=72

(3) 优化存储

  • 使用高性能磁盘（如 SSD）。
  • 监控磁盘 IO，避免因磁盘瓶颈导致 Broker 性能下

2.3. 生产端限流

(1) 控制生产速率

  • 在 Producer 代码中限制发送速率：
Properties props = new Properties();
props.put("max.block.ms", 1000);      // 发送缓冲区满时阻塞时间
props.put("linger.ms", 100);          // 消息发送延迟（批量发送）
props.put("batch.size", 16384);       // 批量大小（字节）

(2) 动态分区选择

  • 自定义分区策略，避免热点分区导致单个分区积压。

2.4. 消息积压应急处理

  • 跳过积压数据（慎用，可能丢失消息）：
# 将消费者组的 Offset 重置到最新位置
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \--group YourGroup --reset-offsets --to-latest --topic YourTopic --execute

(2) 临时消费者组

  • 创建新的消费者组，并行消费积压消息：
# 启动独立消费者，指定新的 group.id
kafka-console-consumer.sh --bootstrap-server localhost:9092 \--topic YourTopic --group EmergencyGroup --from-beginning

(3) 消息转储

  • 将积压消息导出到其他存储（如 HDFS、数据库），后续离线处理：
kafka-console-consumer.sh --bootstrap-server localhost:9092 \--topic YourTopic --group DumpGroup --from-beginning > /data/backup.txt

2.5. 监控与诊断
(1) 关键监控指标

  • 消费延迟（Consumer Lag）：消费者当前 Offset 与最新 Offset 的差值。
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group YourGroup
  • 生产/消费 TPS：通过 JMX 或监控工具（如 Prometheus + Grafana）实时跟踪。

(2) 工具

  • Kafka Manager：可视化监控集群状态、分区分布、消费延迟。
  • Burrow：专门监控 Consumer Lag，支持自动告警。

2.6. 预防措施

(1) 容量规划

  • 根据业务峰值提前评估分区数、Broker 节点数和磁盘容量。
  • 设置合理的消息保留时间（log.retention.hours），定期清理旧数据。

(2) 消费者容错

  • 捕获消费异常，避免单条消息阻塞整个消费者。
  • 实现死信队列（DLQ），将处理失败的消息单独存储。
    (3) 流量控制
  • 生产端启用限流（如 Token Bucket 算法）。
  • 消费端通过背压（Backpressure）机制动态调整拉取速率。

小结

  1.Kafka 解决积压的核心方法：
  2.提升消费并行度：增加分区和消费者实例。
  3.优化消费逻辑：异步处理、批量操作。
  4.应急处理：重置 Offset、临时消费者组。
  5.监控预警：实时跟踪 Consumer Lag。
  6.与 RocketMQ 不同，Kafka 的分区机制和消费者组模型更依赖水平扩展能力，需提前规划分区数并动态调整资源。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/diannao/69931.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

RISC-V平台编译 state-thread x264 ffmpeg zlog

RISC-V平台编译 state-thread x264 ffmpeg zlog

1.state-threads 源码下来之后 直接 make linux-debug 目录下生成了对应的.a 和 .h文件 gcc test.c -o test -l st -L . #include <stdio.h> #include <stdlib.h> #include <string.h> #include <errno.h> #include <sys/socket.h&g…
阅读更多...
用easyExcel如何实现？

用easyExcel如何实现？

要使提供的 ExcelModelListener 类来解析 Excel 文件并实现批量存储数据库的功能&#xff0c;需要结合 EasyExcel 库来读取 Excel 数据。具体来说&#xff0c;可以使用 EasyExcel.read() 方法来读取 Excel 文件&#xff0c;并指定 ExcelModelListener 作为事件监听器。 下面是…
阅读更多...
BUU37 [DASCTF X GFCTF 2024｜四月开启第一局]web1234【代码审计/序列化/RCE】

BUU37 [DASCTF X GFCTF 2024｜四月开启第一局]web1234【代码审计/序列化/RCE】

Hint1&#xff1a;本题的 flag 不在环境变量中 Hint2&#xff1a;session_start&#xff08;&#xff09;&#xff0c;注意链子挖掘 题目&#xff1a; 扫描出来www.zip class.php <?phpclass Admin{public $Config;public function __construct($Config){//安全获取基…
阅读更多...
Mysql中使用sql语句生成雪花算法Id

Mysql中使用sql语句生成雪花算法Id

&#x1f353; 简介&#xff1a;java系列技术分享(&#x1f449;持续更新中…&#x1f525;) &#x1f353; 初衷:一起学习、一起进步、坚持不懈 &#x1f353; 如果文章内容有误与您的想法不一致,欢迎大家在评论区指正&#x1f64f; &#x1f353; 希望这篇文章对你有所帮助,欢…
阅读更多...
Go框架面试突击！30道高频题解析

Go框架面试突击！30道高频题解析

前言 有粉丝朋友问我能不能整理Go主流框架方面的面试题&#xff0c;安排&#xff01; 这篇文章分享了gRPC、GoFrame、GoZero、GoMicro、GORM、Gin等主流框架的30道面试题和详解。 需要大厂面经的朋友们也可以直接加我好友&#xff0c;私信我。 gRPC 1.gRPC是什么&#xff…
阅读更多...
@JsonRawValue 注解

@JsonRawValue 注解

这里写目录标题 1. 问题2. JsonRawValue 注解说明 1. 问题 在实际开发中我遇到这样一个问题&#xff0c;查询数据库的结果返回的content内容是含有转移符的JSON字符串&#xff0c;但是我需要返回的不包含转移的String字符串。经过我一顿折腾并未发现解决办法&#xff0c;直到J…
阅读更多...
node.js+兰空图床实现随机图

node.js+兰空图床实现随机图

之前博客一直用的公共的随机图API&#xff0c;虽然图片的质量都挺不错的&#xff0c;但是稳定性都比较一般&#xff0c;遂打算使用之前部署的兰空图床&#xff0c;自己弄一个随机图 本文章服务器操作基于雨云——新一代云服务提供商的云服务器进行操作&#xff0c;有兴趣的话可…
阅读更多...
MySQL binlog的三种模式

MySQL binlog的三种模式

MySQL 的 binlog&#xff08;Binary Log&#xff09; 默认的模式是 STATEMENT&#xff0c;即 语句模式。在这种模式下&#xff0c;MySQL 会记录所有对数据库执行的数据修改操作&#xff08;如 INSERT、UPDATE、DELETE 等&#xff09;的 SQL 语句。 1. binlog 的三种模式 MySQL…
阅读更多...
25农村发展研究生复试面试问题汇总 农村发展专业知识问题很全！ 农村发展复试全流程攻略 农村发展考研复试真题汇总

25农村发展研究生复试面试问题汇总 农村发展专业知识问题很全！ 农村发展复试全流程攻略 农村发展考研复试真题汇总

农村发展复试当然有好的建议&#xff01;前提是复试重点面试题背好&#xff01; 你是不是也在为农村发展考研复试发愁&#xff1f;担心自己准备不充分、表现不好&#xff1f;别急&#xff01;今天&#xff0c;学姐——复试面试拿下90分成功上岸的学姐&#xff0c;来给大家分享…
阅读更多...
读取本地excel并生成map，key为第一列，value为第二列

读取本地excel并生成map，key为第一列，value为第二列

添加依赖&#xff1a;在 pom.xml 文件中添加以下依赖&#xff1a; <dependencies><dependency><groupId>org.apache.poi</groupId><artifactId>poi</artifactId><version>5.2.3</version></dependency><dependency&…
阅读更多...
代码随想录算法【Day44】

代码随想录算法【Day44】

Day44 1143.最长公共子序列 class Solution { public:int longestCommonSubsequence(string text1, string text2) {vector<vector<int>> dp(text1.size() 1, vector<int>(text2.size() 1, 0));for (int i 1; i < text1.size(); i) {for (int j 1; …
阅读更多...
【transformers.Trainer填坑】在自定义compute_metrics时logits和labels数据维度不一致问题

【transformers.Trainer填坑】在自定义compute_metrics时logits和labels数据维度不一致问题

问题描述 我在使用 transformers.Trainer 训练我的模型时&#xff0c;我自定义了 compute_loss 函数和compute_metrics函数&#xff0c;我的模型是一个简单的二分类模型。 在自定义 compute_loss 时这样写的&#xff1a; def compute_loss(self, model, inputs, return_outp…
阅读更多...
论文学习记录之《CLR-VMB》

论文学习记录之《CLR-VMB》

目录 一、基本介绍 二、介绍 三、方法 3.1 FWI中的数据驱动方法 3.2 CLR-VMB理论 3.3 注意力块 四、网络结构 4.1 网络架构 4.2 损失函数 五、实验 5.1 数据准备 5.2 实验设置 5.3 训练和测试 5.4 定量分析 5.5 CLR方案的有效性 5.6 鲁棒性 5.7 泛化性 六、讨…
阅读更多...
【STM32】舵机SG90

【STM32】舵机SG90

1.舵机原理 舵机内部有一个电位器&#xff0c;当转轴随电机旋转&#xff0c;电位器的电压会发生改变&#xff0c;电压会带动转一定的角度&#xff0c;舵机中的控制板就会电位器输出的电压所代表的角度&#xff0c;与输入的PWM所代表的角度进行比较&#xff0c;从而得出一个旋转…
阅读更多...
算法刷题-链表系列-移除链表、设计链表、翻转列表

算法刷题-链表系列-移除链表、设计链表、翻转列表

题目要求 所有主要考察对链表的增删查改的功能 总结 对于有些从头遍历到尾的方法&#xff0c;创建一个头结点使得所有的结点能以统一的方式且全部被遍历到&#xff0c;不会出现头结点不被遍历的问题。对于遍历的条件&#xff0c;有的时候curNode ! nullptr&#xff0c;有的时…
阅读更多...
Django项目中创建app并快速上手（pycharm Windows）

Django项目中创建app并快速上手（pycharm Windows）

1.打开终端 我选择的是第二个 2.运行命令 python manage.py startapp 名称 例如&#xff1a; python manage.py startapp app01 回车&#xff0c;等待一下&#xff0c;出现app01的文件夹说明创建成功 3.快速上手 1.app注册 增加一行 "app01.apps.App01Config"&#…
阅读更多...
Windows系统安装搭建悟空crm客户管理系统 教程

Windows系统安装搭建悟空crm客户管理系统 教程

1、在安装悟空 CRM 之前&#xff0c;需要确保你的 Windows 系统上已经安装了以下软件&#xff1a; Web 服务器&#xff1a;推荐使用 Apache 或 Nginx&#xff0c;这里以 Nginx 为例。你可以使用集成环境套件如 XAMPP 来简化安装过程&#xff0c;它包含了 Nginx 、MySQL、PHP 等…
阅读更多...
深度学习框架探秘｜TensorFlow vs PyTorch：AI 框架的巅峰对决

深度学习框架探秘｜TensorFlow vs PyTorch：AI 框架的巅峰对决

在深度学习框架中&#xff0c;TensorFlow 和 PyTorch 无疑是两大明星框架。前面两篇文章我们分别介绍了 TensorFlow&#xff08;点击查看&#xff09; 和 PyTorch&#xff08;点击查看&#xff09;。它们引领着 AI 开发的潮流&#xff0c;吸引着无数开发者投身其中。但这两大框…
阅读更多...
java每日精进 2.13 Ganache（区块链本地私有化部署）

java每日精进 2.13 Ganache（区块链本地私有化部署）

需求&#xff1a;使用区块链实现数据村存储&#xff0c;记录一些不可篡改的交互信息&#xff0c;网络环境为内外网均需要部署&#xff1b; 1.准备工作&#xff08;软件安装&#xff09; 1.1 安装 Node.js 和 npm 1.2 安装 Ganache 地址如下&#xff1a;windows有可视化界面 &a…
阅读更多...
w206基于Spring Boot的农商对接系统的设计与实现

w206基于Spring Boot的农商对接系统的设计与实现

&#x1f64a;作者简介&#xff1a;多年一线开发工作经验&#xff0c;原创团队&#xff0c;分享技术代码帮助学生学习&#xff0c;独立完成自己的网站项目。 代码可以查看文章末尾⬇️联系方式获取&#xff0c;记得注明来意哦~&#x1f339;赠送计算机毕业设计600个选题excel文…
阅读更多...
最新文章

Copyright @ 2022~2023