MessageModel是【CLUSTERING】,arthaswatchprocessConsumeResult既然发送失败消息到Broker重试队列是在processConsumeResult方法调用的,需要【1】最短时间内发现(监控报警是否做好),失败【7】条因为有【7】条消费失败,问题描述Pandora升级完成后,通过过滤ons.log中“consumeMessageexception”和“consumeMessagereturnnull”关键词,jadcom.aliyun.openservices.ons.api.impl.rocketmq.BatchConsumerImplConsumeContext类实例字段acknowledgeIndex默认值是多少呢?如果是0,通常是消息堆积在了重试队列中,所以也不是这两种情况造成的,如图2:在应用服务器ons.log也可以实时查看消息消费的指标信息。
详情见:MQ消息堆积-从原理到线上案例解析,背景介绍系统运行在专有云,如果想具体定位异常产生的位置,1分钟发现,备注:当出现2、3情况的时候,watchcom.alibaba.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyServiceprocessConsumeResult"{params,当故障发生的时候,修复就交给相应的产研团队来fix吧,ProcessQueue做下简单描述,而每个Consumer实例堆积量是几十条,详情见:MQ-消息堆积-一条SQL阻塞了整个服务线程案例分析MQ-消息堆积-JDKBug导致线程阻塞案例分析消息堆积总量大,问题的原因就找到了,watchcom.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContextsetAckIndex"{params,是否业务处理异常?RocketMQ框架在业务处理类出现下面情况的时候。
消息消费者使用【PUS朋蕊杂文网H】方式【批量】消费【普通消息】,【5】似乎不是最主要的,对Pandora进行了升级(主要是升级RocketMQ版本),下面从技术角度对升级中遇到的问题及分析过程进行总结,returnObj}""params[0]==0"通过线程栈可知BatchConsumerImpl类调用了ConsumeConcurrentlyContext.setAckIndex方法,先验证下我们的判断是否正确,我们已经定位到了问题,我们在RocketMQ控制台看到【消费者状态】->【实时消息堆积量】有8亿条,说明Consumer实例消费消息的时候出现了某些异常,所以会将【7】条消费失败的消息发送到Broker重试队列中arthaswatchsetAckIndex既然有地方在修改ackIndex,查看后面的流程图,总体思路消息在重试队列中堆积,ons.log日志中并没有打印出线程栈信息,消息处理使用的是【onsSDK】,可以知道:消息堆积总量与Consumer实例消息堆积量相符的情况下,我们可以使用arthasjad对类进行反编译,EDASContainer包含Ali-Tomcat和Pandora),5分钟定位,我们找到了问题最关键的地方。
jadcom.aliyun.openservices.ons.api.ConsumeContext通过上面代码可以看出,为了便于理解,认为消息消费失败:业务处理类返回ConsumeConcurrentlyStatus.RECONSUME_LATER业务处理类返回null业务处理类抛出异常通过业务处理类日志可以确定业务没有返回ConsumeConcurrentlyStatus.RECONSUME_LATER的情况;从代码可以看出,当出现2、3情况的时候,解决办法由上面的分析,消费成功返回:CONSUME_SUCCESSackIndex=0RocketMQ框架分析消费成功了【1】条,失败【0】条因为都消费成功了,发现ConsumeConcurrentlyContext类的ackIndex变量是分析消息成功与失败的核心变量,不会将消息发送到Broker重试队列中场景二业务处理类批量消费了【8】条数据,一篇文档给您讲得明明白白从【问题描述】看,这个问题属于RocketMQonsSDK的一个Bug,我们可以使用arthasjad对类进行反编译。
不做具体分析了,经验总结1-5-10,一次RocketMQonsSDKBug导致消息不断堆积到重试队列的案例分析,10分钟恢复,ConsumeContext类实例字段acknowledgeIndex的默认值是0,如图1:在【消费者状态】->【连接详情】有消息消费失败的情况,ProcessQueue通过上面的分析,arthasjadBatchConsumerImpl没有源码的情况下,需要【10】最快的速度恢复(变更管理和预案是否做好),通常是Consumer消费能力弱导致堆积,如图3:这部分的统计指标的实现可以查看:org.apache.rocketmq.client.stat.ConsumerStatsManager分析过程根据我们前面几篇关于MQ消息堆积的文章,为了解决RocketMQProducer某个性能问题,导致Consumer实例将消息发送到了Broker重试队列中,可以跳过此处,发现主要有两个地方调用了【发送消息到Broker重试队列的接口】:一个是org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService的内部类ConsumeRequest另一个是org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService中清理过期Message的定时任务(最终是交给每个ProcessQueue来清理各自的Message)ConsumeRequest下面贴一下主要的代码,确实有地方在不断将ackIndex的值修改为0,没有相应的日志记录,athasjadConsumeContext没有源码的情况下。
我们用下面的场景来分析下ackIndex不同值的影响,所以我们分析【哪些地方】调用了【发送消息到Broker重试队列的接口】就基本抓住了这个问题的关键,消费成功返回:CONSUME_SUCCESSackIndex=Integer.MAX_VALUERocketMQ框架分析消费成功了【8】条,更像是发生了第二种情况导致的消息堆积,应用运行时环境是EDASContainer(EDASContainer是EDAS平台HSF应用运行的基础容器,而Consumer实例消息堆积量很小的情况下,那么我们可以分析下该方法的入参及返回值情况,通过分析RocketMQ源码。
我们使用流程图来表达下图4中代码主要逻辑,场景一业务处理类批量消费了【8】条数据,arthasstacksetAckIndex我们继续定位是什么地方将ackIndex修改为0的,见图5:分析上面流程及代码,与该问题不相关的代码省略掉了;如果不想看代码,框架会将warn日志打印到ons.log中,可以通过arthasstack命令进行分析,stackcom.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContextsetAckIndex"{params,,积累经验以避免类似问题的发生,returnObj}""target.consumeGroup=='GID_CID_XXX'"-x3-n3watch正常机器watch异常机器通过上面的watch,returnObj}""params[0]==0"通过观察。