Amazon SQS Java Messaging Lib

The Amazon SQS Java Messaging Library provides an asynchronous, batched messaging client for Amazon SQS, supporting both AWS SDK v1 (AmazonSQS) and v2 (SqsClient). It features configurable batching with linger time, FIFO ordering, message attributes, and Micrometer metrics.

The batch size should be chosen based on the size of individual messages and available network bandwidth as well as the observed latency and throughput improvements based on the real life load. These are configured to some sensible defaults assuming smaller message sizes and the optimal batch size for server side processing.

For detailed architecture, threading model, batching behavior, and exception handling, see the Technical Guide.

Request Batch

Combine multiple requests to optimally utilise the network.

Article Martin Fowler Request Batch

Compatible JDK 8, 11, 17, 21 and 25

Compatible AWS JDK v1 >= 1.12

Compatible AWS JDK v2 >= 2.18

This library supports Kotlin as well

1. Quick Start

1.1 Prerequisite

In order to use Amazon SQS Java Messaging Lib within a Maven project, simply add the following dependency to your pom.xml. There are no other dependencies for Amazon SQS Java Messaging Lib, which means other unwanted libraries will not overwhelm your project.

You can pull it from the central Maven repositories:

Maven

For AWS SDK v1

<dependency>
    <groupId>com.github.mvallim</groupId>
    <artifactId>amazon-sqs-java-messaging-lib-v1</artifactId>
    <version>1.4.0</version>
</dependency>

For AWS SDK v2

<dependency>
    <groupId>com.github.mvallim</groupId>
    <artifactId>amazon-sqs-java-messaging-lib-v2</artifactId>
    <version>1.4.0</version>
</dependency>

If you want to try a snapshot version, add the following repository:

<repository>
    <id>sonatype-snapshots</id>
    <name>Sonatype Snapshots</name>
    <url>https://oss.sonatype.org/content/repositories/snapshots</url>
    <snapshots>
        <enabled>true</enabled>
    </snapshots>
</repository>

Gradle

For AWS SDK v1

implementation 'com.github.mvallim:amazon-sqs-java-messaging-lib-v1:1.4.0'

For AWS SDK v2

implementation 'com.github.mvallim:amazon-sqs-java-messaging-lib-v2:1.4.0'

If you want to try a snapshot version, add the following repository:

repositories {
  maven {
    url "https://oss.sonatype.org/content/repositories/snapshots"
  }
}

1.2 Usage

Properties QueueProperty

Property	Type	Description
fifo	boolean	refers if SQS is fifo or not.
maximumPoolSize	int	refers maximum threads for producer.
queueUrl	string	refers queue url.
linger	int	refers to the time to wait before sending messages out to SQS.
maxBatchSize	int	refers to the maximum amount of data to be collected before sending the batch.

Note

The buffer of message store in memory is calculated using maximumPoolSize * maxBatchSize; huge values demand huge memory.

Note on effective capacity: the default queue implementation (RingBufferBlockingQueue) internally rounds its capacity up to the next power of two, to allow fast bitwise index calculation. This means the actual allocated capacity may be up to ~2x the value computed above — e.g. maximumPoolSize=10 and maxBatchSize=10 yields a requested capacity of 100, but the queue actually allocates 128 slots. If you need to budget memory precisely, use RingBufferBlockingQueue#remainingCapacity() (or a LinkedBlockingQueue via the Custom BlockingQueue option below, which does not round up) rather than relying on the maximumPoolSize * maxBatchSize formula as an exact figure.

Custom BlockingQueue

final QueueProperty queueProperty = QueueProperty.builder()
  .fifo(false)
  .linger(100)
  .maxBatchSize(10)
  .maximumPoolSize(20)
  .queueUrl("http://localhost:4566/000000000000/queue")
  .build();

AmazonSqsTemplate<MyMessage> sqsTemplate = AmazonSqsTemplate.builder(amazonSQS, queueProperty)
  .queueRequests(new LinkedBlockingQueue<>(100))
  .build();

Custom ObjectMapper

final QueueProperty queueProperty = QueueProperty.builder()
  .fifo(false)
  .linger(100)
  .maxBatchSize(10)
  .maximumPoolSize(20)
  .queueUrl("http://localhost:4566/000000000000/queue")
  .build();

