【RabbitMQ】当队列中消息数量超过最大长度的淘汰策略
说明
最近在研究RabbitMQ如何实现延时队列时发现消息进入死信队列的情况之一就是当消息数量超过队列设置的最大长度时会被丢入死信队列，看到这时我就产生了一个疑问，到底是最后插入的消息还是最早插入的消息会被丢入死信队列呢？遗憾的是看了几篇博客都是模棱两可的答案，还有的说超过数量后该消息会被放入死信队列，看完之后还是对这个问题将信将疑。所以我决定去探究一下正确答案
答案
遇事不决肯定是先看官方文档最靠谱啦，在官网中扒拉了半天终于找到说明这个问题的页面了，就是上面引用的链接，重点如下：
翻译过来就是：在RabbitMQ中，当消息的数量或大小达到限制后，默认的操作是删除最旧的消息或将最旧的消息放入死信队列，这取决于该队列是否配置了死信队列。 我们可以通过使用overflow配置指定的处理策略，如果overflow被设置为reject-publish或reject-publish-dlx，那么会将最新插入的消息丢弃。如果该队列开启了confirm机制，发布者会收到nack的信息，如果一个消息被路由到多个队列，只要其中一个队列拒绝发布者就会收到nack消息，但是没被拒绝的队列可以正确接收到消息。reject-publish和reject-publish-dlx的区别是后者还会将拒绝的消息放入死信队列。
验证
下面我们使用demo来验证一下各个策略的现象
默认策略（drop-head）
- application.yml配置如下：
rabbitmq:
host: 127.0.0.1
port: 5672
username: username
password: password
virtual-host: /test
publisher-confirm-type: correlated # 配置启用confirm机制
- 使用RabbitMQConfig创建业务队列和对应死信队列
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);
}
}
- 运行结果
看上面的代码可知，我们设置了队列大小为10，但是我们向队列发送了11条消息，最后日志打印如下：2023-07-18 02:37:52.941 INFO 24308 --- [ntContainer#1-1] com.edws.rabbitmq.MaxLengthConsumer : 死信队列消费者接收到消息：0
和官方文档说的一样，默认最旧的一条消息被放入死信队列
reject-publish策略
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策略
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
通过日志可以看出，最新的一条消息被拒绝且被放入死信队列