先说结论，16次
版本
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.9.1</version>
</dependency>
源码
首先RocketMq默认模式是push，不是pull
org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl
执行sendMessageBack方法时通过getMaxReconsumeTimes()获取了最大重试次数
public void sendMessageBack(MessageExt msg, int delayLevel, 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, 5000L, this.getMaxReconsumeTimes());
} catch (Exception var10) {
this.log.error("sendMessageBack Exception, " + this.defaultMQPushConsumer.getConsumerGroup(), var10);
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, "RETRY_TOPIC", msg.getTopic());
MessageAccessor.setReconsumeTime(newMsg, String.valueOf(msg.getReconsumeTimes() + 1));
MessageAccessor.setMaxReconsumeTimes(newMsg, String.valueOf(this.getMaxReconsumeTimes()));
MessageAccessor.clearProperty(newMsg, "TRAN_MSG");
newMsg.setDelayTimeLevel(3 + msg.getReconsumeTimes());
this.mQClientFactory.getDefaultMQProducer().send(newMsg);
} finally {
msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(), this.defaultMQPushConsumer.getNamespace()));
}
可以看到取的defaultMQPushConsumer.getMaxReconsumeTimes()值，为-1时取16
private int getMaxReconsumeTimes() {
return this.defaultMQPushConsumer.getMaxReconsumeTimes() == -1 ? 16 : this.defaultMQPushConsumer.getMaxReconsumeTimes();
}
org.apache.rocketmq.client.consumer.DefaultMQPushConsumer
确认设置maxReconsumeTimes为-1，In concurrently mode, -1 means 16，In orderly mode, -1 means Integer.MAX_VALUE
/**
* Max re-consume times.
* In concurrently mode, -1 means 16;
* In orderly mode, -1 means Integer.MAX_VALUE.
*
* If messages are re-consumed more than {@link #maxReconsumeTimes} before success.
*/
private int maxReconsumeTimes = -1;
public void setMaxReconsumeTimes(final int maxReconsumeTimes) {
this.maxReconsumeTimes = maxReconsumeTimes;
}
public int getMaxReconsumeTimes() {
return this.maxReconsumeTimes;
}
setMaxReconsumeTimes在客户端sdk并没地方使用，在RocketMQ完整源码里，是被io.openmessaging.rocketmq.consumer.PushConsumerImpl 使用过，那边也是设的默认16
private int rmqMaxRedeliveryTimes = 16;
设置重试次数
创建DefaultMQPushConsumer后，设置即可
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
consumer.setMaxReconsumeTimes(3);
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/93676.html
