19｜案例：如何排查RocketMQ消息消费积压问题？

丁威

你好，我是丁威。
我想，几乎每一位使用过消息中间件的小伙伴，都会在消息消费时遇到消费积压的问题。在处理这类问题时，大部分同学都会选择横向扩容。但不幸的是，这种解决办法治标不治本，到最后问题还是得不到解决。
说到底，消费端出现消息消费积压是一个结果，但引起这个结果的原因是什么呢？在没有弄清楚原因之前谈优化和解决方案都显得很苍白。
这节课，我们就进一步认识一下消费积压和 RocketMQ 的消息消费模型，看看怎么从根本上排查消费积压的问题。
RocketMQ 的消息消费模型在 RocketMQ 消费领域中，判断消费端遇到的瓶颈通常会用到两个重要的指标：Delay 和 LastConsumeTime。
在开源版本的控制台 rocketmq-console 界面中，我们可以查阅消费端的这两个指标：
Delay 指的是消息积压数量，它是由 BrokerOffset（服务端当前最大的逻辑偏移量）减去 ConsumerOffset（消费者消费的当前位点）计算出来的。如果 Delay 值很大，说明消费端遇到了瓶颈。
LastConsumeTime 表示上一次成功消费消息的存储时间。这个值如果很大，同样能说明消费端遇到了瓶颈。如果这个值线上为 1970 年，表示消费者当前消费位点对应的消息在服务端已经过期，被删除了。

公开

同步至部落

取消

完成

0/2000

荧光笔

直线

曲线

笔记

复制

AI

深入了解
翻译
英语
中文简体
中文繁体
法语
德语
日语
韩语
俄语
西班牙语
阿拉伯语
解释
总结

本文介绍了如何排查RocketMQ消息消费积压问题。首先，文章介绍了RocketMQ的消息消费模型，包括Delay和LastConsumeTime两个重要指标。然后详细解释了RocketMQ的消息消费处理模型，指出了可能导致消费积压的两个原因。接着，文章提出了限流机制来避免内存溢出，并介绍了如何排查RocketMQ消息消费积压问题，包括使用jstack命令跟踪线程栈，定位消费端慢的具体代码行。最后，总结了消费积压问题的根本原因，并分享了一些小经验。整体来说，本文通过深入分析RocketMQ消息消费积压问题的原因和排查方法，为读者提供了解决该问题的思路和方法。

仅可试看部分内容，如需阅读全部内容，请付费购买文章所属专栏
《中间件核心技术与实战》，新⼈⾸单¥59

立即购买

登录后留言

全部留言(2)

最新
精选

小杰
因为消费位点是consumer定时发给broker的，而每次发的时候只发最小确认offset，所以有可能消费者挂了又起来，大量消息都得重新消费，老师是这个原因吗？
作者回复: 非常正确，👏👏，虽然rocketmq无法避免消息重复投送，但还是有义务尽量少发生重复推送。
2022-08-04归属地：上海
2
3
Sam Fu
老师可以讲下消费成功或者失败本地以及broker offset的更新机制吗？是每条消息消费完都会更新本地offset还是拉取的一批全部处理完才会更新. 如果这一批有一条消费失败的就全部扔回吗还是只扔消费失败的
作者回复: 好，你这个我分两个问题回答。 1、关于消息的位点提交机制，为了提高降低服务端持久化位点的压力，RocketMQ的位点提交机制是批量+异步方式。具体是消费端在处理完一条消息后，会提交位点，但这个位点提交只是将位点存储在本地内存缓存中，然后定时(默认每隔5s)将本地缓存一次性的提交到Broker，Broker收到位点提交机制后，更新内存中的位点缓存，然后每隔5s持久化到位点存储文件。 2、如果一批失败了是怎么提交，一般是指并发消费，我们可以留意一下MessageListenerConcurrently中consumeMessage这个方法声明：ConsumeConcurrentlyStatus consumeMessage(final List<MessageExt> msgs, final ConsumeConcurrentlyContext context); 一批消息对应一个处理状态，所以如果其中一条消息失败了，这个时候这一批消息都会重试。
2023-02-04归属地：北京

1

收起评论