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

RabbitMQ 高级特性之延迟队列

news 2025/7/12 4:31:56

1. 简介

在某些场景下，当生产者发送消息后，可能不需要让消费者立即接收到，而是让消息延迟一段时间后再发送给消费者。

2. 实现方式

2.1 TTL + 死信队列

给消息设置过期时间后，若消息在这段时间内没有被消费，就会将消息发送到死信队列中，我们可以利用这一特性，将需要延迟发送的消息设置过期时间，然后再让消费者从死信队列中获取消息，这样就实现了消息的延迟发送。

队列与交换机配置如下：

@Configuration
public class DLConfig {/*** 正常队列、交换机* @return*/@Bean("norQueue")public Queue norQueue() {return QueueBuilder.durable(Constants.NOR_QUEUE).deadLetterExchange(Constants.DL_EXCHANGE) //绑定死信交换机.deadLetterRoutingKey(Constants.DL_ROUTINGKEY).build();}@Bean("norExchange")public DirectExchange norExchange() {return ExchangeBuilder.directExchange(Constants.NOR_EXCHANGE).build();}@Bean("norBind")public Binding norBind(@Qualifier("norExchange") DirectExchange directExchange,@Qualifier("norQueue") Queue queue) {return BindingBuilder.bind(queue).to(directExchange).with(Constants.NOR_ROUTINGKEY);}/*** 死信队列、交换机*/@Bean("dlQueue")public Queue dlQueue() {return QueueBuilder.durable(Constants.DL_QUEUE).build();}@Bean("dlExchange")public DirectExchange dlExchange() {return ExchangeBuilder.directExchange(Constants.DL_EXCHANGE).build();}@Bean("dlBind")public Binding dlBind(@Qualifier("dlExchange") DirectExchange directExchange,@Qualifier("dlQueue") Queue queue) {return BindingBuilder.bind(queue).to(directExchange).with(Constants.DL_ROUTINGKEY);}
}

生产者代码如下：

    @RequestMapping("/dl1")public String dl1() {String messageInfo = "dl... " + new Date();MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setExpiration("10000"); //10s 后过期return message;}};rabbitTemplate.convertAndSend(Constants.NOR_EXCHANGE, Constants.NOR_ROUTINGKEY, messageInfo, messagePostProcessor);return "消息发送成功";}

消费者代码如下：

@Component
@Slf4j
public class DLListener {/*** ttl + 死信队列 -> 延时队列* @param message*/@RabbitListener(queues = Constants.DL_QUEUE)public void listener(Message message) {String messageInfo = new String(message.getBody());log.info("接收到消息: {}, time: {}", messageInfo, new Date());}
}

由于消息发送到了死信队列，于是我们只需要从死信队列中获取消息即可。

代码运行结果如下：

从运行结果中可以看出，消息延迟了 10s 才被消费。

这种实现方式的问题：

但是，当我们连续发送两条消息，第一条消息的过期时间为 15s，第二条消息的过期时间为 10s，代码运行结果如下：

这里我们看到，虽然第二条消息先过期，但却和第一条消息一起被消费，按照正常情况下第二条消息应该率先被消费，于是这种实现方式存在一定的问题。

2.2 使用插件

2.2.1 安装插件

虽然 RabbitMQ 没有提供延迟队列的使用方式，但是提供了延迟队列的插件，我们可以安装插件并使用。

插件安装地址：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

需要下载 .ez 的插件。

需要根据本机的 RabbitMQ 版本选择匹配的插件版本，不然无法使用。

插件下载完成后，需要将插件放到 /usr/lib/rabbitmq/plugins 目录下，若没有需要进行创建。

安装完成后，使用下面这行命令启动插件：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

启动完成后，需要重启 RabbitMQ 服务，这样插件就能正常运行。

2.2.2 使用插件

插件安装完成后，交换机的类型就会多出下面一种：

即延迟队列，于是我们在声明交换机是，就能够声明这个类型的交换机。

队列与交换机配置如下：

@Configuration
public class DelayConfig {@Bean("delayQueue")public Queue delayQueue() {return QueueBuilder.durable(Constants.DELAY_QUEUE).build();}/*** 延迟交换机* @return*/@Bean("delayExchange")public DirectExchange delayExchange() {return ExchangeBuilder.directExchange(Constants.DELAY_EXCHANGE).delayed().build();}@Bean("delayBind")public Binding delayBind(@Qualifier("delayExchange") DirectExchange directExchange,@Qualifier("delayQueue") Queue queue) {return BindingBuilder.bind(queue).to(directExchange).with(Constants.DELAY_ROUTINGKEY);}
}

在声明交换机时，使用了 delayed 来声明该队列是延迟队列。

生产者代码如下：

    @RequestMapping("/delay")public String delay() {String messageInfo = "delay ...";rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE, Constants.DELAY_ROUTINGKEY, messageInfo + "25000ms", message -> {message.getMessageProperties().setDelayLong(20000L); //过期时间，单位为 msreturn message;});rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE, Constants.DELAY_ROUTINGKEY, messageInfo + "10000ms", message -> {message.getMessageProperties().setDelayLong(10000L); //过期时间，单位为 msreturn message;});return "消息发送成功";}

消费者代码如下：

@Component
@Slf4j
public class DelayListener {@RabbitListener(queues = Constants.DELAY_QUEUE)public void listener(Message message) {log.info("接收到消息: {}, time: {}", new String(message.getBody()), new Date());}
}

运行结果如下：