RocketMQ-没有消费者的消息堆积场景分析

本文涉及的产品
可观测可视化 Grafana 版,10个用户账号 1个月
可观测监控 Prometheus 版,每月50GB免费额度
云原生网关 MSE Higress,422元/月
简介: RocketMQ-没有消费者的消息堆积场景分析

背景介绍

前面几篇文章分析了几个引起消息堆积的典型场景,分别是:

这次的消息堆积场景之前没有遇到过,记录下来以备忘。

问题描述

分析过程

初步判断

为了便于表达和理解,我们只关注与该问题有关的部分逻辑。

因为消息堆积量不断在增加,所以判断该Group ID已经在Broker上有了订阅关系,很可能是使用该Group ID的Consumer实例下线后没有取消订阅关系导致的,如图:

正常运行

在正常情况下,控制台上可以看到Group ID的【订阅关系】及【消费者状态】,如图:

异常之后

异常之后就变成了【问题描述】中的样子,此时我们不清楚:

  • 该GID订阅了哪个topic
  • 该GID被哪个应用消费者使用后出现的异常
  • 该GID对应的消息生产者是哪个

在以上事情没有弄清楚之前,也不敢对该GID做取消订阅、删除之类的操作。

确定topic

消息堆积是通过消费者的offset信息统计的,该信息存储在Broker上的store/config/consumerOffset.json中,consumerOffset.json格式如图:

我们在consumerOffset.json文件中找到了GID对应的topic,此处有个细节(后面代码处有解释):

  • 该GID在groupTopicMap中没有重试队列Topic
  • 该GID在offsetTable中没有重试队列Topic上的offset

确定Producer

通过Topic查询Message

通过MessageID确定ECS IP

通过上面的查询无法直接定位到ECS,我们可以通过Message ID计算出ECS IP,方法如下:

String ip = MessageClientIDSetter.getIPStrFromID(Message ID)

如果懒得写代码,也可以使用arthas来查询:

此时整个链路逐渐清晰起来了,还缺少最关键的Consumer信息。

确定Consumer

代码Review

查询了近期发版的所有代码,没有找到与该GID相关的信息。

Broker端找线索

我们试图通过Broker端的日志来确认两件事情:

  • 该GID的Consumer在什么时候从哪些IP建立了与Broker的交互
  • 该GID的Consumer在什么时候从哪些IP断开了与Broker的交互

Broker heartBeat

通过以上代码打印的日志,我们可以过滤出该GID与Broker建立交互时候的相关信息。

Broker unregisterClient

在Consumer实例shutdown的时候,会向Broker发送unregisterClient请求,会调用ConsumerManager中相应的unregisterConsumer方法:

通过以上代码打印的日志,我们可以过滤出该GID与Broker断开交互时候的相关信息。

理想是美好的,现实是残酷的

Broker端最多保留了不到2天的日志,所以这条路也走不通了。

柳暗花明

同时我们也在想:除了程序,还有其他途径变更这种订阅关系吗?答案是有的。

命令行

控制台

到这里估计您已经知道引起这次消息堆积的原因了。

经验总结

  • 完善监控告警、提高应急响应能力
  • 最小权限原则
  • RocketMQ控制台是否应该增加操作记录的功能?
相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
3月前
|
消息中间件 监控 数据挖掘
基于RabbitMQ与Apache Flink构建实时分析系统
【8月更文第28天】本文将介绍如何利用RabbitMQ作为数据源,结合Apache Flink进行实时数据分析。我们将构建一个简单的实时分析系统,该系统能够接收来自不同来源的数据,对数据进行实时处理,并将结果输出到另一个队列或存储系统中。
211 2
|
1月前
|
消息中间件 前端开发 Java
java高并发场景RabbitMQ的使用
java高并发场景RabbitMQ的使用
76 0
|
3月前
|
消息中间件 存储 负载均衡
我服了,RocketMQ消费者负载均衡内核是这样设计的
文章为理解RocketMQ的负载均衡机制提供了深入的技术洞察,并对如何在实际应用中扩展和定制负载均衡策略提供了有价值的见解。
我服了,RocketMQ消费者负载均衡内核是这样设计的
|
3月前
|
消息中间件 存储 数据中心
RocketMQ的长轮询(Long Polling)实现分析
文章深入分析了RocketMQ的长轮询实现机制,长轮询结合了推送(push)和拉取(pull)两种消息消费模式的优点,通过客户端和服务端的配合,确保了消息的实时性同时将主动权保留在客户端。文中首先解释了长轮询的基本概念和实现步骤,然后通过一个简单的实例模拟了长轮询的过程,最后详细介绍了RocketMQ中DefaultMQPushConsumer的长轮询实现方式,包括PullMessage服务、PullMessageProcessor服务和PullCallback回调的工作原理。
108 1
|
3月前
|
消息中间件 存储 负载均衡
RocketMQ消费者消费消息核心原理(含长轮询机制)
这篇文章深入探讨了Apache RocketMQ消息队列中消费者消费消息的核心原理,特别是长轮询机制。文章从消费者和Broker的交互流程出发,详细分析了Push和Pull两种消费模式的内部实现,以及它们是如何通过长轮询机制来优化消息消费的效率。文章还对RocketMQ的消费者启动流程、消息拉取请求的发起、Broker端处理消息拉取请求的流程进行了深入的源码分析,并总结了RocketMQ在设计上的优点,如单一职责化和线程池的使用等。
RocketMQ消费者消费消息核心原理(含长轮询机制)
|
3月前
|
消息中间件 固态存储 RocketMQ
RocketMQ消息堆积常见场景与处理方案
文章分析了在使用RocketMQ时消息堆积的常见场景,如消费者注册失败或消费速度慢于生产速度,并提供了相应的处理方案,包括提高消费并行度、批量消费、跳过非重要消息以及优化消费代码业务逻辑等。
|
3月前
|
消息中间件 缓存 Java
RocketMQ - 消费者消费方式
RocketMQ - 消费者消费方式
78 0
|
3月前
|
消息中间件 RocketMQ
RocketMQ - 消费者进度保存机制
RocketMQ - 消费者进度保存机制
71 0
|
19天前
|
消息中间件 JSON Java
开发者如何使用轻量消息队列MNS
【10月更文挑战第19天】开发者如何使用轻量消息队列MNS
56 6
|
14天前
|
消息中间件 存储 Kafka
MQ 消息队列核心原理,12 条最全面总结!
本文总结了消息队列的12个核心原理,涵盖消息顺序性、ACK机制、持久化及高可用性等内容。关注【mikechen的互联网架构】,10年+BAT架构经验倾囊相授。

相关产品

  • 云消息队列 MQ