Java AWS SQS消费者仅手动调用才消费，自动触发问题排查

问题描述 #

我正在使用AWS SDK for Java（版本2.21.0）开发AWS SQS队列的消费者。目前消费者能正常拉取并处理消息，但只有手动调用消费方法时才会执行，无法在消息发布后自动触发消费。需要排查问题原因，实现消息发布后自动处理的需求。

提供的代码 #

AmazonSQSAsync Bean定义 #

@Bean
@Primary
public AmazonSQSAsync amazonSQSAsync() {
    BasicAWSCredentials credentials = new BasicAWSCredentials(Enviromentals.KeyProvider.ACCESS_KEY,
            Enviromentals.KeyProvider.SECRET_KEY);
    return AmazonSQSAsyncClientBuilder
            .standard()
            .withRegion(Regions.US_EAST_1)
            .withCredentials(new AWSStaticCredentialsProvider(credentials))
            .build();
}

消费者方法 #

public List<Message> consumeMessageFromSQS() {
    AmazonSQS sqsClient = sqsClientBuilder();
    System.out.println("al escuchar entro a este metodo");
    //ReceiveMessageRequest request = new ReceiveMessageRequest(Enviromentals.AwsEnvs.DEPOSIT_QUEUE_ENDPOINT_URI).withMaxNumberOfMessages(10);

    List<Message> sqsMessages = sqsClient.receiveMessage(Enviromentals.AwsEnvs.DEPOSIT_QUEUE_ENDPOINT_URI).getMessages();
    for (Message message : sqsMessages) {
        //run process for message
        System.out.println(message.getBody());
        
        //dequeue message after using it
        //also perfect step so check if message was successfully processed
        dequeuMessageFromSQS(message);
    }
    return sqsMessages;
}

问题根源 #

当前代码的消费逻辑是单次同步拉取，仅在手动调用consumeMessageFromSQS方法时才会执行一次拉取操作，没有持续监听队列的机制，自然无法在有新消息时自动触发处理。此外还存在两个小问题：

• 消费方法中手动创建SQS客户端，没有复用已定义的AmazonSQSAsync Bean，造成资源浪费
• 未启用SQS长轮询，会导致频繁的空请求，增加API调用次数和延迟

解决方案 #

方案1：使用Spring Cloud AWS的@SqsListener（推荐，适合Spring环境） #

如果你的项目是Spring/Spring Boot项目，直接用Spring Cloud AWS的消息监听注解可以快速实现自动消费：

1. 确保引入对应版本的Spring Cloud AWS依赖（与你的AWS SDK 2.21.0匹配）
2. 创建监听类，用@SqsListener指定队列URL或名称：

@Component
public class SQSMessageListener {

    private final AmazonSQSAsync amazonSQSAsync;

    // 注入已定义的SQS客户端Bean
    public SQSMessageListener(AmazonSQSAsync amazonSQSAsync) {
        this.amazonSQSAsync = amazonSQSAsync;
    }

    @SqsListener(value = "${aws.sqs.deposit-queue-uri}") // 替换为你的队列URL或名称
    public void handleMessage(Message message) {
        // 处理消息逻辑
        System.out.println("收到消息：" + message.getBody());
        
        // 处理完成后删除消息
        amazonSQSAsync.deleteMessage(
            Enviromentals.AwsEnvs.DEPOSIT_QUEUE_ENDPOINT_URI,
            message.getReceiptHandle()
        );
    }
}

Spring会自动启动监听线程，新消息到达时自动触发handleMessage方法。

方案2：手动实现后台轮询（无Spring Cloud依赖时使用） #

在应用启动时启动一个后台线程，循环调用消费逻辑，并启用SQS长轮询减少空请求：

@Component
public class SQSMessagePoller {

    private final AmazonSQSAsync amazonSQSAsync;
    private final String queueUrl = Enviromentals.AwsEnvs.DEPOSIT_QUEUE_ENDPOINT_URI;
    private volatile boolean running = true;

    public SQSMessagePoller(AmazonSQSAsync amazonSQSAsync) {
        this.amazonSQSAsync = amazonSQSAsync;
    }

    @PostConstruct
    public void startPolling() {
        new Thread(() -> {
            while (running) {
                ReceiveMessageRequest request = new ReceiveMessageRequest(queueUrl)
                        .withMaxNumberOfMessages(10)
                        .withWaitTimeSeconds(20); // 启用长轮询，最长等待20秒

                List<Message> messages = amazonSQSAsync.receiveMessage(request).getMessages();
                for (Message message : messages) {
                    // 处理消息
                    System.out.println("收到消息：" + message.getBody());
                    // 删除消息
                    amazonSQSAsync.deleteMessage(queueUrl, message.getReceiptHandle());
                }
            }
        }).start();
    }

    @PreDestroy
    public void stopPolling() {
        running = false;
    }
}

• withWaitTimeSeconds(20)启用长轮询，SQS会等待20秒直到有消息或超时，减少空轮询次数
• 用@PostConstruct在应用启动时启动线程，@PreDestroy在应用关闭时停止线程

方案3：使用SDK异步API实现监听 #

利用AmazonSQSAsync的异步方法结合回调，实现非阻塞的持续监听：

@Component
public class SQSAsyncListener {

    private final AmazonSQSAsync amazonSQSAsync;
    private final String queueUrl = Enviromentals.AwsEnvs.DEPOSIT_QUEUE_ENDPOINT_URI;

    public SQSAsyncListener(AmazonSQSAsync amazonSQSAsync) {
        this.amazonSQSAsync = amazonSQSAsync;
    }

    @PostConstruct
    public void startListening() {
        receiveMessagesAsync();
    }

    private void receiveMessagesAsync() {
        ReceiveMessageRequest request = new ReceiveMessageRequest(queueUrl)
                .withMaxNumberOfMessages(10)
                .withWaitTimeSeconds(20);

        amazonSQSAsync.receiveMessageAsync(request, new AsyncHandler<ReceiveMessageRequest, ReceiveMessageResult>() {
            @Override
            public void onSuccess(ReceiveMessageRequest request, ReceiveMessageResult result) {
                List<Message> messages = result.getMessages();
                for (Message message : messages) {
                    // 处理消息
                    System.out.println("收到消息：" + message.getBody());
                    // 删除消息
                    amazonSQSAsync.deleteMessageAsync(queueUrl, message.getReceiptHandle());
                }
                // 递归调用，继续监听
                receiveMessagesAsync();
            }

            @Override
            public void onError(Exception exception) {
                // 处理错误，比如重试
                System.err.println("监听出错：" + exception.getMessage());
                // 延迟后重试
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                receiveMessagesAsync();
            }
        });
    }
}

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