验证RocketMQ默认重试

编程入门行业动态更新时间:2024-10-09 21:29:01

验证RocketMQ默认<a href=https://www.elefans.com/category/jswz/34/1771334.html style= 重试"/>

验证RocketMQ默认重试

背景：线上有丢消息（无重试）的情况。经过排查，是因为并发时会有一个消息消费失败，但没有进行重试。检查配置无误后决定升级MQ版本。在升级版本前后经过下面的验证。发现新版MQ重试没问题，而老版MQ存在兼容问题而不能重试消息。

先说下结论，后边事对此结论的验证：
RocketMQ的默认重试次数为16次，且默认重试间隔时间为（1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h）

一、环境说明

老版MQ版本：V4_4_0
新版MQ版本：V4_9_3
java

二、java代码

1.生产者

@RestController
@Api(value = "测试服务", description = "测试服务")
@RequestMapping(value = "/test", method = RequestMethod.POST, produces = MediaType.APPLICATION_JSON_UTF8_VALUE)
public class TestController {private final static Logger logger = LoggerFactory.getLogger(TestController.class);@Resource(name = "testReconsumProducer")private MqNormalProducer testReconsumProducer;@RequestMapping(value = "/testReconsumProducer")public Result<String> testReconsumProducer() {Result<String> result = Result.instance();String msgId = null;// 消息体当前时间MqMessage mqMessage = new MqMessage("testMessage,time:" + TimeUtils.parseDate(TimeUtils.currentTime()));// 发消息RocketSendResult rocketResult = (RocketSendResult) testReconsumProducer.sendMsg(mqMessage);// 获取消息的msgidmsgId = rocketResult.getSendResult().getMsgId();// 生产者日志打印（因为测试的集群环境，为方便跟踪，使用slf4j记录）logger.error("{}生产者,测试MQ重试，rocketResult：{}", rocketResult.getSendResult().getMsgId(), JSON.toJSONString(rocketResult));result.setData(msgId);return result;}
}

2.消费者

@Service("testReconsumeEventHandler")
public class TestReconsumeEventHandler implements MqListener {private final static Logger logger = LoggerFactory.getLogger(TestReconsumeEventHandler.class);@Overridepublic MqAction consume(MqMessage mqMessage, MqConsumeContext mqConsumeContext) {// 获取消息（因为做了抽象，所以才需要强转）RocketMessage rocketMessage = (RocketMessage) mqMessage;// 消费者日志打印（第一次消费会打印一次，其余每重试一次会打印一次，预计1+16次）logger.error("{}消费者,测试MQ重试，重新消费:{}", rocketMessage.getOriginMessage().getMsgId(), JSON.toJSONString(rocketMessage));// 直接进行重发消息return MqAction.ReconsumeLater;}
}