Springcloud学习笔记63---RocketMq超时重试，导致重复消费的问题，解决方案

1. 重复消费的背景

当Consumer处理时间过长，在超时时间内没有返回给Broker消费状态，那么Broker也会自动重试。

设定一个超时时间，达到超时时间的那个消费当作消费失败处理。

Java客户端中的DefaultPushConsumer中的构造方法中的consumeTimeout字段（默认15分钟）。

package william.rmq.consumer.quickstart;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import william.rmq.common.constant.RocketMQConstant;

import javax.annotation.PostConstruct;
import java.util.List;

/**
* @Auther: ZhangShenao
* @Date: 2018/9/7 11:06
* @Description:RocketMQ消息消费者
*/
@Slf4j
@Service
public class MessageConsumer implements MessageListenerConcurrently {
   @Value("${spring.rocketmq.namesrvAddr}")
   private String namesrvAddr;

   private final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("DefaultConsumer");


   @PostConstruct
   public void start() {
       try {
           consumer.setNamesrvAddr(namesrvAddr);

           //从消息队列头部开始消费
           consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

           //设置集群消费模式
           consumer.setMessageModel(MessageModel.CLUSTERING);

           //设置消费超时时间(分钟)
           consumer.setConsumeTimeout(RocketMQConstant.CONSUMER_TIMEOUT_MINUTES);

           //订阅主题
           consumer.subscribe("DefaultCluster", "*");

           //注册消息监听器
           consumer.registerMessageListener(this);

           //启动消费端
           consumer.start();

           log.info("Message Consumer Start...");
           System.err.println("Message Consumer Start...");
       } catch (MQClientException e) {
           log.error("Message Consumer Start Error!!",e);
       }

   }

   @Override
   public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
       if (CollectionUtils.isEmpty(msgs)) {
           return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
       }

       MessageExt message = msgs.get(0);
       try {
           //逐条消费
           String messageBody = new String(message.getBody(), RemotingHelper.DEFAULT_CHARSET);
           System.err.println("Message Consumer: Handle New Message: messageId: " + message.getMsgId() + ",topic: " +
                   message.getTopic() + ",tags: " + message.getTags() + ",messageBody: " + messageBody);

           //模拟耗时操作2分钟，大于设置的消费超时时间
           Thread.sleep(1000L * 60 * 2);
           return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
       } catch (Exception e) {
           log.error("Consume Message Error!!", e);
           return ConsumeConcurrentlyStatus.RECONSUME_LATER;
       }
   }

}

解决方案：

(1) 调大consumeTimeout字段的值；

(2) 消费回调的方法进行幂等性校验，进行逻辑校验；

参考文献：

https://blog.csdn.net/m0_45406092/article/details/120531495  (【RocketMQ】消息重试、重试次数设置、死信队列)

https://zhuanlan.zhihu.com/p/25265380