Java 中间件：RabbitMQ 延迟队列（死信交换机实现）

我们可以在声明队列时，通过参数指定 DLX 和 DLK：

Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "dlx.exchange");
args.put("x-dead-letter-routing-key", "dlx.routing.key");

4. 延迟队列的构建逻辑

结合以上概念，延迟队列的实现流程如下：

1. 创建一个 延迟队列（Delay Queue），设置 TTL，并绑定到一个 普通交换机（Delay Exchange）。
2. 为该延迟队列配置 DLX 和 DLK，指向 死信交换机（DLX Exchange）。
3. 创建一个 业务队列（Business Queue），绑定到死信交换机（使用 DLK 作为 routing key）。
4. 生产者将消息发送到 延迟交换机，消息进入延迟队列。
5. 消息在延迟队列中等待 TTL 时间。
6. TTL 到期后，消息变为死信，被自动路由到 死信交换机，进而进入 业务队列。
7. 消费者监听 业务队列，处理延迟后的消息。

环境准备

在开始编码前，请确保你已准备好以下环境：

• RabbitMQ 服务器：版本建议 3.8+（可通过 Docker 快速启动）
• Java 开发环境：JDK 11+
• 构建工具：Maven 或 Gradle
• Spring Boot：2.7.x 或 3.x（本文以 3.x 为例）

启动 RabbitMQ（Docker 方式）

docker run -d --hostname my-rabbit --name rabbitmq \
  -p 5672:5672 -p 15672:15672 \
  rabbitmq:3-management

访问 http://localhost:15672，使用默认账号 guest/guest 登录管理界面。

Spring Boot 项目搭建

创建一个新的 Spring Boot 项目，添加以下依赖（Maven）：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

配置 application.yml：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /

定义交换机、队列与绑定关系

我们将定义三组核心组件：

1. 延迟交换机（delay.exchange）：类型为 direct
2. 延迟队列（delay.queue）：带 TTL 和 DLX 配置
3. 死信交换机（dlx.exchange）：类型为 direct
4. 业务队列（business.queue）：绑定到死信交换机

使用 Java Config 方式声明

@Configuration
public class RabbitMQConfig {
    // ====== 延迟相关配置 ======
    public static final String DELAY_EXCHANGE_NAME = "delay.exchange";
    public static final String DELAY_QUEUE_NAME = "delay.queue";
    public static final String DELAY_ROUTING_KEY = "delay.key";

    // ====== 死信/业务相关配置 ======
    public static final String DLX_EXCHANGE_NAME = "dlx.exchange";
    public static final String BUSINESS_QUEUE_NAME = "business.queue";
    public static final String DLX_ROUTING_KEY = "dlx.key";

    // 1. 声明延迟交换机（Direct 类型）
    @Bean
    public DirectExchange delayExchange() {
        return new DirectExchange(DELAY_EXCHANGE_NAME);
    }

    // 2. 声明死信交换机（Direct 类型）
    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange(DLX_EXCHANGE_NAME);
    }

    // 3. 声明延迟队列，并配置 TTL 和 DLX
    @Bean
    public Queue delayQueue() {
        Map<String, Object> args = new HashMap<>();
        // 设置消息 TTL 为 5000 毫秒（可根据实际需求动态设置）
        args.put("x-message-ttl", 5000);
        // 指定死信交换机
        args.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME);
        // 指定死信路由键
        args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);
        // 可选：设置队列为持久化
        return QueueBuilder.durable(DELAY_QUEUE_NAME).withArguments(args).build();
    }

    // 4. 声明业务队列（死信最终到达的地方）
    @Bean
    public Queue businessQueue() {
        return QueueBuilder.durable(BUSINESS_QUEUE_NAME).build();
    }

    // 5. 绑定延迟队列到延迟交换机
    @Bean
    public Binding delayBinding() {
        return BindingBuilder.bind(delayQueue()).to(delayExchange()).with(DELAY_ROUTING_KEY);
    }

    // 6. 绑定业务队列到死信交换机
    @Bean
    public Binding businessBinding() {
        return BindingBuilder.bind(businessQueue()).to(dlxExchange()).with(DLX_ROUTING_KEY);
    }
}

上述配置会在 RabbitMQ 启动时自动创建所需的交换机、队列和绑定关系。

消息生产者实现

生产者负责将消息发送到 延迟交换机，并指定 routing key。

@Service
public class DelayMessageProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendDelayMessage(String message, long delayMillis) {
        // 创建消息属性
        MessageProperties properties = new MessageProperties();
        // 动态设置消息级别的 TTL（覆盖队列 TTL）
        properties.setExpiration(String.valueOf(delayMillis));
        properties.setContentType("text/plain");
        Message msg = new Message(message.getBytes(StandardCharsets.UTF_8), properties);

