如何在Spring的SQS监听器中实现指数退避重试策略？

方案1：结合Spring Retry与SQS可见性超时调整 #

通过动态修改消息的可见性超时，搭配Spring Retry控制重试逻辑，实现指数退避效果，完美匹配你现有SQS队列的最大接收次数限制。

步骤1：引入依赖 #

如果使用Maven，在pom.xml中添加Spring Retry及AOP依赖：

<dependency>
    <groupId>org.springframework.retry</groupId>
    <artifactId>spring-retry</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-aop</artifactId>
</dependency>

步骤2：配置指数退避的RetryTemplate #

创建配置类，定义符合需求的重试模板：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

@Configuration
public class SqsRetryConfig {

    @Bean
    public RetryTemplate sqsRetryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();

        // 指数退避策略：初始间隔30秒（对应SQS默认可见性超时），每次间隔翻倍，最大限制为300秒
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(30000);
        backOffPolicy.setMultiplier(2);
        backOffPolicy.setMaxInterval(300000);

        // 重试次数设置为2次（首次消费+2次重试，刚好匹配SQS最大接收次数3）
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(2);

        retryTemplate.setBackOffPolicy(backOffPolicy);
        retryTemplate.setRetryPolicy(retryPolicy);
        return retryTemplate;
    }
}

步骤3：在监听器中应用重试逻辑 #

注入RetryTemplate和AmazonSQS客户端，在消息处理失败时动态调整可见性超时：

import com.amazonaws.services.sqs.AmazonSQS;
import com.amazonaws.services.sqs.model.ChangeMessageVisibilityRequest;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;
import io.awspring.cloud.sqs.annotation.SqsListener;

@Component
public class SqsMessageListener {

    private final RetryTemplate sqsRetryTemplate;
    private final AmazonSQS amazonSQS;

    public SqsMessageListener(RetryTemplate sqsRetryTemplate, AmazonSQS amazonSQS) {
        this.sqsRetryTemplate = sqsRetryTemplate;
        this.amazonSQS = amazonSQS;
    }

    @SqsListener("your-queue-name")
    public void listen(String message, String queueUrl, String receiptHandle) {
        try {
            sqsRetryTemplate.execute((RetryCallback<Void, Exception>) context -> {
                // 替换为你的实际业务处理逻辑
                processMessage(message);
                return null;
            }, context -> {
                // 重试失败时，根据退避策略调整消息可见性超时
                long backOffPeriod = context.getBackOffContext().getSleepPeriod();
                amazonSQS.changeMessageVisibility(new ChangeMessageVisibilityRequest()
                        .withQueueUrl(queueUrl)
                        .withReceiptHandle(receiptHandle)
                        .withVisibilityTimeout((int) (backOffPeriod / 1000)));
                return null;
            });
        } catch (Exception e) {
            // 所有重试失败后，可在此记录日志，消息会自动进入死信队列（若已配置）
        }
    }

    private void processMessage(String message) throws Exception {
        // 业务逻辑处理失败时抛出异常，触发重试
        throw new Exception("消息处理失败，触发指数退避重试");
    }
}

方案2：自定义MessageListenerContainer错误处理器 #

若不想引入Spring Retry，可直接通过自定义容器的错误处理器，基于SQS消息的接收次数计算退避超时：

步骤1：配置监听器容器工厂 #

import com.amazonaws.services.sqs.AmazonSQS;
import com.amazonaws.services.sqs.model.ChangeMessageVisibilityRequest;
import io.awspring.cloud.sqs.listener.SqsMessageListenerContainerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class SqsContainerConfig {

    @Bean
    public SqsMessageListenerContainerFactory<?> sqsListenerContainerFactory(AmazonSQS amazonSQS) {
        return SqsMessageListenerContainerFactory.builder()
                .configureContainer(container -> {
                    container.setErrorHandler((msg, exception) -> {
                        // 获取当前消息的接收次数
                        int receiveCount = Integer.parseInt(msg.getAttributes().get("ApproximateReceiveCount"));
                        // 计算指数退避超时：30秒 * 2^(接收次数-1)，最大限制300秒
                        long timeout = 30 * (long) Math.pow(2, receiveCount - 1);
                        timeout = Math.min(timeout, 300);

                        // 更新消息可见性超时
                        amazonSQS.changeMessageVisibility(new ChangeMessageVisibilityRequest()
                                .withQueueUrl(msg.getQueueUrl())
                                .withReceiptHandle(msg.getReceiptHandle())
                                .withVisibilityTimeout((int) timeout));
                    });
                })
                .build();
    }
}

步骤2：绑定容器工厂到监听器 #

import io.awspring.cloud.sqs.annotation.SqsListener;
import org.springframework.stereotype.Component;

@Component
public class SqsMessageListener {

    @SqsListener(value = "your-queue-name", containerFactory = "sqsListenerContainerFactory")
    public void listen(String message) throws Exception {
        // 业务逻辑，失败时抛出异常触发退避逻辑
        processMessage(message);
    }

    private void processMessage(String message) throws Exception {
        throw new Exception("消息处理失败，触发退避");
    }
}

注意事项 #

• 两种方案都需注意SQS可见性超时的上限为12小时，需合理设置最大退避间隔
• 若配置了死信队列，消息达到最大接收次数后会自动转入死信队列，无需额外处理

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