Shulou(Shulou.com)06/02 Report--

RocketMQ consumption failure retry mechanism example analysis, many novices are not very clear about this, in order to help you solve this problem, the following editor will explain for you in detail, people with this need can come to learn, I hope you can gain something.

Phenomenon: mq consumes once, retries 3 times, and then stops, as shown in the following example

First time (reconsumeTimes=0)

MQ_CON_MSG gmcf-lsc-zhongbang-repu-calc-from-topic MSG MessageExt [queueId=1, storeSize=453, queueOffset=25, sysFlag=0, bornTimestamp=1566785215908, bornHost=/10.42.0.77:54608, storeTimestamp=1566785215908, storeHost=/10.42.0.244:10911, msgId=0A2A00F400002A9F000000000B77CE84, commitLogOffset=192401028, bodyCRC=53737244, reconsumeTimes=0, preparedTransactionOffset=0, toString () = Message {topic='gmcf-lsc-zhongbang-repu-calc-from-topic', flag=0, properties= {MIN_OFFSET=0, _ catChildMessageId1=gmcf-lsc-job-0a2a004d-435218-15071, HASH_CODE=690132963, MAX_OFFSET=26, _ catRootMessageId=lts-0a2a006b-434483-0 CONSUME_START_TIME=1566785215911, SERIALIZE_CLASS=java.lang.String, _ catParentMessageId=lts-0a2a006b-435218-524, _ catChildMessageId=gmcf-lsc-job-0a2a004d-435218-15072, UNIQ_KEY=0A2A004D000938AF386882EAA5A40112, WAIT=true}, body= [71, 77, 66, 69, 67, 49, 57, 48, 56, 50, 52, 49, 48, 52, 48, 52, 54, 56, 57, 52, 48, 52, 48, 56, 48], transactionId='null'}]

First retry (reconsumeTimes=1,DELAY=3)

MQ_CON_MSG% RETRY%gmcf-lsc-consumerGroup MSG MessageExt [queueId=0, storeSize=608, queueOffset=1187, sysFlag=0, bornTimestamp=1566785215923, bornHost=/10.42.0.77:54608, storeTimestamp=1566785226241, storeHost=/10.42.0.244:10911, msgId=0A2A00F400002A9F000000000B785900, commitLogOffset=192436480, bodyCRC=893293938, reconsumeTimes=1, preparedTransactionOffset=0, toString () = Message {topic='gmcf-lsc-zhongbang-repu-calc-from-topic', flag=0, properties= {_ catChildMessageId1=gmcf-lsc-job-0a2a004d-435218-15075, _ catRootMessageId=lts-0a2a006b-434483-0, CONSUME_START_TIME=1566785226242, SERIALIZE_CLASS=java.lang.String _ catParentMessageId=lts-0a2a006b-435218-15076, MIN_OFFSET=0, REAL_TOPIC=%RETRY%gmcf-lsc-consumerGroup, ORIGIN_MESSAGE_ID=0A2A00F400002A9F000000000B77D049, RETRY_TOPIC=gmcf-lsc-zhongbang-repu-calc-from-topic, HASH_CODE=1506102461, MAX_OFFSET=1188, UNIQ_KEY=0A2A004D000938AF386882EAA5B30113, WAIT=false, DELAY=3, REAL_QID=0}, body= [71, 77, 66, 69, 67, 49, 57, 56, 50, 52, 49, 48, 49, 49, 50, 48, 55, 48, 52, 48 56, 49], transactionId='null'}]

The second retry (reconsumeTimes=2, DELAY=4)

