AWS SQS 监听与轮询

Question

我目前已经在 Fargate 上运行的 Spring Boot 项目中实现了 SQS 侦听器。

有可能在幕后，看似监听器的 SqsAsyncClient 实际上正在轮询。

另外，作为 PoC，我在不同的队列上实现了 Lambda 函数触发器。当队列中有项目并将其发布到我的服务时，这将被调用。这对我来说似乎不必要地复杂，但如果我只有一个服务实例，则可以消除单点故障。

我想我的主要困惑点是我是否不必要地担心 SQS 队列上的轮询与监听以及它是否重要。

用于示例目的的代码：

@Component
@Slf4j
@RequiredArgsConstructor
public class SqsListener {

private final SqsAsyncClient sqsAsyncClient;
private final Environment environment;
private final SmsMessagingServiceImpl smsMessagingService;

@PostConstruct
public void continuousListener() {
    String queueUrl = environment.getProperty("aws.sqs.sms.queueUrl");
    Mono<ReceiveMessageResponse> responseMono = receiveMessage(queueUrl);
    Flux<Message> messages = getItems(responseMono);
    messages.subscribe(message -> disposeOfFlux(message, queueUrl));
}

protected Flux<Message> getItems(Mono<ReceiveMessageResponse> responseMono) {
   return responseMono.repeat().retry()
            .map(ReceiveMessageResponse::messages)
            .map(Flux::fromIterable)
            .flatMap(messageFlux -> messageFlux);

}

protected void disposeOfFlux(Message message, String queueUrl) {
    log.info("Inbound SMS Received from SQS with MessageId: {}", message.messageId());
    if (someConditionIsMet()) 
        deleteMessage(queueUrl, message);
}

protected Mono<ReceiveMessageResponse> receiveMessage(String queueUrl) {
    return Mono.fromFuture(() -> sqsAsyncClient.receiveMessage(
                    ReceiveMessageRequest.builder()
                            .maxNumberOfMessages(5)
                            .messageAttributeNames("All")
                            .queueUrl(queueUrl)
                            .waitTimeSeconds(10)
                            .visibilityTimeout(30)
                            .build()));
}

protected void deleteMessage(String queueUrl, Message message) {
    sqsAsyncClient.deleteMessage(DeleteMessageRequest.builder()
                    .queueUrl(queueUrl)
                    .receiptHandle(message.receiptHandle())
                    .build())
            .thenAccept(deleteMessageResponse -> log.info("deleted message with handle {}", message.receiptHandle()));
}
}