Spring AMQP消息重试策略实现及死信队列配置问题排查

你好！咱们一步步来解决你的问题：首先明确Spring AMQP完全支持你需要的场景，然后给出具体实现方案，最后排查并修复你的死信队列配置问题。

一、Spring AMQP对目标场景的支持情况 #

是的，Spring AMQP结合RabbitMQ原生特性，完全能实现基础设施异常时消息退队延迟重试、重试超限触发告警的需求，核心依赖以下特性组合：

• RabbitMQ死信交换（DLX）+ 消息TTL实现延迟重试逻辑
• Spring AMQP内置重试机制或自定义重试次数控制
• 异常分类捕获与告警触发逻辑

二、目标场景的具体实现方案 #

1. 核心思路 #

当消费端捕获到Oracle宕机、Redis连接异常这类基础设施级异常时：

• 拒绝当前消息并不让它重新入队，触发死信机制将消息转入死信队列
• 死信队列配置TTL，消息过期后自动路由回原业务队列实现延迟重试
• 通过消息头或外部存储记录重试次数，达到阈值时向管理员发送告警邮件

2. 分步实现 #

（1）配置带死信特性的业务队列

给业务队列绑定死信交换，设置延迟重试时间：

@Configuration
public class MQConfig {
    public static final String BUSINESS_QUEUE = "my.business.queue";
    public static final String DLX_EXCHANGE = "my.dlx.exchange";
    public static final String DEAD_LETTER_QUEUE = "my.deadletter.queue";

    @Bean
    public Queue businessQueue() {
        Map<String, Object> args = new HashMap<>();
        // 绑定死信交换
        args.put("x-dead-letter-exchange", DLX_EXCHANGE);
        // 死信路由回原业务队列实现重试（也可配置专门重试队列）
        args.put("x-dead-letter-routing-key", BUSINESS_QUEUE);
        // 延迟重试时间（10秒）
        args.put("x-message-ttl", 10000);
        return new Queue(BUSINESS_QUEUE, true, false, false, args);
    }

    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange(DLX_EXCHANGE);
    }

    @Bean
    public Queue deadLetterQueue() {
        return new Queue(DEAD_LETTER_QUEUE);
    }

    // 死信队列绑定死信交换，用于归档最终重试失败的消息
    @Bean
    public Binding dlqBinding() {
        return BindingBuilder.bind(deadLetterQueue()).to(dlxExchange()).with(DEAD_LETTER_QUEUE);
    }
}

（2）消费端异常处理与重试控制

在监听方法中捕获基础设施异常，判断重试次数，触发告警：

@Component
public class MessageConsumer {
    private static final Logger LOGGER = LoggerFactory.getLogger(MessageConsumer.class);
    @Autowired
    private AlertService alertService; // 自定义告警服务，发送邮件

    @RabbitListener(queues = MQConfig.BUSINESS_QUEUE)
    public void handleMessage(ExampleObject exampleObject, Channel channel, 
                              @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
                              @Header(value = "retry-count", required = false) Integer retryCount) throws IOException {
        try {
            // 模拟业务逻辑：调用Oracle/Redis
            businessService.processData(exampleObject);
            channel.basicAck(deliveryTag, false); // 正常消费确认
        } catch (SQLRecoverableException | RedisConnectionFailureException e) {
            // 初始化重试次数
            int currentRetry = retryCount == null ? 1 : retryCount + 1;
            final int MAX_RETRY = 3;

            if (currentRetry >= MAX_RETRY) {
                // 达到最大重试次数，发送告警并归档消息
                alertService.sendInfrastructureAlert("Oracle/Redis异常，消息重试3次失败：" + exampleObject);
                // 路由到永久死信队列归档
                channel.basicPublish(MQConfig.DLX_EXCHANGE, MQConfig.DEAD_LETTER_QUEUE, null, 
                                     JSON.toJSONBytes(exampleObject));
                channel.basicAck(deliveryTag, false);
            } else {
                // 更新重试次数，拒绝消息让它进入死信队列延迟重试
                AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                        .headers(Map.of("retry-count", currentRetry))
                        .build();
                channel.basicReject(deliveryTag, false);
            }
        } catch (Exception e) {
            // 其他业务异常直接拒绝，不重试
            LOGGER.error("业务处理失败", e);
            channel.basicReject(deliveryTag, false);
        }
    }
}