MQ_CON_MSG% RETRY%gmcf-lsc-consumerGroup MSG MessageExt [queueId=0, storeSize=608, queueOffset=1209, sysFlag=0, bornTimestamp=1566785215923, bornHost=/10.42.0.77:54608, storeTimestamp=1566785256680, storeHost=/10.42.0.244:10911, msgId=0A2A00F400002A9F000000000B791399, commitLogOffset=192484249, bodyCRC=893293938, reconsumeTimes=2, preparedTransactionOffset=0, toString () = Message {topic='gmcf-lsc-zhongbang-repu-calc-from-topic', flag=0, properties= {_ catChildMessageId1=gmcf-lsc-job-0a2a004d-435218-15075, _ catRootMessageId=lts-0a2a006b-434483-0, CONSUME_START_TIME=1566785256728, SERIALIZE_CLASS=java.lang.String _ catParentMessageId=lts-0a2a006b-435218-15076, MIN_OFFSET=0, REAL_TOPIC=%RETRY%gmcf-lsc-consumerGroup, ORIGIN_MESSAGE_ID=0A2A00F400002A9F000000000B77D049, RETRY_TOPIC=gmcf-lsc-zhongbang-repu-calc-from-topic, HASH_CODE=1506102461, MAX_OFFSET=1210, UNIQ_KEY=0A2A004D000938AF386882EAA5B30113, WAIT=false, DELAY=4, REAL_QID=0}, body= [71, 77, 66, 69, 67, 49, 57, 56, 50, 52, 49, 48, 49, 49, 50, 48, 55, 48, 52, 48 56, 49], transactionId='null'}]

The third retry (reconsumeTimes=3, DELAY=5)

MQ_CON_MSG% RETRY%gmcf-lsc-consumerGroup MSG MessageExt [queueId=0, storeSize=608, queueOffset=1228, sysFlag=0, bornTimestamp=1566785215923, bornHost=/10.42.0.77:54608, storeTimestamp=1566785316978, storeHost=/10.42.0.244:10911, msgId=0A2A00F400002A9F000000000B79F598, commitLogOffset=192542104, bodyCRC=893293938, reconsumeTimes=3, preparedTransactionOffset=0, toString () = Message {topic='gmcf-lsc-zhongbang-repu-calc-from-topic', flag=0, properties= {_ catChildMessageId1=gmcf-lsc-job-0a2a004d-435218-15075, _ catRootMessageId=lts-0a2a006b-434483-0, CONSUME_START_TIME=1566785316980, SERIALIZE_CLASS=java.lang.String _ catParentMessageId=lts-0a2a006b-435218-15076, MIN_OFFSET=0, REAL_TOPIC=%RETRY%gmcf-lsc-consumerGroup, ORIGIN_MESSAGE_ID=0A2A00F400002A9F000000000B77D049, RETRY_TOPIC=gmcf-lsc-zhongbang-repu-calc-from-topic, HASH_CODE=1506102461, MAX_OFFSET=1231, UNIQ_KEY=0A2A004D000938AF386882EAA5B30113, WAIT=false, DELAY=5, REAL_QID=0}, body= [71, 77, 66, 69, 67, 49, 57, 56, 50, 52, 49, 48, 49, 49, 50, 48, 55, 48, 52, 48 56, 49], transactionId='null'}]

According to the phenomenon, we ask two questions?

1. Why only retry 4 times? Instead of trying again all the time?

