抛出AmqpRejectAndDontRequeueException时Spring AMQP的RetryCache未清空

问题 #

我编写了一个简单的Rabbit监听器，用于测试其处理多条无效消息的能力，该监听器始终抛出AmqpRejectAndDontRequeueException。相关配置代码如下：

监听器代码 #

@RabbitListener(
    id = TestConfig.LISTENER_ID,
    queues = TestConfig.DATA_QUEUE,
    containerFactory = TestConfig.LISTENER_FACTORY
)
public void consume(String data, Message message) {
    if (true) {
        throw new AmqpRejectAndDontRequeueException("don't requeue");
    }
}

配置类代码 #

static final String LISTENER_ID = "listenerId";

static final String DATA_QUEUE = "data.queue";

static final String LISTENER_FACTORY = "listenerFactory";

private final AmqpAdmin amqpAdmin;

private final ConnectionFactory connectionFactory;

TestConfig(AmqpAdmin amqpAdmin, ConnectionFactory connectionFactory) {
    this.amqpAdmin = amqpAdmin;
    this.connectionFactory = connectionFactory;
}

@Bean
RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory) {
    return new RabbitTransactionManager(connectionFactory);
}

@Bean
public Queue consumedDataQueue() {
    Queue queue = new Queue(DATA_QUEUE);
    queue.setAdminsThatShouldDeclare(amqpAdmin);
    return queue;
}

@Bean(name = LISTENER_FACTORY)
public SimpleRabbitListenerContainerFactory listenerFactory(RabbitTransactionManager rabbitTransactionManager) {

    StatefulRetryOperationsInterceptor backOffRetryInterceptor = statefulRetryOperationsInterceptor();
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();

    factory.setConnectionFactory(connectionFactory);
    factory.setConcurrentConsumers(1);
    factory.setAutoStartup(true);
    factory.setTransactionManager(rabbitTransactionManager);
    factory.setAdviceChain(backOffRetryInterceptor);

    return factory;
}

private StatefulRetryOperationsInterceptor statefulRetryOperationsInterceptor() {

    RejectAndDontRequeueRecoverer messageRecoverer = new RejectAndDontRequeueRecoverer(); // to be changed

    RetryTemplate retryTemplate = new RetryTemplate();
    retryTemplate.setRetryContextCache(new MapRetryContextCache(3));
    retryTemplate.setRetryPolicy(new SimpleRetryPolicy(1));

    return RetryInterceptorBuilder.stateful()
      .retryOperations(retryTemplate)
      .recoverer(messageRecoverer)
      .build();
}

当监听器抛出该异常时，MapRetryContextCache持续填充但未被清空，最终应用抛出RetryCacheCapacityExceededException。我尝试通过在监听器中抛出自定义异常，并在自定义消息恢复器中处理的方式解决，但此时消息仍会按照重试策略重新入队一次。

想请教：

• 我的操作哪里有误？
• 是否不应在有状态拦截器中使用AmqpRejectAndDontRequeueException？
• 如何在有状态重试拦截器中实现拒绝消息且不重新入队？

问题分析与解决 #

1. 核心错误点 #

你在有状态重试拦截器中直接抛出AmqpRejectAndDontRequeueException的做法不符合拦截器的工作逻辑：

• 有状态重试依赖RetryContextCache跟踪每条消息的重试状态，而AmqpRejectAndDontRequeueException会被Rabbit容器直接识别为「无需重试」的信号，跳过拦截器的后续流程，导致缓存中的重试上下文条目无法被清理，最终堆积触发RetryCacheCapacityExceededException。
• 改用自定义异常后消息仍重试一次，是因为你配置的SimpleRetryPolicy(1)表示最多重试1次（即首次执行+1次重试，共2次），这是配置的预期行为，而非错误。

2. 正确实现方案 #

要在有状态重试拦截器中实现「拒绝消息且不重新入队」，需遵循以下步骤：

（1）替换监听器中的异常类型

不要直接抛出AmqpRejectAndDontRequeueException，改用自定义业务异常，让重试拦截器接管异常处理：

public class InvalidMessageException extends RuntimeException {
    public InvalidMessageException(String message) {
        super(message);
    }
}

// 监听器代码
public void consume(String data, Message message) {
    throw new InvalidMessageException("无效消息，拒绝且不重入队");
}

（2）调整重试策略，让自定义异常直接进入恢复流程

修改SimpleRetryPolicy，指定自定义异常不参与重试，直接触发恢复逻辑：

private StatefulRetryOperationsInterceptor statefulRetryOperationsInterceptor() {
    RejectAndDontRequeueRecoverer messageRecoverer = new RejectAndDontRequeueRecoverer();

    // 配置异常重试规则：自定义异常不重试
    Map<Class<? extends Throwable>, Boolean> retryRules = new HashMap<>();
    retryRules.put(InvalidMessageException.class, false);
    // 可添加其他需要重试的异常，比如NullPointerException.class -> true

    RetryTemplate retryTemplate = new RetryTemplate();
    retryTemplate.setRetryContextCache(new MapRetryContextCache(3));
    // 最大重试次数1，同时结合异常规则过滤
    retryTemplate.setRetryPolicy(new SimpleRetryPolicy(1, retryRules));

    return RetryInterceptorBuilder.stateful()
            .retryOperations(retryTemplate)
            .recoverer(messageRecoverer)
            .build();
}

如果希望所有异常都不重试，直接将重试次数设为0即可：

retryTemplate.setRetryPolicy(new SimpleRetryPolicy(0));

（3）确保缓存自动清理

当异常进入RejectAndDontRequeueRecoverer后，有状态拦截器会自动清理RetryContextCache中的对应条目，不会出现缓存堆积。恢复器会向RabbitMQ发送拒绝指令，且不会将消息重新入队，完全符合需求。

3. 额外注意事项 #

• 配合RabbitTransactionManager使用时，拒绝消息的操作会在事务提交后执行，保证事务一致性。
• 不要混淆容器级异常处理和拦截器重试逻辑：AmqpRejectAndDontRequeueException是直接通知容器跳过重试，而有状态重试拦截器的逻辑在容器之上，直接抛出该异常会绕过拦截器的缓存清理流程。

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