（3）可选：使用Spring AMQP内置重试机制

如果不想手动管理重试次数，可通过RetryTemplate配置内置重试：

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setMessageConverter(jackson2JsonMessageConverter());
    factory.setRetryTemplate(retryTemplate());
    factory.setDefaultRequeueRejected(false); // 重试失败后进入死信队列
    return factory;
}

@Bean
public RetryTemplate retryTemplate() {
    RetryTemplate retryTemplate = new RetryTemplate();
    // 仅对基础设施异常重试，最多3次
    retryTemplate.setRetryPolicy(new SimpleRetryPolicy(3, Map.of(
            SQLRecoverableException.class, true,
            RedisConnectionFailureException.class, true
    )));
    // 指数退避延迟：第一次1秒，第二次2秒，第三次4秒
    ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
    backOffPolicy.setInitialInterval(1000);
    retryTemplate.setBackOffPolicy(backOffPolicy);
    return retryTemplate;
}

三、你的代码死信队列不生效的问题排查 #

看了你的代码，发现几个关键问题导致消息无法进入死信队列，反而出现无限循环：

1. 生产者与消费者队列不匹配 #

你用outgoingSender发送消息到OUTGOING_QUEUE，但@RabbitListener监听的是INCOMING_QUEUE，消息根本没被消费，自然不会触发死信逻辑。

2. 异常抛出位置错误 #

你在生产者端的sender方法中抛出AmqpRejectAndDontRequeueException，但这个异常只有在消费端抛出才会触发死信机制，生产者抛出异常只会导致消息发送失败，不会进入死信队列。

3. 数组越界导致定时任务崩溃 #

循环条件for (int i = 0; i <= int1.length; i++)会导致ArrayIndexOutOfBoundsException（数组长度为5，索引最大为4），整个定时任务失败，消息无法正常发送。

修复后的核心代码示例 #

修正队列监听关联

// 改为监听发送消息的OUTGOING_QUEUE
@RabbitListener(queues = MQConfig.OUTGOING_QUEUE)
public void handleMessage(ExampleObject exampleObject, Channel channel, 
                          @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
    LOGGER.info("Received object: " + exampleObject.getValue());
    try {
        // 模拟Oracle宕机异常
        if (exampleObject.getValue() == 20) {
            throw new SQLRecoverableException("Oracle server down");
        }
        channel.basicAck(deliveryTag, false);
    } catch (SQLRecoverableException e) {
        // 拒绝消息，触发死信逻辑
        channel.basicReject(deliveryTag, false);
    }
}

修正生产者循环逻辑

@Scheduled(fixedDelay = 5000)
public void sender() {
    Integer int1[] = new Integer[]{10,20,30,40,50};
    // 修正循环条件，避免数组越界
    for (int i = 0; i < int1.length; i++){
        ExampleObject ex = new ExampleObject();
        ex.setValue(int1[i]);
        LOGGER.info("Sending object: " + ex.getValue());
        outgoingSender.convertAndSend(ex);
    }
}

修正死信交换配置（避免名称混淆）

建议死信交换名称与死信队列名称区分开，避免路由混乱：

@Bean
public DirectExchange dlx() {
    return new DirectExchange("my.dlx.exchange");
}

调整后，当消费端捕获到基础设施异常并拒绝消息时，消息会进入死信队列，等待TTL到期后自动路由回原队列实现延迟重试，达到最大次数后可触发告警。

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