异常现象

监控日志展示如下：

[2019-10-30 14:31:23.339 INFO ] [ConsumeMessageThread_7] (com.xxx.service.mq.MQConsumerService:93) - 消费消息：msgId=0A064C3E000179C63692734B339201B0 topic=topic_xxx tag=yyy reconsumeTimes=12

reconsumeTimes 代表消费重试次数。 同时日志中频繁显示同一条msgId。

排查原因

通过debug发现在执行业务代码时，抛出的异常被mq捕获，如下：

org.apache.rocketmq.client.impl.consumer.ConsumeMessageOrderlyService.ConsumeRequest#run try { this.processQueue.getLockConsume().lock(); if (this.processQueue.isDropped()) { log.warn("consumeMessage, the message queue not be able to consume, because it's dropped. {}", this.messageQueue); break; } status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context); } catch (Throwable e) { log.warn("consumeMessage exception: {} Group: {} Msgs: {} MQ: {}", // RemotingHelper.exceptionSimpleDesc(e), // ConsumeMessageOrderlyService.this.consumerGroup, // msgs, // messageQueue); hasException = true; } finally { this.processQueue.getLockConsume().unlock(); } if (null == status) { status = ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT; }

当发生异常时messageListener.consumeMessage方法的返回值为null 当status为null时，status会被赋值为ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT状态(即隔一段时间后重试)。

上文造成消息重试的异常信息如下

java.lang.NoSuchMethodError: com.soaclient.dto.xxx.LogQueryDto.getMqTypeIn()Ljava/util/List;

由于该异常为Error级别，而业务代码异常捕获级别为Exception，导致异常没有被捕获。

改进方法

1、修改业务代码异常 2、修改异常捕获级别为throwable级别，并输出异常日志 3、增加重试次数限制，当大于一定次数时，则不再重试。 如果不加重试，针对顺序消费的情况，可能会出现消费被阻塞的情况。 而无序消费，默认最多重试16次。

题外话

当为无序消费时，代码如下：

try { ConsumeMessageConcurrentlyService.this.resetRetryTopic(msgs); if (msgs != null && !msgs.isEmpty()) { for (MessageExt msg : msgs) { MessageAccessor.setConsumeStartTimeStamp(msg, String.valueOf(System.currentTimeMillis())); } } status = listener.consumeMessage(Collections.unmodifiableList(msgs), context); } catch (Throwable e) { log.warn("consumeMessage exception: {} Group: {} Msgs: {} MQ: {}", RemotingHelper.exceptionSimpleDesc(e), // ConsumeMessageConcurrentlyService.this.consumerGroup, msgs, messageQueue); hasException = true; } if (null == status) { log.warn("consumeMessage return null, Group: {} Msgs: {} MQ: {}", ConsumeMessageConcurrentlyService.this.consumerGroup, msgs, messageQueue); status = ConsumeConcurrentlyStatus.RECONSUME_LATER; }

参考

消息队列 MQ > 高级特性 > 消息重试 https://help.aliyun.com/document_detail/43490.html?spm=a2c4g.11186623.6.555.35b96c99oNQKPt