        // 发送到延迟交换机
        rabbitTemplate.send(RabbitMQConfig.DELAY_EXCHANGE_NAME,
                RabbitMQConfig.DELAY_ROUTING_KEY, msg);
        System.out.println("【生产者】发送延迟消息：'" + message + "'，延迟时间：" + delayMillis + "ms");
    }
}

💡 关键点：通过 MessageProperties.setExpiration() 可以动态设置每条消息的延迟时间，这比固定队列 TTL 更灵活。但要注意：消息 TTL 必须小于等于队列 TTL，否则队列 TTL 会生效。因此，通常我们会将队列 TTL 设置为一个较大的值（如 24 小时），而具体延迟时间由消息自身控制。

消息消费者实现

消费者监听的是 业务队列（business.queue），因为只有当延迟结束后，消息才会到达这里。

@Component
public class BusinessMessageConsumer {
    private static final Logger log = LoggerFactory.getLogger(BusinessMessageConsumer.class);

    @RabbitListener(queues = RabbitMQConfig.BUSINESS_QUEUE_NAME)
    public void consumeBusinessMessage(String message) {
        log.info("【消费者】收到延迟后的消息：{}", message);
        // 在此处执行你的业务逻辑，例如：取消订单、发送通知等
        processBusinessLogic(message);
    }

    private void processBusinessLogic(String message) {
        // 模拟业务处理
        System.out.println(">>> 正在处理业务：" + message);
    }
}

使用 @RabbitListener 注解即可轻松监听队列。

控制器测试接口

为了方便测试，我们提供一个 REST 接口：

@RestController
@RequestMapping("/delay")
public class DelayController {
    @Autowired
    private DelayMessageProducer producer;

    @PostMapping("/send")
    public ResponseEntity<String> sendDelayMessage(
            @RequestParam String content,
            @RequestParam(defaultValue = "5000") Long delayMillis) {
        producer.sendDelayMessage(content, delayMillis);
        return ResponseEntity.ok("消息已发送，将在 " + delayMillis + "ms 后处理");
    }
}

测试步骤

1. 启动 Spring Boot 应用。
2. 观察控制台输出：

   【生产者】发送延迟消息：'Hello Delayed World'，延迟时间：10000ms （等待 10 秒后...）
   【消费者】收到延迟后的消息：Hello Delayed World
   >>> 正在处理业务：Hello Delayed World
   

3. 调用接口：

   curl -X POST "http://localhost:8080/delay/send?content=Hello%20Delayed%20World&delayMillis=10000"
   

动态延迟时间的实现细节

前面提到，可以通过消息级别的 expiration 实现动态延迟。但这有一个重要前提：延迟队列的 TTL 必须足够大，以容纳所有可能的延迟时间。

最佳实践：设置队列 TTL 为最大值

// 在 delayQueue() 方法中
args.put("x-message-ttl", 86400000); // 24 小时（毫秒）

这样，任何小于 24 小时的 expiration 都能生效。

注意事项

• RabbitMQ 的 TTL 精度不是毫秒级的，实际延迟可能会有几十毫秒的误差。
• 如果同时设置了队列 TTL 和消息 TTL，取较小值。
• 消息 TTL 必须是字符串形式的正整数（如 "5000"），不能是 5000L。

多级延迟队列（应对不同延迟时间）

如果系统中有多种延迟需求（如 1 分钟、5 分钟、30 分钟、1 小时），是否需要为每种延迟创建一个队列？

答案是：不一定，但推荐这样做以提高效率。

方案一：单队列 + 动态 TTL（简单但低效）

• 优点：配置简单，只需一个延迟队列。
• 缺点：RabbitMQ 内部使用 优先级队列 存储带 TTL 的消息，但 只有队首消息的 TTL 才会被检查。这意味着：
  • 如果队列中有大量长延迟消息，短延迟消息排在后面，会导致 实际延迟时间远大于预期。
  • 例如：先发一个 1 小时的消息，再发一个 1 秒的消息，后者必须等前者过期后才能被处理！

📌 这是 RabbitMQ TTL 机制的一个重要限制！详情可参考官方文档：RabbitMQ Time-To-Live

方案二：多队列（推荐）

为常用的延迟时间创建多个延迟队列：

生产者根据所需延迟时间选择对应的 routing key。

示例代码（多队列配置）

public class MultiDelayConfig {
    public static final String DELAY_EXCHANGE = "multi.delay.exchange";
    public static final String DLX_EXCHANGE = "dlx.exchange";
    public static final String BUSINESS_QUEUE = "business.queue";
    public static final String DLX_ROUTING_KEY = "dlx.key";

    // 定义延迟时间常量
    public static final long DELAY_1_MIN = 60000;
    public static final long DELAY_5_MIN = 300000;
    public static final long DELAY_30_MIN = 1800000;

    @Bean
    public DirectExchange multiDelayExchange() {
        return new DirectExchange(DELAY_EXCHANGE);
    }