AmazonSqsTemplate<MyMessage> sqsTemplate = AmazonSqsTemplate.builder(amazonSQS, queueProperty)
  .objectMapper(new ObjectMapper())
  .build();

Custom BlockingQueue and ObjectMapper

final QueueProperty queueProperty = QueueProperty.builder()
  .fifo(false)
  .linger(100)
  .maxBatchSize(10)
  .maximumPoolSize(20)
  .queueUrl("http://localhost:4566/000000000000/queue")
  .build();

AmazonSqsTemplate<MyMessage> sqsTemplate = AmazonSqsTemplate.builder(amazonSQS, queueProperty)
  .queueRequests(new LinkedBlockingQueue<>(100))
  .objectMapper(new ObjectMapper())
  .build();

With Micrometer metrics

final QueueProperty queueProperty = QueueProperty.builder()
  .fifo(false)
  .linger(100)
  .maxBatchSize(10)
  .maximumPoolSize(20)
  .queueUrl("http://localhost:4566/000000000000/queue")
  .build();

AmazonSqsTemplate<MyMessage> sqsTemplate = AmazonSqsTemplate.builder(amazonSQS, queueProperty)
  .meterRegistry(new SimpleMeterRegistry())
  .build();

Standard SQS

final QueueProperty queueProperty = QueueProperty.builder()
  .fifo(false)
  .linger(100)
  .maxBatchSize(10)
  .maximumPoolSize(20)
  .queueUrl("http://localhost:4566/000000000000/queue")
  .build();

AmazonSqsTemplate<MyMessage> sqsTemplate = AmazonSqsTemplate.builder(amazonSQS, queueProperty).build();

final RequestEntry<MyMessage> requestEntry = RequestEntry.builder()
  .withValue(new MyMessage())
  .withMessageHeaders(Map.of())
  .build();

sqsTemplate.send(requestEntry);

FIFO SQS

final QueueProperty queueProperty = QueueProperty.builder()
  .fifo(true)
  .linger(100)
  .maxBatchSize(10)
  .maximumPoolSize(1)
  .queueUrl("http://localhost:4566/000000000000/queue")
  .build();

AmazonSqsTemplate<MyMessage> sqsTemplate = AmazonSqsTemplate.builder(amazonSQS, queueProperty).build();

final RequestEntry<MyMessage> requestEntry = RequestEntry.builder()
  .withValue(new MyMessage())
  .withMessageHeaders(Map.of())
  .withGroupId(UUID.randomUUID().toString())
  .withDeduplicationId(UUID.randomUUID().toString())
  .build();

sqsTemplate.send(requestEntry);

Send With Callback

final QueueProperty queueProperty = QueueProperty.builder()
  .fifo(true)
  .linger(100)
  .maxBatchSize(10)
  .maximumPoolSize(1)
  .queueUrl("http://localhost:4566/000000000000/queue")
  .build();

AmazonSqsTemplate<MyMessage> sqsTemplate = AmazonSqsTemplate.builder(amazonSQS, queueProperty).build();

final RequestEntry<MyMessage> requestEntry = RequestEntry.builder()
  .withValue(new MyMessage())
  .withMessageHeaders(Map.of())
  .withGroupId(UUID.randomUUID().toString())
  .withDeduplicationId(UUID.randomUUID().toString())
  .build();

sqsTemplate.send(requestEntry).addCallback(
  success -> LOGGER.info("Sent: {}", success.getMessageId()),
  failure -> LOGGER.error("Failed: {} [{}]", failure.getMessage(), failure.getCode())
);

sqsTemplate.send(requestEntry).addCallback(
  success -> LOGGER.info("Sent: {}", success.getMessageId())
);

Send And Wait

final QueueProperty queueProperty = QueueProperty.builder()
  .fifo(true)
  .linger(100)
  .maxBatchSize(10)
  .maximumPoolSize(1)
  .queueUrl("http://localhost:4566/000000000000/queue")
  .build();

AmazonSqsTemplate<MyMessage> sqsTemplate = AmazonSqsTemplate.builder(amazonSQS, queueProperty).build();

