1. 前言
Consumer启动后会立即触发一次「重平衡」操作，给自己分配MessageQueue，对于新分配的MessageQueue会提交拉取请求，开始拉取消息进行消费。应用在消费消息时，返回消费状态CONSUME_SUCCESS或RECONSUME_LATER，如果消息消费失败，消息并不会丢失，Broker会在稍后一段时间重新投递该消息，如果超过16次都消费失败，Broker会认为Consumer已经不具备消费这条消息的能力，会将消息扔到死信队列，这个时候就需要人工介入处理了。
以上是RocketMQ消费者消息重试的大致流程，具体的实现原理是怎样的呢？通过源码分析下吧。
2. 源码分析
PullMessageService是消息拉取服务，它是一个单独的线程池，Consumer重平衡时，对于新分配MessageQueue会提交消息拉取请求交给它处理。
接收到消息拉取请求后，线程会被唤醒，通过PullAPIWrapper类给Broker发送消息拉取请求。Broker给Consumer投递消息后，PullMessageService会提交消息消费请求给ConsumeMessageService处理。
// 构建消费请求，让消费者去消费消息
DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(pullResult.getMsgFoundList(),processQueue,pullRequest.getMessageQueue(),dispatchToConsume);
ConsumeMessageService是消息消费服务，它也是一个单独的线程池，拉取到消息后会调用它的submitConsumeRequest方法提交消费请求到线程池等待被调度执行。它有两个实现类，ConsumeMessageConcurrentlyService用来并发消费消息，ConsumeMessageOrderlyService用来消费顺序消息，RocketMQ可以严格保证消息有序，本文暂且只看前者。
// 创建消费请求
ConsumeRequest consumeRequest = new ConsumeRequest(msgs, processQueue, messageQueue);
try {// 提交到线程池执行this.consumeExecutor.submit(consumeRequest);
} catch (RejectedExecutionException e) {this.submitConsumeRequestLater(consumeRequest);
}
ConsumeRequest提交到线程池后开始异步执行，它的run方法会调用开发者注册的MessageListener对消息进行消费。只有返回CONSUME_SUCCESS才代表消息成功消费，返回NULL、消费超时、消费抛异常都算消费失败。
ConsumeConcurrentlyStatus status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);
随后，调用processConsumeResult方法处理消费结果。
- 相关数据的统计。
- 集群模式下，消费失败的消息回传给Broker。
- 回传失败，提交异步任务5秒后消费重试。
- 上报消费位点，但不包括回传失败的消息。
/*** 处理消费结果* 1.相关数据统计* 2.根据ackIndex将失败消息回传给Broker* 3.回传失败，提交异步任务，5s后重试消费* 4.上报消费位点*/
public void processConsumeResult(final ConsumeConcurrentlyStatus status,final ConsumeConcurrentlyContext context,final ConsumeRequest consumeRequest) {// ackIndex前的消息代表成功，之后的代表失败int ackIndex = context.getAckIndex();if (consumeRequest.getMsgs().isEmpty())return;// 统计数据switch (status) {case CONSUME_SUCCESS:if (ackIndex >= consumeRequest.getMsgs().size()) {ackIndex = consumeRequest.getMsgs().size() - 1;}int ok = ackIndex + 1;int failed = consumeRequest.getMsgs().size() - ok;this.getConsumerStatsManager().incConsumeOKTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), ok);this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), failed);break;case RECONSUME_LATER:// 消费失败，重置ackIndex，整批消息都需要重新处理ackIndex = -1;this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(),consumeRequest.getMsgs().size());break;default:break;}switch (this.defaultMQPushConsumer.getMessageModel()) {case BROADCASTING:// 广播模式下，直接丢掉for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {MessageExt msg = consumeRequest.getMsgs().get(i);log.warn("BROADCASTING, the message consume failed, drop it, {}", msg.toString());}break;case CLUSTERING:// 集群模式下，回传给Broker// 消息回传失败列表List<MessageExt> msgBackFailed = new ArrayList<MessageExt>(consumeRequest.getMsgs().size());// ackIndex之前的消息代表消费成功，之后的消息失败需要发回Brokerfor (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {MessageExt msg = consumeRequest.getMsgs().get(i);// 回传给Brokerboolean result = this.sendMessageBack(msg, context);if (!result) {// 回传失败，暂存到msgBackFailed，稍后重新消费msg.setReconsumeTimes(msg.getReconsumeTimes() + 1);msgBackFailed.add(msg);}}if (!msgBackFailed.isEmpty()) {// 回传失败的消息从msgs删除，不会上报失败消息的位点consumeRequest.getMsgs().removeAll(msgBackFailed);// 提交消费请求，5s后消费重试this.submitConsumeRequestLater(msgBackFailed, consumeRequest.getProcessQueue(), consumeRequest.getMessageQueue());}break;default:break;}long offset = consumeRequest.getProcessQueue().removeMessage(consumeRequest.getMsgs());if (offset >= 0 && !consumeRequest.getProcessQueue().isDropped()) {// 上报消费位点，不包含发回Broker失败的消息this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), offset, true);}
}
Consumer使用TreeMap存储拉取到的消息，Key是消息的Offset，Value是消息。红黑树结构，因此Key是有序的。上报的消费位点取的是最小的Key，即最小的Offset。
广播模式下，消费失败消息会丢弃。集群模式下，消费失败的消息需要回传给Broker，对应的方法是sendMessageBack。由于Consumer恢复需要一定的时间，如果Broker立马重新投递该消息，大概率还是会消费失败，所以Broker会将消息发送到延时队列中，隔一段时间再重新投递。延时多久，通过delayLevelWhenNextConsume属性来设置：-1代表不重试，直接扔到死信队列；0代表由Broker控制重试频率；>0代表由客户端控制重试频率。
public boolean sendMessageBack(final MessageExt msg, final ConsumeConcurrentlyContext context) {// 消息重试的延时级别int delayLevel = context.getDelayLevelWhenNextConsume();msg.setTopic(this.defaultMQPushConsumer.withNamespace(msg.getTopic()));try {this.defaultMQPushConsumerImpl.sendMessageBack(msg, delayLevel, context.getMessageQueue().getBrokerName());return true;} catch (Exception e) {log.error("sendMessageBack exception, group: " + this.consumerGroup + " msg: " + msg.toString(), e);}return false;
}
消息如果回传失败，会再重试一次。第一次是正常的消息回传，发送CONSUMER_SEND_MSG_BACK命令，Broker根据消息Offset从CommitLog中读取消息，再重设Topic为重试队列，延时级别为3+重试次数，将消息重新写入CommitLog。
第一次回传如果失败，RocketMQ会进行一次重试，第二次会构建一个新的Message对象，拷贝原Message的内容和属性，重设Topic为重试队列，重设延时级别，发送一个新的消息给Broker。两者的逻辑是类似的。
public void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName)throws RemotingException, MQBrokerException, InterruptedException, MQClientException {try {String brokerAddr = (null != brokerName) ? this.mQClientFactory.findBrokerAddressInPublish(brokerName): RemotingHelper.parseSocketAddressAddr(msg.getStoreHost());// 第一次 消息回传this.mQClientFactory.getMQClientAPIImpl().consumerSendMessageBack(brokerAddr, msg,this.defaultMQPushConsumer.getConsumerGroup(), delayLevel, 5000, getMaxReconsumeTimes());} catch (Exception e) {log.error("sendMessageBack Exception, " + this.defaultMQPushConsumer.getConsumerGroup(), e);// 回传失败，再试一次，发送延时消息Message newMsg = new Message(MixAll.getRetryTopic(this.defaultMQPushConsumer.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, MessageConst.PROPERTY_RETRY_TOPIC, msg.getTopic());MessageAccessor.setReconsumeTime(newMsg, String.valueOf(msg.getReconsumeTimes() + 1));MessageAccessor.setMaxReconsumeTimes(newMsg, String.valueOf(getMaxReconsumeTimes()));MessageAccessor.clearProperty(newMsg, MessageConst.PROPERTY_TRANSACTION_PREPARED);// 延时级别newMsg.setDelayTimeLevel(3 + msg.getReconsumeTimes());this.mQClientFactory.getDefaultMQProducer().send(newMsg);} finally {msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(), this.defaultMQPushConsumer.getNamespace()));}
}
虽然Topic设置的是重试队列，但由于是延时消息，Broker存储消息时会改写Topic，放到延时队列中，然后由异步线程定时扫描这些消息。消息的延时时间一到，ScheduleMessageService会将需要交付的消息重新写回CommitLog，因为Consumer启动时会自动订阅重试Topic，所以Broker会将这些消息重新投递给Consumer消费，然后重复上面的流程。
消息的重试次数会记录在消息的Properties中，消息重试的延时级别会随着重试次数的增多而变大，消息重试次数越多，重试的间隔时间会越长，默认的重试间隔策略为3+reconsumeTimes，即10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h...，最大重试次数为16，超过最大重试次数，Broker会认为Consumer已经不具备消费这条消息的能力了，会将消息投递到死信队列。
3. 总结
ConsumeMessageService在消费消息时，会根据消费状态去处理消费结果。消费成功的处理很简单，从本地缓存中删除消息，然后上报消费位点即可。
对于消费失败的消息，Consumer会将其重新回传给Broker，然后将这一批消息视为消费成功，直接上报消费位点即可。但是消息回传给Broker的过程可能会失败，一旦失败就不能上报这些消息的消费位点了，否则消息就丢失了，Consumer会先提交一个异步任务，5秒后重新消费这批回传失败的消息，然后上报位点时忽略这些回传失败的消息Offset。
Broker在接收到Consumer回传的消息后，会根据重试次数设置延时级别，将消息暂时投递到延时队列中，然后异步线程定时扫描延时消息，消息到期后再将其重新投递到重试队列中。由于Consumer启动时会自动订阅自身所在Group的重试Topic的，自然也就能正常消费重试消息了。