👉 这是一个或许对你有用的社群
🐱 一对一交流/面试小册/简历优化/求职解惑，欢迎加入「芋道快速开发平台」知识星球。下面是星球提供的部分资料：
👉这是一个或许对你有用的开源项目
国产 Star 破 10w+ 的开源项目，前端包括管理后台 + 微信小程序，后端支持单体和微服务架构。
功能涵盖 RBAC 权限、SaaS 多租户、数据权限、商城、支付、工作流、大屏报表、微信公众号、CRM 等等功能：
  • Boot 仓库：https://gitee.com/zhijiantianya/ruoyi-vue-pro
  • Cloud 仓库：https://gitee.com/zhijiantianya/yudao-cloud
  • 视频教程：https://doc.iocoder.cn
【国内首批】支持 JDK 21 + SpringBoot 3.2.2、JDK 8 + Spring Boot 2.7.18 双版本 
来源：勇哥Java实战
这篇文章，我们聊聊如何应对 RocketMQ 消息堆积。

1 基础概念

消费者在消费的过程中，消费的速度跟不上服务端的发送速度，未处理的消息会越来越多，消息出现堆积进而会造成消息消费延迟。
虽然笔者经常讲：RocketMQ 、Kafka 具备堆积的能力，但是以下场景需要重点关注消息堆积和延迟的问题：
  1. 业务系统上下游能力不匹配造成的持续堆积，且无法自行恢复。
  2. 业务系统对消息的消费实时性要求较高，即使是短暂的堆积造成的消息延迟也无法接受。
基于 Spring Boot + MyBatis Plus + Vue & Element 实现的后台管理系统 + 用户小程序，支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能
  • 项目地址：https://github.com/YunaiV/ruoyi-vue-pro
  • 视频教程：https://doc.iocoder.cn/video/

2 消费原理

客户端使用 Push 模式 启动后，消费消息时，分为以下两个阶段：
  • 阶段一：拉取消息
    客户端通过长轮询批量拉取的方式从 Broker 服务端获取消息，将拉取到的消息缓存到本地缓冲队列中。
    客户端批量拉取消息，常见内网环境下都会有很高的吞吐量，例如：1个单线程单分区的低规格机器（4C8GB）可以达到几万 TPS ，如果是多个分区可以达到几十万 TPS 。所以这一阶段一般不会成为消息堆积的瓶颈。
  • 阶段二：消费消息
    提交消费线程，客户端将本地缓存的消息提交到消费线程中，使用业务消费逻辑进行处理。
    此时客户端的消费能力就完全依赖于业务逻辑的复杂度（消费耗时 ）和消费逻辑并发度 了。如果业务处理逻辑复杂，处理单条消息耗时都较长，则整体的消息吞吐量肯定不会高，此时就会导致客户端本地缓冲队列达到上限，停止从服务端拉取消息。
通过以上客户端消费原理可以看出，消息堆积的主要瓶颈在于本地客户端的消费能力，即消费耗时 和消费并发度 。
想要避免和解决消息堆积问题，必须合理的控制消费耗时和消息并发度，其中消费耗时的优先级高于消费并发度，必须先保证消费耗时的合理性，再考虑消费并发度问题。
基于 Spring Cloud Alibaba + Gateway + Nacos + RocketMQ + Vue & Element 实现的后台管理系统 + 用户小程序，支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能
  • 项目地址：https://github.com/YunaiV/yudao-cloud
  • 视频教程：https://doc.iocoder.cn/video/

3 消费瓶颈

3.1 消费耗时

影响消费耗时的消费逻辑主要分为 CPU 内存计算和外部 I/O 操作，通常情况下代码中如果没有复杂的递归和循环的话，内部计算耗时相对外部 I/O 操作来说几乎可以忽略。
外部 I/O 操作通常包括如下业务逻辑：
  • 读写外部数据库，例如 MySQL 数据库读写。
  • 读写外部缓存等系统，例如 Redis 读写。
  • 下游系统调用，例如 Dubbo 调用或者下游 HTTP 接口调用。