final RequestEntry<MyMessage> requestEntry = RequestEntry.builder()
  .withValue(new MyMessage())
  .withMessageHeaders(Map.of())
  .withGroupId(UUID.randomUUID().toString())
  .withDeduplicationId(UUID.randomUUID().toString())
  .build();

sqsTemplate.send(requestEntry).addCallback(
  success -> LOGGER.info("Sent: {}", success.getMessageId()),
  failure -> LOGGER.error("Failed: {} [{}]", failure.getMessage(), failure.getCode())
);

sqsTemplate.await().join();

Send And Shutdown

final QueueProperty queueProperty = QueueProperty.builder()
  .fifo(true)
  .linger(100)
  .maxBatchSize(10)
  .maximumPoolSize(1)
  .queueUrl("http://localhost:4566/000000000000/queue")
  .build();

AmazonSqsTemplate<MyMessage> sqsTemplate = AmazonSqsTemplate.builder(amazonSQS, queueProperty).build();

final RequestEntry<MyMessage> requestEntry = RequestEntry.builder()
  .withValue(new MyMessage())
  .withMessageHeaders(Map.of())
  .withGroupId(UUID.randomUUID().toString())
  .withDeduplicationId(UUID.randomUUID().toString())
  .build();

sqsTemplate.send(requestEntry).addCallback(
  success -> LOGGER.info("Sent: {}", success.getMessageId()),
  failure -> LOGGER.error("Failed: {} [{}]", failure.getMessage(), failure.getCode())
);

sqsTemplate.shutdown();

Full Example with Builder

QueueProperty queueProperty = QueueProperty.builder()
  .fifo(false)
  .linger(100L)
  .maxBatchSize(10)
  .maximumPoolSize(5)
  .queueUrl("http://localhost:4566/000000000000/queue")
  .build();

AmazonSqsTemplate<MyMessage> template = AmazonSqsTemplate.builder(sqsClient, queueProperty)
  .meterRegistry(new SimpleMeterRegistry())
  .queueRequests(new RingBufferBlockingQueue<>(1024))
  .objectMapper(new ObjectMapper())
  .build();

template.send(RequestEntry.<MyMessage>builder()
  .withValue(new MyMessage("hello"))
  .withMessageHeaders(Map.of("source", "app-1"))
  .withGroupId(UUID.randomUUID().toString())
  .build());

template.await().thenRun(template::shutdown).join();

---

Metrics

When a MeterRegistry is provided via the builder, the library records these Micrometer metrics:

SQS Publish

Tags: queue = <queueUrl>

Metric	Type	Description
sqs.publish.attempts	Counter	Total SendMessageBatch attempts
sqs.publish.success	Counter	Successful messages
sqs.publish.failure	Counter	Failed messages (dynamic tags: error_code, error_type)
sqs.publish.duration	Timer	Publish latency (p50/p95/p99)
sqs.publish.batch.size	DistributionSummary	Messages per batch
sqs.publish.inflight	Gauge	In-flight publish batches

Blocking Queue

Tags: name = <queueName>

Metric	Type	Description
blocking.queue.puts.total	Counter	Successful put operations
blocking.queue.puts.failed	Counter	Put operations that threw an exception
blocking.queue.put.duration	Timer	Put latency (percentile histogram)
blocking.queue.takes.total	Counter	Successful take operations
blocking.queue.takes.failed	Counter	Take operations that threw an exception
blocking.queue.take.duration	Timer	Take latency (percentile histogram)
blocking.queue.size	Gauge	Current queue depth

Executor

Tags: name = <executorName>

Metric	Type	Description
executor.active	Gauge	Tasks currently executing
executor.tasks.succeeded	Counter	Tasks completed without exception
executor.tasks.failed	Counter	Tasks completed with exception
executor.task.duration	Timer	Task wall-clock duration

---

Contributing

Please read CONTRIBUTING.md for details on our code of conduct, and the process for submitting pull requests to us.

Versioning

We use GitHub for versioning. For the versions available, see the tags on this repository.

Authors

• Marcos Vallim - Founder, Author, Development, Test, Documentation - mvallim

See also the list of contributors who participated in this project.

License

This project is licensed under the Apache License - see the LICENSE file for details