RabbitMQ延时队列实现方法

在基于CentOS或Rocky Linux的Docker环境中部署RabbitMQ 3.8后，实现延时队列是一个常见的需求。延时队列的核心应用场景非常明确，比如处理订单的超时自动取消，或者轮询获取支付服务的状态。要实现它，绕不开一个核心概念：死信。

需要知道什么是死信

死信是什么

死信（Dead Letter Message），顾名思义，就是那些在RabbitMQ中“流离失所”的消息——它们因为某些原因无法被正常投递或消费。但RabbitMQ并不会简单地将它们丢弃，而是会贴上一个“死信”标签，然后自动将其重新发布到一个你预先配置好的死信交换机（Dead Letter Exchange, DLX）。最终，这些消息会被DLX路由到一个专门的死信队列（Dead Letter Queue, DLQ）中等待后续处理。

那么，消息在什么情况下会“死亡”呢？主要有以下四种触发条件：

死信交换机与RepulishMessageRecoverer区别

这里需要厘清一个容易混淆的点：RabbitMQ原生的死信交换机（DLX）机制，和Spring AMQP提供的RepublishMessageRecoverer工具，虽然目的都是处理“问题消息”，但运作层面和时机截然不同。

1.死信队列+TTL的实现

理解了死信，尤其是“消息TTL过期会成为死信”这一条，实现延时队列的思路就清晰了：让消息在一个队列中等待指定时间（TTL），过期后成为死信，再被自动路由到真正的目标队列，从而实现延迟消费。

实现结构：

具体实现上，通常会在消费者服务中，使用@RabbitListener注解来声明并绑定最终消费消息的死信队列和交换机：

@RabbitListener(bindings = @QueueBinding(
        value=@Queue(name="dlx.queue",durable =" true"),
        exchange = @Exchange(name="dlx.direct"),
        key = {"hi"}
))
public void listenDlxQueue(String message)throws Exception{
    log.info("消费者监听到 dlx.queue的消息，{}",message);
}

然后，在一个配置类（例如NormalConfiguration）中，通过@Bean声明一个普通的队列和交换机，并将这个普通队列绑定到死信交换机上：

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class NormalConfiguration {
    @Bean
    public DirectExchange normalExchange(){
        return  new DirectExchange("normal.direct");
    }
    @Bean
    public Queue normalQueue(){
        return  QueueBuilder
                .durable("normal.queue")
                .deadLetterExchange("dlx.direct") // 关键：指定死信交换机
                .build();
    }
    //推荐依赖注入的方式
    //需要注意代码规范，方法名应为小驼峰，反之，会找不到bean
    @Bean
    public Binding normalExchangeBinding(Queue normalQueue, DirectExchange normalExchange){
        return  BindingBuilder
                .bind(normalQueue)
                .to(normalExchange)
                .with("hi");
    }
}

发送延迟消息时，通过setExpiration方法为消息设置存活时间（毫秒）：

@Test
public void testSendDelayMessage() throws Exception {
    rabbitTemplate.convertAndSend("normal.direct", "hi", "hello everyone__", message -> {
        message.getMessageProperties().setExpiration("10000"); // 延迟10秒
        return message;
    });
}

2.延时消息插件

使用死信队列+TTL的方式虽然可行，但配置起来略显繁琐，尤其是需要多种不同延迟时长时，可能需要为每个时长创建单独的队列。为此，RabbitMQ官方提供了一个更优雅的解决方案：延时消息插件（rabbitmq_delayed_message_exchange）。

这个插件的原理是引入了一种特殊类型的交换机。消息发送到这种交换机后，会被插件暂存起来，直到预设的延迟时间到达，才会被投递到目标队列。这大大简化了延迟消息的处理流程。

官方文档： https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq

插件下载地址（GitHub）： rabbitmq/rabbitmq-delayed-message-exchange (版本需与RabbitMQ 3.8.17匹配)

安装插件

对于Docker安装的RabbitMQ，需要先找到插件数据卷的挂载点：

docker volume inspect mq-plugins

从返回结果中找到Mountpoint路径（例如/var/lib/docker/volumes/mq-plugins/_data），将下载的插件文件复制到该目录下。然后进入容器启用插件：

docker exec -it mq rabbitmq-plugins enable rabbitmq_delayed_message_exchange

基于注解方式

启用插件后，声明交换机时，通过delayed = "true"属性将其标记为延迟交换机即可：

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "delay.queue", durable = "true"),
        exchange = @Exchange(value = "delay.direct", delayed = "true"), // 声明为延迟交换机
        key = "hidelay"
))
public void listenDelayQueue(String msg) {
    log.info("delay.queue：" + msg);
}

基于@Bean方式

同样，在@Bean声明方式中，使用ExchangeBuilder的.delayed()方法：

@Configuration
public class DirectConfiguration {
    @Bean
    public DirectExchange delayExchange() {
        return ExchangeBuilder
                .directExchange("delay.direct")
                .delayed() // 关键：这里声明为延迟交换机
                .durable(true)
                .build();
    }
    @Bean
    public Queue delayedQueue() {
        return new Queue("delay.queue");
    }
    @Bean
    public Binding delayQueueBinding() {
        return BindingBuilder.bind(delayedQueue()).to(delayExchange()).with("delay");
    }
}

发送消息时，使用setDelay方法设置延迟时间（毫秒）：

@Test
public void testSendDelayMessageByplugin() {
    rabbitTemplate.convertAndSend("delay.direct", "hidelay", "hello", new MessagePostProcessor() {
        @Override
        public Message postProcessMessage(Message message) throws AmqpException {
            message.getMessageProperties().setDelay(10000); // 延迟10秒
            return message;
        }
    });
    log.info("消息发送成功");
}

总结与对比

两种方案各有优劣，选择时需要根据实际业务场景权衡：

简单来说，如果你的业务延迟时间是固定的、且并发量巨大，死信队列方案更稳定、兼容性更好。如果你的业务需要高度灵活、动态的延迟时间，并且对精度要求较高，那么官方插件是更简洁高效的选择，前提是你的运维环境允许安装插件。