RocketMQ源码：消费重试次数默认值与设置自定义值

先说结论，16次

版本

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.9.1</version>
</dependency>
1
2
3
4
5

源码

首先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()));
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

可以看到取的defaultMQPushConsumer.getMaxReconsumeTimes()值，为-1时取16

private int getMaxReconsumeTimes() {
    return this.defaultMQPushConsumer.getMaxReconsumeTimes() == -1 ? 16 : this.defaultMQPushConsumer.getMaxReconsumeTimes();
}
1
2
3

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;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

setMaxReconsumeTimes在客户端sdk并没地方使用，在RocketMQ完整源码里，是被io.openmessaging.rocketmq.consumer.PushConsumerImpl 使用过，那边也是设的默认16

private int rmqMaxRedeliveryTimes = 16;
1

设置重试次数

创建DefaultMQPushConsumer后，设置即可

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
consumer.setMaxReconsumeTimes(3);
1
2