    @Bean
    public Queue delayQueue1m() {
        return createDelayQueue("delay.queue.1m", DELAY_1_MIN);
    }

    @Bean
    public Queue delayQueue5m() {
        return createDelayQueue("delay.queue.5m", DELAY_5_MIN);
    }

    @Bean
    public Queue delayQueue30m() {
        return createDelayQueue("delay.queue.30m", DELAY_30_MIN);
    }

    private Queue createDelayQueue(String queueName, long ttl) {
        Map<String, Object> args = new HashMap<>();
        args.put("x-message-ttl", ttl);
        args.put("x-dead-letter-exchange", DLX_EXCHANGE);
        args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);
        return QueueBuilder.durable(queueName).withArguments(args).build();
    }

    // 绑定...（略）
}

生产者根据延迟时间选择 routing key：

public String getRoutingKeyByDelay(long delayMillis) {
    if (delayMillis <= DELAY_1_MIN) return "delay.key.1m";
    else if (delayMillis <= DELAY_5_MIN) return "delay.key.5m";
    else if (delayMillis <= DELAY_30_MIN) return "delay.key.30m";
    else throw new IllegalArgumentException("Unsupported delay time");
}

这种方式虽然增加了队列数量，但能保证延迟精度，是生产环境的最佳实践。

死信交换机的高级用法

除了实现延迟队列，DLX 还可用于其他场景：

1. 消息重试 + 延迟退避

当消费者处理失败时，可以将消息重新发送到一个带有 TTL 的重试队列，实现'延迟后重试'。

2. 异常消息归档

将无法处理的消息路由到一个'死信归档队列'，便于后续人工排查。

常见问题与排查

问题 1：消息没有延迟，立即被消费了？

可能原因：

• 消费者监听的是 延迟队列 而不是 业务队列。
• 队列未正确配置 DLX 参数。
• 消息 TTL 设置错误（如设为 0 或负数）。

排查步骤：

1. 登录 RabbitMQ 管理界面（http://localhost:15672）。
2. 检查 delay.queue 的 Arguments 是否包含 x-dead-letter-exchange 和 x-message-ttl。
3. 查看消息是否进入了 business.queue。

问题 2：延迟时间不准确？

可能原因：

• 使用了单队列 + 动态 TTL，且队列中有更长 TTL 的消息阻塞。
• RabbitMQ 节点负载过高，导致定时器延迟。

解决方案：

• 采用多队列方案。
• 监控 RabbitMQ 资源使用情况。

问题 3：消息丢失？

可能原因：

• 队列或交换机未设置为 持久化（durable），RabbitMQ 重启后丢失。
• 消息未设置 deliveryMode=2（持久化消息）。

解决方案：

• 确保队列、交换机、消息均为持久化。

在 rabbitTemplate 中设置：

rabbitTemplate.setConfirmCallback(...);
rabbitTemplate.setReturnCallback(...);

性能与扩展性考量

吞吐量

• RabbitMQ 的延迟队列基于内存队列实现，高并发下可能成为瓶颈。
• 建议对延迟队列进行监控（消息堆积、内存使用）。

替代方案对比

📌 RabbitMQ 官方曾提供 Delayed Message Plugin，但由于维护问题，不推荐在生产环境使用。详情见：RabbitMQ Delayed Message Plugin Deprecation Notice

完整项目结构示例

src/
├── main/
│   ├── java/
│   │   └── com.example.delayqueue/
│   │       ├── DelayQueueApplication.java
│   │       ├── config/RabbitMQConfig.java
│   │       ├── service/DelayMessageProducer.java
│   │       ├── consumer/BusinessMessageConsumer.java
│   │       └── controller/DelayController.java
│   └── resources/
│       └── application.yml
└── test/

总结与最佳实践

通过本文，我们详细探讨了如何利用 RabbitMQ 的 死信交换机（DLX） 和 TTL 机制实现延迟队列。以下是关键总结：

1. 核心原理：消息在延迟队列中存活 TTL 时间后变为死信，经 DLX 路由到业务队列。
2. 动态延迟：通过消息级别的 expiration 实现，但需确保队列 TTL 足够大。
3. 精度保障：务必使用多队列方案，避免单队列导致的延迟不准问题。
4. 持久化：队列、交换机、消息均应设置为持久化，防止 RabbitMQ 重启丢失。
5. 监控：关注延迟队列的消息堆积情况，避免内存溢出。
6. 替代方案：对于超高精度或超长延迟需求，可考虑 Redis 或专用调度系统。

延迟队列是构建健壮分布式系统的基石之一。掌握 RabbitMQ 的这一技巧，不仅能解决实际业务问题，还能加深你对消息中间件底层机制的理解。

🌐 延伸阅读：RabbitMQ 官方文档 - TTLRabbitMQ 官方文档 - Dead Letter ExchangesSpring AMQP Reference Guide