一次RocketMQ ons SDK Bug导致消息不断堆积到重试队列的案例分析

本文涉及的产品
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
可观测可视化 Grafana 版,10个用户账号 1个月
简介: 一次RocketMQ ons SDK Bug导致消息不断堆积到重试队列的案例分析

背景介绍

系统运行在专有云,应用运行时环境是EDAS Container( EDAS Container是EDAS 平台 HSF 应用运行的基础容器,EDAS Container 包含 Ali-Tomcat 和 Pandora),消息处理使用的是【ons SDK】,消息消费者使用【PUSH】方式【批量】消费【普通消息】,MessageModel是【CLUSTERING】。

为了解决RocketMQ Producer某个性能问题,对Pandora进行了升级(主要是升级RocketMQ版本)。

下面从技术角度对升级中遇到的问题及分析过程进行总结,积累经验以避免类似问题的发生。

问题描述

Pandora升级完成后,我们在RocketMQ控制台看到【消费者状态】->【实时消息堆积量】有8亿条,而每个Consumer实例堆积量是几十条,如图1:

在【消费者状态】->【连接详情】有消息消费失败的情况,如图2:

在应用服务器ons.log也可以实时查看消息消费的指标信息,如图3:

这部分的统计指标的实现可以查看:org.apache.rocketmq.client.stat.ConsumerStatsManager

分析过程

根据我们前面几篇关于MQ消息堆积的文章,可以知道:

  1. 消息堆积总量与Consumer实例消息堆积量相符的情况下,通常是Consumer消费能力弱导致堆积
  2. 消息堆积总量大,而Consumer实例消息堆积量很小的情况下,通常是消息堆积在了重试队列中

从【问题描述】看,更像是发生了第二种情况导致的消息堆积。

总体思路

消息在重试队列中堆积,说明Consumer实例消费消息的时候出现了某些异常,导致Consumer实例将消息发送到了Broker重试队列中,所以我们分析【哪些地方】调用了【发送消息到Broker重试队列的接口】就基本抓住了这个问题的关键。

通过分析RocketMQ源码,发现主要有两个地方调用了【发送消息到Broker重试队列的接口】:

  1. 一个是org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService的内部类ConsumeRequest
  2. 另一个是org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService中清理过期Message的定时任务(最终是交给每个ProcessQueue来清理各自的Message)

ConsumeRequest

下面贴一下主要的代码,与该问题不相关的代码省略掉了;如果不想看代码,可以跳过此处,查看后面的流程图。

为了便于理解,我们使用流程图来表达下图4中代码主要逻辑,见图5:

分析上面流程及代码,发现ConsumeConcurrentlyContext类的ackIndex变量是分析消息成功与失败的核心变量。

是否业务处理异常?

RocketMQ框架在业务处理类出现下面情况的时候,认为消息消费失败:

  1. 业务处理类返回ConsumeConcurrentlyStatus.RECONSUME_LATER
  2. 业务处理类返回null
  3. 业务处理类抛出异常

通过业务处理类日志可以确定业务没有返回ConsumeConcurrentlyStatus.RECONSUME_LATER的情况;

从代码可以看出,当出现2、3情况的时候,框架会将warn日志打印到ons.log中,通过过滤ons.log中“consumeMessage exception”和“consumeMessage return null”关键词,没有相应的日志记录,所以也不是这两种情况造成的。

备注:

当出现2、3情况的时候,ons.log日志中并没有打印出线程栈信息,如果想具体定位异常产生的位置,可以通过arthas stack命令进行分析。

arthas watch processConsumeResult

既然发送失败消息到Broker重试队列是在processConsumeResult方法调用的,那么我们可以分析下该方法的入参及返回值情况。

watch com.alibaba.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService 
processConsumeResult "{params,returnObj}" "target.consumeGroup=='GID_CID_XXX'" -x 3 -n3

watch正常机器

watch异常机器

通过上面的watch,我们找到了问题最关键的地方,我们用下面的场景来分析下ackIndex不同值的影响。

场景一

  1. 业务处理类批量消费了【8】条数据,消费成功返回:CONSUME_SUCCESS
  2. ackIndex=Integer.MAX_VALUE
  3. RocketMQ框架分析消费成功了【8】条,失败【0】条
  4. 因为都消费成功了,不会将消息发送到Broker重试队列中

场景二

  1. 业务处理类批量消费了【8】条数据,消费成功返回:CONSUME_SUCCESS
  2. ackIndex=0
  3. RocketMQ框架分析消费成功了【1】条,失败【7】条
  4. 因为有【7】条消费失败,所以会将【7】条消费失败的消息发送到Broker重试队列中

arthas watch setAckIndex

既然有地方在修改ackIndex,先验证下我们的判断是否正确。

watch com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext setAckIndex "{params,returnObj}" "params[0]==0"

通过观察,确实有地方在不断将ackIndex的值修改为0。

arthas stack setAckIndex

我们继续定位是什么地方将ackIndex修改为0的。

