当前位置：首页 > article >正文

笔记：记一次使用RabbitMq的x-delayed-message延迟消息插件，出现消息立即消费，延迟时间后再次消费，引发的重复消费问题

article 2025/3/15 8:05:58

笔记：记一次使用RabbitMq的x-delayed-message延迟插件，出现消息立即消费，延迟时间后再次消费，引发的重复消费问题

RabbitTemplate配置如下：

	@Bean
	public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {
		connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
		connectionFactory.setPublisherReturns(true);
		RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
		rabbitTemplate.setMandatory(true);
		rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
		rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
			if (ack) {
				log.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause);
			} else {
				log.info("消息发送失败:correlationData({}),ack({}),cause({})", correlationData, ack, cause);
			}
		});
		rabbitTemplate.setReturnsCallback(returnCallback -> {
			// 失败回调返回的消息
			log.info("返回消息：{},返回code：{}，回复文本：{}，交换机：{}，路由：{}", returnCallback.getMessage().toString(), returnCallback.getReplyCode(), returnCallback.getReplyText(), returnCallback.getExchange(), returnCallback.getRoutingKey());
			// 重新发送消息
			rabbitTemplate.convertAndSend(returnCallback.getExchange(), returnCallback.getRoutingKey(), returnCallback.getMessage());
		});

		return rabbitTemplate;
	}

这里可以看到同时调用了setConfirmCallback和setReturnsCallback两个方法，而调用延迟消息的时候可以看到控制台打印如下：

2025-03-14 09:34:44.384  INFO 314260 --- [nectionFactory2] o.s.m.r.config.RabbitMqConfiguration     : 消息发送成功:correlationData(null),ack(true),cause(null)
2025-03-14 09:34:44.430  INFO 314260 --- [nectionFactory1] o.s.m.r.config.RabbitMqConfiguration     : 返回消息：(Body:'[B@1fd8ee4(byte[804])' MessageProperties [headers={spring_listener_return_correlation=2fac62ae-4a07-4437-82b6-4db702c52426, __TypeId__=org.springblade.modules.rabbit.message.MessageStruct}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, receivedDelay=30000, deliveryTag=0]),返回code：312，回复文本：NO_ROUTE，交换机：writeOffDelay.exchange，路由：writeOffDelay.routingKey
2025-03-14 09:34:44.452  INFO 314260 --- [nectionFactory1] o.s.m.r.config.RabbitMqConfiguration     : 消息发送成功:correlationData(null),ack(true),cause(null)

我们看到了两个消息发送成功和一个返回消息，也就是说我们同时发送了两次消息，一次是延迟队列消息推送，一次是因为失败又进行了普通消息推送，这就出现了立即消费一次，然后在设置的延迟时间之后又消费了一次，因为本身就推送了两条消息。

为什么会出现这种情况呢，因为第一次消息推送，是由延迟插件进行处理的，此时消息需要等待延迟并未进入队列进行消费，所以消息返回code是312：NO_ROUTE，无法路由到队列，因为设置了mandatory为true，即监听消息无法抵达队列时，进入setReturnsCallback方法进行失败消息处理，在上面的配置中，进入setReturnsCallback会再次进行推送。

我们从RabbitMq Management控制台也能看出问题，如下图所示：
在这里插入图片描述
我们可以看到在9点41分10秒左右进来2条消息（黄线，因为设置的刷新时间是5秒，0.4/s也就是2条消息），同时消费了一条（蓝线），再之后9点41分40秒左右消费了一条（蓝线），说明我们先消费的一条普通非延迟的消息，而在30秒之后又消费了一条延迟消息。

解决方式也很简单：
1、把rabbitTemplate.setMandatory(true)改为false，因为x-delayed-message延迟消息插件不支持mandatory设置，但是如果出现消息无法到达队列的情况就无法作出监听和对应的策略。
2、把上面配置中的setReturnsCallback注释或者删除，但是问题同第一点，如果出现消息无法到达队列的情况就无法作出监听和对应的策略。
3、对setReturnsCallback返回的消息做逻辑处理，判断消息code为312的时候就不做重新发送处理，这样既能正常使用普通消息推送也能使用延迟消息推送，因为通常情况下并不会出现找不到路由和队列的情况，除了延迟消息这个插件。

目前采用的是第3种方案。

查看全文

http://www.kler.cn/a/585268.html