这类外部调用的逻辑和系统容量需要提前梳理，掌握每个调用操作预期的耗时，这样才能判断消费逻辑中I/O操作的耗时是否合理。
通常消费堆积都是由于这些下游系统出现了服务异常、容量限制导致的消费耗时增加。
例如：某业务消费逻辑中需要调用下游 Dubbo 接口 ，单次消费耗时为 20 ms，平时消息量小未出现异常。业务侧进行大促活动时，下游 Dubbo 服务未进行优化，消费单条消息的耗时增加到 200 ms，业务侧可以明显感受到消费速度大幅下跌。此时，通过提升消费并行度并不能解决问题，需要大幅提高下游 Dubbo 服务性能才行。

3.2 消费并发度

绝大部分消息消费行为都属于 IO 密集型，即可能是操作数据库，或者调用 RPC，这类消费行为的消费速度在于后端数据库或者外系统的吞吐量，通过增加消费并行度，可以提高总的消费吞吐量，但是并行度增加到一定程度，反而会下降。
所以，应用必须要设置合理的并行度。如下有几种修改消费并行度的方法：
  • 同一个 ConsumerGroup 下，通过增加 Consumer 实例数量来提高并行度（需要注意的是超过订阅队列数的 Consumer 实例无效）。可以通过加机器，或者在已有机器启动多个进程的方式。
  • 提高单个 Consumer 实例的消费并行线程，通过修改参数 consumeThreadMin、consumeThreadMax 实现。

4 解决策略

当面对消息堆积问题时，我们需要明确到底哪个环节出现问题了，不要慌张，也不要贸然动手。

4.1 确认消息的消费耗时是否合理

首先，我们需要查看消费耗时 ，确认消息的消费耗时是否合理。查看消费耗时一般来讲有两种方式：
1、打印日志
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context)
{

try
 {

for
 (MessageExt messageExt : msgs) {

long
 start = System.currentTimeMillis();

// TODO 业务逻辑
          logger.info(
"MessageId:"
 + messageExt.getMsgId() + 
" costTime:"
 + (System.currentTimeMillis() - start));

        }

return
 ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

     } 
catch
 (Exception e) {

        logger.error(
"consumeMessage error:"
, e);

return
 ConsumeConcurrentlyStatus.RECONSUME_LATER;

     }

}

2、查看消息轨迹
当确定好消费耗时后，可以根据耗时大小，采取不同的措施。
  • 若查看到消费耗时较长，则需要查看客户端 JVM 堆栈信息排查具体业务逻辑，并优化消费逻辑。
  • 若查看到消费耗时正常，则有可能是因为消费并发度不够导致消息堆积，需要逐步调大消费线程或扩容节点来解决。

4.2 查看客户端 JVM 的堆栈

假如消费耗时非常高，需要查看 Consumer 实例 JVM 的堆栈 。
  1. 通过 jps -m 或者 ps -ef | grep java 命令获取当前正在运行的 Java 程序，通过启动主类即可获得应用的进程 pid ;
  2. 通过 jstack pid > stack.log 命令获取线程的堆栈。
  3. 执行以下命令，查看 ConsumeMessageThread 的信息 。
cat stack.log | grep ConsumeMessageThread -A 
10
 --color

常见的异常堆栈信息如下：
  • 示例1：空闲无堆积的堆栈 。
    消费空闲情况下消费线程都会处于 WAITING 状态等待从消费任务队里中获取消息。
  • 示例2：消费逻辑有抢锁休眠等待等情况 。
    消费线程阻塞在内部的一个睡眠等待上，导致消费缓慢。
  • 示例3：消费逻辑操作数据库等外部存储卡住 。
    消费线程阻塞在外部的 HTTP 调用上，导致消费缓慢。

5 总结

客户端使用 Push模式 启动后，消费消息时，分为以下两个阶段：拉取消息和消费消息 。
客户端消费原理可以看出，消息堆积的主要瓶颈在于本地客户端的消费能力，即消费耗时和消费并发度 。
首先分析消费耗时，然后根据耗时大小，采取不同的措施。
  • 若查看到消费耗时较长，则查看客户端堆栈信息排查具体业务逻辑，并优化消费逻辑。
  • 若查看到消费耗时正常，则有可能是因为消费并发度不够导致消息堆积，需要逐步调大消费线程或扩容节点来解决。
欢迎加入我的知识星球，全面提升技术能力。
👉 加入方式，“长按”或“扫描”下方二维码噢：
星球的内容包括：项目实战、面试招聘、源码解析、学习路线。
文章有帮助的话，在看，转发吧。
谢谢支持哟 (*^__^*）
继续阅读
阅读原文