如何在AMQP消息发布到DLQ前添加Header实现TTL指数退避？

完全可以实现这个需求，RejectAndDontRequeueRecoverer是可行的基础方向，但需要结合自定义重试次数跟踪、TTL计算逻辑以及死信队列的循环转发配置来完成指数退避效果。以下是具体实现方案：

1. 核心配置思路 #

默认情况下，RabbitMQ将消息转发到死信队列时，会沿用原队列的TTL，不会自动修改。所以需要：

• 给业务队列和死信队列配置双向死信转发（死信队列过期后将消息转回业务队列）
• 跟踪消息的重试次数，基于次数计算指数退避TTL
• 在消息进入死信队列前，手动设置expiration header覆盖队列TTL

2. 死信队列与交换机配置 #

@Bean
public Queue businessQueue() {
    return QueueBuilder.durable("business-queue")
            // 绑定死信交换机和路由键
            .deadLetterExchange("dlx-exchange")
            .deadLetterRoutingKey("dlk-routing-key")
            .build();
}

@Bean
public Queue dlqQueue() {
    return QueueBuilder.durable("dlq-queue")
            // 死信过期后转回业务队列交换机，实现重试循环
            .deadLetterExchange("business-exchange")
            .deadLetterRoutingKey("business-routing-key")
            .build();
}

@Bean
public DirectExchange dlxExchange() {
    return new DirectExchange("dlx-exchange");
}

@Bean
public DirectExchange businessExchange() {
    return new DirectExchange("business-exchange");
}

@Bean
public Binding dlqBinding() {
    return BindingBuilder.bind(dlqQueue()).to(dlxExchange()).with("dlk-routing-key");
}

@Bean
public Binding businessBinding() {
    return BindingBuilder.bind(businessQueue()).to(businessExchange()).with("business-routing-key");
}

3. 自定义指数退避恢复器 #

扩展MessageRecoverer，实现重试次数跟踪、TTL计算，并手动将消息发送到死信队列：

@Component
public class ExponentialBackoffDlqRecoverer implements MessageRecoverer {

    private final RabbitTemplate rabbitTemplate;
    private final String dlxExchange;
    private final String dlxRoutingKey;
    // 最大重试次数，避免无限循环
    private static final int MAX_RETRY_COUNT = 5;

    public ExponentialBackoffDlqRecoverer(RabbitTemplate rabbitTemplate,
                                          @Value("${rabbitmq.dlx.exchange}") String dlxExchange,
                                          @Value("${rabbitmq.dlx.routing-key}") String dlxRoutingKey) {
        this.rabbitTemplate = rabbitTemplate;
        this.dlxExchange = dlxExchange;
        this.dlxRoutingKey = dlxRoutingKey;
    }

    @Override
    public void recover(Message message, Throwable cause) {
        MessageProperties props = message.getMessageProperties();
        // 获取当前重试次数，默认0
        Integer retryCount = props.getHeader("x-retry-count");
        retryCount = retryCount == null ? 0 : retryCount + 1;

        if (retryCount >= MAX_RETRY_COUNT) {
            // 达到最大重试次数，发送到最终死信队列（可单独配置）
            rabbitTemplate.send("final-dlx-exchange", "final-dlk-routing-key", message);
            return;
        }

        // 指数退避计算TTL：2^重试次数 * 10000ms（即10s、20s、40s...）
        long ttl = (long) Math.pow(2, retryCount) * 10000;
        // 更新重试次数和TTL header
        props.setHeader("x-retry-count", retryCount);
        props.setExpiration(String.valueOf(ttl));

        // 发送到死信队列等待过期重试
        rabbitTemplate.send(dlxExchange, dlxRoutingKey, message);
    }
}

4. 配置监听容器与重试拦截器 #

将自定义恢复器绑定到监听容器，关闭本地重试（完全依赖死信队列重试）：

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
        ConnectionFactory connectionFactory,
        ExponentialBackoffDlqRecoverer recoverer) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setAdviceChain(retryInterceptor(recoverer));
    return factory;
}

private RetryOperationsInterceptor retryInterceptor(MessageRecoverer recoverer) {
    return RetryInterceptorBuilder.stateless()
            .maxAttempts(1) // 本地只尝试1次，失败直接进入死信队列
            .recoverer(recoverer)
            .build();
}

关键注意事项 #

• expiration header优先级高于队列TTL，只要正确设置该值，RabbitMQ会使用它作为消息过期时间
• 必须设置死信队列的x-dead-letter-exchange指向业务队列交换机，才能实现重试循环
• 要限制最大重试次数，避免消息无限在业务队列和死信队列间循环

内容的提问来源于stack exchange，提问作者Roberto