"After all, the engineers only needed to refuse to fix anything, and modern industry would grind to a halt." -Michael Lewis

Enable Massive Growth

How to Setup a Reactive SQS Listener Using the AWS SDK and Spring Boot

Sep 2020

The source code for this post can be found on Github.

Following up on the previous post where we showed how to send SQS messages to Localstack using the AWS SDK for Java 2.0, we will now demonstrate how to write code that continuously polls for SQS messages, processes them, then deletes them off the queue.

The App

Building off of the work in the last post, where we had set up an SqsAsyncClient as a Bean:

public class AwsSqsConfig {

    public SqsAsyncClient amazonSQSAsyncClient() {
        return SqsAsyncClient.builder()
                .credentialsProvider(StaticCredentialsProvider.create(new AwsCredentials() {
                    public String accessKeyId() {
                        return "FAKE";

                    public String secretAccessKey() {
                        return "FAKE";

And where we had also set up a local SQS queue in localstack with the CLI:

export AWS_DEFAULT_REGION=us-east-1


aws --endpoint-url http://localhost:4566 sqs create-queue --queue-name "$QUEUE_NAME"

We can implement a simple SQS poller that will:

  • Use long polling, to efficiently only pull messages in a xxx second window if there are messages available to be pulled
  • Only poll if the previous poll has completed
  • Delete the message off the queue after processing

The code that can do that can look like:

public class SQSListenerBean {

    public static final Logger LOGGER = LoggerFactory.getLogger(SQSListenerBean.class);
    private final SqsAsyncClient sqsAsyncClient;
    private final String queueUrl;

    public SQSListenerBean(SqsAsyncClient sqsAsyncClient) {
        this.sqsAsyncClient = sqsAsyncClient;
        try {
            this.queueUrl = this.sqsAsyncClient.getQueueUrl(GetQueueUrlRequest.builder().queueName("my-queue").build()).get().queueUrl();
        } catch (Exception e) {
            throw new RuntimeException(e);

    public void continuousListener() {
        Mono<ReceiveMessageResponse> receiveMessageResponseMono = Mono.fromFuture(() ->

                .flatMap(messageFlux -> messageFlux)
                .subscribe(message -> {
                    LOGGER.info("message body: " + message.body());

                        .thenAccept(deleteMessageResponse -> {
                            LOGGER.info("deleted message with handle " + message.receiptHandle());

In this case, the actual processing of the message is just a log message printing out the message body.

If you start up the app, and send a sample message to that queue with:

export AWS_DEFAULT_REGION=us-east-1

Q_URL=$(aws --endpoint-url http://localhost:4566 sqs get-queue-url --queue-name "my-queue" --output text)
aws --endpoint-url http://localhost:4566 sqs send-message --queue-url "$Q_URL" --message-body "hey there"

You will see the application print out something like:

INFO 17716 --- [c-response-0-21] c.n.reactivesqs.SQSListenerBean          : message body: hey there
INFO 17716 --- [c-response-0-22] c.n.reactivesqs.SQSListenerBean          : deleted message with handle hwwmv...buncha letters...

You could further tweak this to your heart's content.

Nick Fisher is a software engineer in the Pacific Northwest. He focuses on building highly scalable and maintainable backend systems.