条记：记一次使用RabbitMq的x-delayed-message耽误消息插件，出现消息立即 ...

条记：记一次使用RabbitMq的x-delayed-message耽误插件，出现消息立即消耗，耽误时间后再次消耗，引发的重复消耗问题

RabbitTemplate配置如下：

1.         @Bean
2.         public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {
3.                 connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
4.                 connectionFactory.setPublisherReturns(true);
5.                 RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
6.                 rabbitTemplate.setMandatory(true);
7.                 rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
8.                 rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
9.                         if (ack) {
10.                                 log.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause);
11.                         } else {
12.                                 log.info("消息发送失败:correlationData({}),ack({}),cause({})", correlationData, ack, cause);
13.                         }
14.                 });
15.                 rabbitTemplate.setReturnsCallback(returnCallback -> {
16.                         // 失败回调返回的消息
17.                         log.info("返回消息：{},返回code：{}，回复文本：{}，交换机：{}，路由：{}", returnCallback.getMessage().toString(), returnCallback.getReplyCode(), returnCallback.getReplyText(), returnCallback.getExchange(), returnCallback.getRoutingKey());
18.                         // 重新发送消息
19.                         rabbitTemplate.convertAndSend(returnCallback.getExchange(), returnCallback.getRoutingKey(), returnCallback.getMessage());
20.                 });
22.                 return rabbitTemplate;
23.         }

复制代码

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

1. 2025-03-14 09:34:44.384  INFO 314260 --- [nectionFactory2] o.s.m.r.config.RabbitMqConfiguration     : 消息发送成功:correlationData(null),ack(true),cause(null)
2. 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
3. 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种方案。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。