stack com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext setAckIndex "params[0]==0" -n 2

通过线程栈可知BatchConsumerImpl类调用了ConsumeConcurrentlyContext.setAckIndex方法。

arthas jad BatchConsumerImpl

没有源码的情况下,我们可以使用arthas jad对类进行反编译。

jad com.aliyun.openservices.ons.api.impl.rocketmq.BatchConsumerImpl

ConsumeContext类实例字段acknowledgeIndex默认值是多少呢?如果是0,问题的原因就找到了。

athas jad ConsumeContext

没有源码的情况下,我们可以使用arthas jad对类进行反编译。

jad com.aliyun.openservices.ons.api.ConsumeContext

通过上面代码可以看出,ConsumeContext类实例字段acknowledgeIndex的默认值是0。

ProcessQueue

通过上面的分析,我们已经定位到了问题,ProcessQueue做下简单描述,不做具体分析了。

解决办法

由上面的分析,这个问题属于RocketMQ ons SDK的一个Bug,修复就交给相应的产研团队来fix吧。

经验总结

1-5-10,1分钟发现,5分钟定位,10分钟恢复。

当故障发生的时候,需要【1】最短时间内发现(监控报警是否做好),需要【10】最快的速度恢复(变更管理和预案是否做好),【5】似乎不是最主要的。

参考资料

EDAS基本概念

相关实践学习
RocketMQ一站式入门使用
从源码编译、部署broker、部署namesrv,使用java客户端首发消息等一站式入门RocketMQ。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
26天前
|
消息中间件 物联网 网络安全
MQTT常见问题之调用.net sdk 报“不知道这样的主机”如何解决
MQTT(Message Queuing Telemetry Transport)是一个轻量级的、基于发布/订阅模式的消息协议,广泛用于物联网(IoT)中设备间的通信。以下是MQTT使用过程中可能遇到的一些常见问题及其答案的汇总:
|
3月前
|
消息中间件 存储 NoSQL
RabbitMQ的幂等性、优先级队列和惰性队列
【1月更文挑战第12天】用户对于同一操作发起的一次请求或者多次请求的结果是一致的,不会因为多次点击而产生了副作用。举个最简单的例子,那就是支付,用户购买商品后支付,支付扣款成功,但是返回结果的时候网络异常,此时钱已经扣了,用户再次点击按钮,此时会进行第二次扣款,返回结果成功,用户查询余额发现多扣钱了,流水记录也变成了两条。在以前的单应用系统中,我们只需要把数据操作放入事务中即可,发生错误立即回滚,但是再响应客户端的时候也有可能出现网络中断或者异常等等
230 1
|
2月前
|
消息中间件 监控 Java
Spring Boot中的RabbitMQ死信队列魔法:从异常到延迟,一网打尽【RabbitMQ实战 一】
Spring Boot中的RabbitMQ死信队列魔法:从异常到延迟,一网打尽【RabbitMQ实战 一】
59 0
|
25天前
|
消息中间件 存储 安全
【深入浅出RocketMQ原理及实战】「底层原理挖掘系列」透彻剖析贯穿RocketMQ的消息顺序消费和并发消费机制体系的原理分析
【深入浅出RocketMQ原理及实战】「底层原理挖掘系列」透彻剖析贯穿RocketMQ的消息顺序消费和并发消费机制体系的原理分析
25 0
|
1月前
|
消息中间件 前端开发 算法
【十七】RabbitMQ基础篇(延迟队列和死信队列实战)
【十七】RabbitMQ基础篇(延迟队列和死信队列实战)
40 1
|
2月前
|
消息中间件 存储 Apache
精华推荐 | 【深入浅出RocketMQ原理及实战】「性能原理挖掘系列」透彻剖析贯穿RocketMQ的事务性消息的底层原理并在分析其实际开发场景
事务消息(Transactional Message)是指应用本地事务和发送消息操作可以被定义到全局事务中,要么同时成功,要么同时失败。RocketMQ的事务消息提供类似 X/Open XA 的分布事务功能,通过事务消息能达到分布式事务的最终一致。
359 2
精华推荐 | 【深入浅出RocketMQ原理及实战】「性能原理挖掘系列」透彻剖析贯穿RocketMQ的事务性消息的底层原理并在分析其实际开发场景
|
2月前
|
消息中间件 监控 数据挖掘
兔子的后院奇遇:深入了解RabbitMQ中的死信队列【RabbitMQ 四】
兔子的后院奇遇:深入了解RabbitMQ中的死信队列【RabbitMQ 四】
48 0
|
2月前
|
消息中间件 Docker 容器
docker构建rabbitmq并配置延迟队列插件
docker构建rabbitmq并配置延迟队列插件
35 0
|
2月前
|
消息中间件
rabbitmq动态创建队列
rabbitmq动态创建队列
36 0
|
2月前
|
消息中间件 Oracle Java
【RocketMq】Broker 启动脚本分析
【RocketMq】Broker 启动脚本分析
37 0

相关产品

  • 云消息队列 MQ