最近在研究RabbitMQ如何实现延时队列时发现消息进入死信队列的情况之一就是当消息数量超过队列设置的最大长度时会被丢入死信队列,看到这时我就产生了一个疑问,到底是最后插入的消息还是最早插入的消息会被丢入死信队列呢?遗憾的是看了几篇博客都是模棱两可的答案,还有的说超过数量后该消息会被放入死信队列,看完之后还是对这个问题将信将疑。所以我决定去探究一下正确答案
RabbitMQ官方文档
遇事不决肯定是先看官方文档最靠谱啦,在官网中扒拉了半天终于找到说明这个问题的页面了,就是上面引用的链接,重点如下:
翻译过来就是:在RabbitMQ中,当消息的数量或大小达到限制后,默认的操作是删除最旧的消息或将最旧的消息放入死信队列,这取决于该队列是否配置了死信队列。 我们可以通过使用overflow
配置指定的处理策略,如果overflow
被设置为reject-publish
或reject-publish-dlx
,那么会将最新插入的消息丢弃。如果该队列开启了confirm机制,发布者会收到nack的信息,如果一个消息被路由到多个队列,只要其中一个队列拒绝发布者就会收到nack消息,但是没被拒绝的队列可以正确接收到消息。reject-publish
和reject-publish-dlx
的区别是后者还会将拒绝的消息放入死信队列。
下面我们使用demo来验证一下各个策略的现象
rabbitmq: host: 127.0.0.1 port: 5672 username: username password: password virtual-host: /test publisher-confirm-type: correlated # 配置启用confirm机制
import org.springframework.amqp.core.*; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import java.util.HashMap; @Configuration public class RabbitMQConfig { public static final String DELAY_EXCHANGE_NAME = "delay.business.exchange"; public static final String DELAY_QUEUE_NAME = "delay.business.queue"; public static final String DELAY_QUEUE_ROUTING_KEY = "delay.business.queue.routingKey"; public static final String DEAD_LETTER_EXCHANGE_NAME = "dead.letter.exchange"; public static final String DEAD_LETTER_QUEUE_NAME = "dead.letter.queue"; public static final String DEAD_LETTER_QUEUE_ROUTING_KEY = "dead.letter.queue.routingKey"; // 声明延迟队列交换机 @Bean("delayExchange") public DirectExchange delayExchange(){ return new DirectExchange(DELAY_EXCHANGE_NAME); } // 声明死信队列交换机 @Bean("deadLetterExchange") public DirectExchange deadLetterExchange(){ return new DirectExchange(DEAD_LETTER_EXCHANGE_NAME); } // 声明延时队列 @Bean("delayQueue") public Queue delayQueue(){ HashMap<String, Object> map = new HashMap<>(); // x-dead-letter-exchange 这里声明当前队列绑定的死信交换机 map.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE_NAME); // x-dead-letter-routing-key 这里声明当前队列的死信路由key map.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUE_ROUTING_KEY); // 设置该队列最大消息数 map.put("x-max-length", 10); map.put("x-overflow", "reject-publish"); return QueueBuilder.durable(DELAY_QUEUE_NAME).withArguments(map).build(); } // 声明死信队列 @Bean("deadLetterQueue") public Queue deadLetterQueue(){ return new Queue(DEAD_LETTER_QUEUE_NAME); } // 声明延时队列的绑定关系 @Bean public Binding delayBinding(@Qualifier("delayExchange") DirectExchange directExchange, @Qualifier("delayQueue") Queue queue){ return BindingBuilder.bind(queue).to(directExchange).with(DELAY_QUEUE_ROUTING_KEY); } // 声明死信队列的绑定关系 @Bean public Binding deadLetterBinding(@Qualifier("deadLetterExchange") DirectExchange directExchange, @Qualifier("deadLetterQueue") Queue queue){ return BindingBuilder.bind(queue).to(directExchange).with(DEAD_LETTER_QUEUE_ROUTING_KEY); } }
注意,这里我们并没有设置overflow
参数,所以采用的是默认配置
3. 创建消费者监听死信队列
import com.rabbitmq.client.Channel; import com.whs.edws.config.RabbitMQConfig; import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; import java.io.IOException; @Slf4j @Component public class MaxLengthConsumer { @RabbitListener(queues = RabbitMQConfig.DEAD_LETTER_QUEUE_NAME) public void receive(Message message, Channel channel) throws IOException { String s = new String(message.getBody()); log.info("死信队列消费者接收到消息:" + s); channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); } }
@Test void maxLengthTestPublish(){ rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { /** * @param correlationData 相关配置信息 * @param ack 消息队列是否成功收到消息 * @param cause 错误原因 */ @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { if (ack) { logger.info("消息发送成功:" + correlationData.getId()); } else { logger.error("消息发送失败:" + correlationData.getId()); logger.error("错误原因:" + cause); } } }); for (int i = 0; i < 11; i++) { CorrelationData correlationData = new CorrelationData(); correlationData.setId(String.valueOf(i)); rabbitTemplate.convertAndSend(RabbitMQConfig.DELAY_EXCHANGE_NAME, RabbitMQConfig.DELAY_QUEUE_ROUTING_KEY, String.valueOf(i), correlationData); } }
2023-07-18 02:37:52.941 INFO 24308 --- [ntContainer#1-1] com.edws.rabbitmq.MaxLengthConsumer : 死信队列消费者接收到消息:0
reject-publish策略的验证代码只需在默认策略的基础上加上配置即可,我们在定义队列的时候加上配置
// 指定超过队列长度后的策略 map.put("x-overflow", "reject-publish");
执行方法,打印的结果为:
2023-07-18 02:45:07.242 INFO 22328 --- [nectionFactory2] o.s.amqp.rabbit.core.RabbitTemplate : 消息发送失败:10 2023-07-18 02:45:07.242 INFO 22328 --- [nectionFactory2] o.s.amqp.rabbit.core.RabbitTemplate : 错误原因:null
通过日志可以看到,最新插入的消息被丢弃了。至于cause为什么是null,我没找到原因,如果了解的朋友可以在评论里讨论一下
reject-publish-dlx
策略的代码也是只需要在默认代码中加一行配置即可
// 指定超过队列长度后的策略 map.put("x-overflow", "reject-publish-dlx");
打印结果如下:
2023-07-18 02:49:13.246 INFO 10488 --- [nectionFactory2] o.s.amqp.rabbit.core.RabbitTemplate : 消息发送失败:10 2023-07-18 02:49:13.246 INFO 10488 --- [nectionFactory2] o.s.amqp.rabbit.core.RabbitTemplate : 错误原因:null 2023-07-18 02:49:13.252 INFO 10488 --- [ntContainer#1-1] com.whs.edws.rabbitmq.MaxLengthConsumer : 死信队列消费者接收到消息:10
通过日志可以看出,最新的一条消息被拒绝且被放入死信队列