Try {try {if (messageExtWrappers.size) > 0) {try {var22 = messageExtWrappers.iterator (); while (var22.hasNext ()) {messageExt = (MessageExt) var22.next () Span.addEvent ("MQConsumer.from", messageExt.getBornHostString ());} catch (Throwable var16) {;} this.consume (messageExtWrappers, context) } LOGGER.info ("MQ_CON_SUCCESS {} BROKER {} QUEUE {}", new Object [] {topic, broker, queueId}); span.addEvent ("MQConsumer", ConsumeConcurrentlyStatus.CONSUME_SUCCESS.name ()); span.success (); ConsumeConcurrentlyStatus var23 = ConsumeConcurrentlyStatus.CONSUME_SUCCESS; return var23 } catch (MessageListenerConcurrentlyException var17) {LOGGER.error ("MQ_CON_EX {} BROKER {} QUEUE {}", new Object [] {topic, broker, queueId, var17}); throw var17;} catch (Throwable var18) {LOGGER.error ("MQ_CON_EX {} BROKER {} QUEUE {}", new Object [] {topic, broker, queueId, var18}) LOGGER.info ("MQ_CON_RECONSUME {} BROKER {} QUEUE {}", new Object [] {topic, broker, queueId}); span.failed (var18); span.addEvent ("MQConsumer", ConsumeConcurrentlyStatus.RECONSUME_LATER.name ()) If (CollectionUtils.isNotEmpty (msgs) & & ((MessageExt) msgs.get (0)). GetDelayTimeLevel () > = 2 + this.retryTimes) {context.setDelayLevelWhenNextConsume (- 1);}}

As can be seen from the code, if the consumption fails, we control the number of retransmissions by ourselves. The code is as follows:

If (CollectionUtils.isNotEmpty (msgs) & & ((MessageExt) msgs.get (0)). GetDelayTimeLevel () > = 2 + this.retryTimes) {context.setDelayLevelWhenNextConsume (- 1);}

When the retry meets the conditions, instead of retrying, put it in the dlq queue. If you don't control it, you will retry to the highest DelayLevel 18.

Why doesn't the default value of 2.DelayTimeLeve start at 0, but at 3?

We know that the default configuration of RocketMQ is messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h, which represents delay level1-level18, why not start with 1?

With doubt, we continued to dig into the source code, and we found a piece of code in the DefaultMQPullConsumerImpl class.

Public void sendMessageBack (MessageExt msg, int delayLevel, String brokerName, String consumerGroup) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {try {String brokerAddr = null! = brokerName? This.mQClientFactory.findBrokerAddressInPublish (brokerName): RemotingHelper.parseSocketAddressAddr (msg.getStoreHost ()); if (UtilAll.isBlank (consumerGroup)) {consumerGroup = this.defaultMQPullConsumer.getConsumerGroup ();} this.mQClientFactory.getMQClientAPIImpl (). ConsumerSendMessageBack (brokerAddr, msg, consumerGroup, delayLevel, 3000L, this.defaultMQPullConsumer.getMaxReconsumeTimes ()) } catch (Exception var8) {this.log.error ("sendMessageBack Exception," + this.defaultMQPullConsumer.getConsumerGroup (), var8); Message newMsg = new Message (MixAll.getRetryTopic (this.defaultMQPullConsumer.getConsumerGroup ()), msg.getBody ()); String originMsgId = MessageAccessor.getOriginMessageId (msg); MessageAccessor.setOriginMessageId (newMsg, UtilAll.isBlank (originMsgId)? Msg.getMsgId (): originMsgId); newMsg.setFlag (msg.getFlag ()); MessageAccessor.setProperties (newMsg, msg.getProperties ()); MessageAccessor.putProperty (newMsg, "RETRY_TOPIC", msg.getTopic ()); MessageAccessor.setReconsumeTime (newMsg, String.valueOf (msg.getReconsumeTimes () + 1); MessageAccessor.setMaxReconsumeTimes (newMsg, String.valueOf (this.defaultMQPullConsumer.getMaxReconsumeTimes () NewMsg.setDelayTimeLevel (3 + msg.getReconsumeTimes ()); this.mQClientFactory.getDefaultMQProducer () .send (newMsg);}}

You can see from the code that DelayTimeLevel = 3+reconsumeTime

NewMsg.setDelayTimeLevel (3 + msg.getReconsumeTimes ())

So when retrying by default, it actually starts at 3. From the point of view of time, it also verifies why it will be retried 4 times, and the interval between each time is 10s/30s/1m.

Is it helpful for you to read the above content? If you want to know more about the relevant knowledge or read more related articles, please follow the industry information channel, thank you for your support.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.