AWS - MQ - SQS

Posted Jul 18, 2020

By Grace L

22 min read

SQS - Amazon Simple Queue Service

SQS - Amazon Simple Queue Service

a fully managed message queuing service
- web service
- gives access to message queues that store messages waiting to be processed.
- uses a message-oriented API.
- uses pull based (polling) not push based .
used for distributed/decoupled applications.
- simple and quite cost-effective
- decouple and scale the components of a specific application, microservices, distributed systems, and serverless applications
  .
  - for an architectural where loosely coupled components are needed.
  - using asynchronous messaging queues to avoid single points of failure
  - Loosely Coupled System = Important Design Principle
  - SQS moves the data between distributed application components and also helps to decouple the components.
reliable, highly stable hosted queue for storing messages between applications and/or microservices .
- eliminates the complexity and overhead associated with managing and operating message oriented middleware, and empowers developers to focus on differentiating work.
Using SQS to send, store, and receive messages between software components at any volume, without losing messages or requiring other services to be available.
- Get started with SQS by AWS console, CLI or SDK, and three simple commands.
SQS support encryption
An SQS resource policy sets permissions to access a queue
can be used with RedShift, DynamoDB, EC2, ECS, RDS, S3 and Lambda .
- An SQS resource policy sets permissions to access a queue
- Lambda
  - Lambda functions can be invoked based on messages arriving in the queue
  - Lambda can provide rapid and agile scaling if computing/processing is needed for a message in a queue.
  - If computing is under the maximum runtime of Lambda (15 mins), then this computing is much quicker than provisioning an Auto Scaling Group.
  - Lambda allow for rapid and agile scaling while processing SQS messages for up to 15 mins?
- CloudWatch
  - CloudWatch supplies a full set of metrics for the SQS service.
  - Paired together, CloudWatch and Auto Scaling groups can scale based on the number of messages in the queue.
SQS offers 2 types of message queues.
- Standard queues offer maximum throughput, best-effort ordering, and at-least-once delivery.
- SQS FIFO queues are designed to guarantee that messages are processed exactly once, in the exact order that they are sent.

Messages

Messages are generated by one component for consumption by another
Any volume of messages at any level of throughput can be transmitted without losing messages or requiring other services to be always available.
- 256KB in size in any format.
- can be kept in the queue from 1 minute to 14 days (default is 4 days)
  .
information necessary to delete a message in an SQS queue
- Receipt handle
- The queue URL
To access the SNS service from inside a VPC, you would need:
- Since SNS is a public service,
- a VPC needs a NAT Gateway to access SNS.
  - The NAT gateway sends the traffic to the internet gateway (IGW) to connect to the internet (the SNS public endpoint).
- a VPC needs an Internet Gateway (IGW) to access SNS public endpoint.
  - An Internet Gateway allows communication between your VPC and the internet (the public endpoint of SNS).
- a VPC needs a VPC endpoint to access SNS.
  - A VPC endpoint enables you to privately connect your VPC to supported AWS services, such as SNS.

queue

Queue names must be unique within a region.
can support multiple readers and writers on the same queue.
the repository for messages awaiting processing.
- wait for messages and act as a buffer between senders and receivers.
key difference between SQS and SNS.
- SQS is pull,
- SNS is push.
queue types:
- Standard
  :
  - offers at-least-once delivery
    - If a message creating system restarts a queue or reprocesses a log, duplicate messages may be generated, sent to SQS and processed.
  - and best-effort-ordering .
  - provide a loose-FIFO capability that attempts to preserve the order of messages.
  - designed to be massively scalable using a highly distributed architecture
  - receiving messages in sent order is not guaranteed.
- FIFO
  : first in first out
  - available in limited regions currently.
  - allows to order elements that aren’t ordered programmatically.
  - can be delivered more than once
  - exactly-once processing
    - duplicates are not introduced.
    - each message is delivered once and remains available until a consumer processes it and deletes it.
  - limited throughput
    - up to 300 sends, receives, or deletes per second.
  - preserve the exact order in which messages are sent and received.
    - don’t have to place sequencing information in your message.
- LIMITS
  - In-flight messages: messages that have been picked up by a consumer but not yet deleted from the queue.
    - Standard queues: limit of 120,000 in-flight messages per queue.
    - FIFO queues: limit of 20,000 in-flight messages per queue.
  - Queue names can be up to 80 characters.
  - Messages are retained for 4 days by default up to 14 days.
  - FIFO queues support up to 3000 messages per second when batching or 300 per second otherwise.
  - The maximum messages size is 256KB.

short and long polling

Amazon SQS provides short polling and long polling to receive messages from a queue.

By default, queues use short polling.

short polling
1 **default**
- the ReceiveMessage request queries only a subset of the servers (based on a weighted random distribution ) to find messages that are available to include in the response.
  - It queries only a subset of the available servers for messages (based on weighted random execution).
- Amazon SQS sends the response right away, even if the query found no messages.
  - Does not wait for messages to appear in the queue.
- the WaitTimeSeconds parameter of a ReceiveMessage request is set to 0 in two ways:
  - The ReceiveMessage call sets WaitTimeSeconds to 0.
  - The ReceiveMessage call doesn’t set WaitTimeSeconds, but the queue attribute ReceiveMessageWaitTimeSeconds is set to 0.
- More requests are used, which implies higher cost.
- Thus, a particular ReceiveMessage request might not return all of your messages. However, if you have fewer than 1,000 messages in your queue, a subsequent request will return your messages. If you keep consuming from your queues, Amazon SQS samples all of its servers, and you receive all of your messages.
- Amazon SQS samples several of its servers (in gray) and returns messages A, C, D, and B from these servers. Message E isn’t returned for this request, but is returned for a subsequent request.

long polling
- the ReceiveMessage request queries all of the servers for messages.
- Benefits:
  - Uses fewer requests to reduces cost.
    - Same charge per million requests as short polling.
  - Eliminates false empty responses by querying all servers.
    - Eliminate empty responses by allowing Amazon SQS to wait
    - SQS sends a response after it collects at least one available message, up to the maximum number of messages specified in the request.
    - SQS sends an empty response only if the polling wait time expires.
    - SQS waits until a message is available in the queue before sending a response.
  - Requests contain at least one of the available messages up to the maximum number of messages specified in the ReceiveMessage action.
  - Shouldn’t be used if your application expects an immediate response to receive message calls.
  - ReceiveMessage WaitTime > 0 (up to 20 seconds).

use case

Work Queues, decouple components of a distributed application that might not process the same amount of work simultaneously.
Buffering Batch Operations,
- can add scalability and reliability to architecture, and smooth out temporary volume spikes without losing messages or increasing latency.
Request Offloading,
- move slow operations off interactive request paths by enqueueing the request.
Fan-out
- combine SQS with SNS to send identical copies of a message to multiple queues in parallel, for simultaneous processing.
Auto Scaling
- use SQS queues to determine the load on an application, and then combine with Auto Scaling to scale the number of EC2 instances out or in, depending on the traffic volume.

Loose coupling with SQS

a queuing chain pattern that enables 异步的 asynchronous processing .
- Asynchronous processing enables various processes to run at the same time.
An example of image processing is the sequential operations of uploading, storing, and encoding the image, creating a thumbnail, and copyrighting are tightly linked to each other.
This tight linkage complicates the recovery operations when a failure occurs.
In this diagram, EC2 automatic scaling group that completes six processes.
- To begin, it puts a message in the first queue.
- The message stays in the queue until the second group executes the get message, which pulls it out and processes the job.
- When the job is done, a put message is sent to the next queue, and the processed message is deleted.
- A get message is then sent to the third set of EC2 servers.
You can achieve loose coupling of systems by using queues between systems and exchanging messages that transfer jobs.
enables asynchronous linking of systems. This method lets you increase the number of virtual servers that receive and process the messages in parallel.
If there is no image to process, you can configure Auto Scaling to terminate the excess servers.

While you can use this pattern without cloud technology, the queue itself is provided as an AWS Cloud service called Amazon SQS, which makes using this pattern easier.

Amazon SQS offers free tier.
New and existing customers can receive one million queuing requests for free each month.
Some applications may be able to operate within this free tier limit.

Some benefits are to:

Use asynchronous processing to return responses quickly.
Structure the system through loose coupling of EC2 instances.
Handle performance and service requirements by merely increasing or decreasing the number of EC2 instances used in job processing.
If an EC2 instance fails, a message remains in the queue service, which enables processing to continue immediately when the EC2 instance recovers.
- produces a system that is robust to failure.
Queuing chain pattern is one of the cloud design patterns. Select the link to learn more. https://en.clouddesignpattern.org/index.php/CDP:Queuing_Chain_Pattern
Scalable
- potentially millions of messages.
reliable service
- stores all messages redundantly on multiple servers in multiple data centers.
simultaneously read and write.
secure and requires API credentials.

short polling behavior

This diagram shows the short polling behavior of a message being returned after a system component makes a receive request.
Amazon SQS samples several servers, shown in gray, and returns the messages from the servers—messages A, C, D, and B, in this example. Message E is not returned for this particular request, but it would be returned for a subsequent request.
Message Order
- SQS makes a best effort to preserve the order of messages, but because of the distributed nature of the queue, might not receive messages in the exact order you sent them.
  - SQS uses multiple hosts, and each host holds only a portion of all the messages.
  - When a staff member calls for their next message, the consumer process does not see all the hosts or all the messages.
  - As such, messages are not necessarily delivered in the order in which they were generated.
- If your system requires the order to be preserved, should place sequencing information in each message, to reorder the messages on receipt.
At-Least-Once Delivery
- SQS stores copies of your message on multiple servers for redundancy and high availability.
- On rare occasions, one of the servers storing a copy of a message might be unavailable when you receive or delete a message.
- If that occurs, the copy of the message will not be deleted on the unavailable server, and you might get that message copy again when you receive messages.
- Because of this, you must design your application to be 幂等的 idempotent
- it must not be adversely affected if it processes the same message more than once.
Message Sample
- How messages are retrieved from the queue depends on whether you are using short / standard polling (the default behavior), or long polling.
- When retrieve messages from the queue with short polling, SQS samples a subset of the servers, based on a weighted, random distribution. Then, it returns messages from just those servers.
- This means that a particular receive request might not return all your messages. Or, if you have a small number of messages in your queue, such as less than 1,000, a particular request might not return any of your messages, whereas a subsequent request will.
- If you keep retrieving from your queues, SQS will sample all the servers, and you will receive all your messages.

Visibility timeout

the period of time in which a message is invisible to the rest of the application after an application component/reader gets the message from the queue.
- When a consumer instance retrieves a message, that message will be hidden from other consumer instances for a fixed period.
When a message is received, it is locked while being processed.
- The component that receives the message processes it and then deletes it from the queue.
- If the message processing fails, the lock expires, and the message is available again.
prevents it from being processed by other components.
- prevents multiple components from processing the same message.
protects the message to ensure it is always processed.
- if an instance fails and the lock never times out, the message stays invisible.
- The timeout enables the message to become unlocked, and visible, so it can be processed by another instance.
- When the application needs more time for processing, the “invisible” timeout can be changed dynamically via the ChangeMessageVisibility operation.
The maximum visibility timeout for an SQS message is 12 hours.
An Amazon SQS message can contain up to 10 metadata attributes.
DelaySeconds
- When a new message is added to the SQS queue, it will be hidden from consumer instances for a fixed period.
WaitTimeSeconds / ReceiveMessageWaitTime
- When the consumer instance polls for new work, the SQS service will allow it to wait a certain time for one or more messages to be available before closing the connection.
- waits until a message is available in the queue before sending a response.
- Short polling/Long polling

dead letter queue, DLQ

a queue of messages that could not be processed.
provides the ability to sideline and isolate messages that cannot be processed for analysis out of band.
A DLQ receives messages after a maximum number of processing attempts is reached.
A DLQ restores message flow when FIFO is enabled and a message cannot be processed.
A dead letter queue is like any other SQS queue.
- Messages can be sent to it and received from it.
can create a DLQ from the SQS API and the SQS console.

Queues can be shared with other AWS accounts and anonymously.
- The owner of the queue pays for shared queue access.
- A permission gives access to another person to use your queue in a particular way.
A developer associates an access policy statement , specifies the permissions being granted, with the queue to be shared.
- A policy is the actual document that contains the permissions you granted.
the messages cannot be shared between queues in different Regions.
- SQS in each Region is independent in message stores and queue names;
SQS provides APIs to create and manage the access policy statements, which are AddPermission, RemovePermission, SetQueueAttributes, and GetQueueAttributes.

Attributes

Amazon SQS message attributes

message body
- contains message payload
message_attributes
- message_headers (can be used to apply different routing and filtering message using their headers information)
- used as message metadata and not the message itself.
- include structured metadata (such as timestamps, geospatial data, category, signatures, and identifiers) with messages using message attributes.
- consumer can use message attributes to handle a message in a particular way without having to process the message body first.

Message attribute components

All components of a message attribute are included in the 256 KB message size restriction. The Name, Type, Value, and the message body must not be empty or null.

Each message attribute consists of the following components:

Name:
- can contain the following characters: A-Z, a-z, 0-9, underscore (_), hyphen (-), and period (.).
- Can be up to 256 characters long
- Can’t start with AWS. or Amazon. (or any casing variations)
- Is case-sensitive
- Must be unique among all attribute names for the message
- Must not start or end with a period
- Must not have periods in a sequence
TypeL: The message attribute data type.
- Supported types include String, Number, and Binary.
- can also add custom information for any data type.
- The data type has the same restrictions as the message body
- Can be up to 256 characters long
- Is case-sensitive
ValueL: The message attribute value.
- For String data types, the attribute values has the same restrictions as the message body.

Message attribute data types

Message attribute data types instruct Amazon SQS how to handle the corresponding message attribute values. For example, if the type is Number, Amazon SQS validates numerical values.
Amazon SQS supports the logical data types String, Number, and Binary with optional custom data type labels with the format .custom-data-type
- String: store Unicode text using any valid XML characters.
- Number: store positive or negative numerical values.
  - can have up to 38 digits of precision, and it can be between 10^-128 and 10^+126.
  - Amazon SQS removes leading and trailing zeroes.
- Binary:
  - store any binary data such as compressed data, encrypted data, or images.
- Custom
  - create a custom data type, append a custom-type label to any data type.
  - For example:
  - Number.byte, Number.short, Number.int, and Number.float can help distinguish between number types.
  - Binary.gif and Binary.png can help distinguish between file types.
  - Amazon SQS doesn’t interpret, validate, or use the appended data.
  - The custom-type label has the same restrictions as the message body.

Defining attributes

  
// String
final Map<String, MessageAttributeValue> messageAttributes = new HashMap<>();
messageAttributes.put(
  "Name",
  new MessageAttributeValue().withDataType("String").withStringValue("Jane"));

// Number
final Map<String, MessageAttributeValue> messageAttributes = new HashMap<>();
messageAttributes.put(
  "AccurateWeight",
  new MessageAttributeValue().withDataType("Number").withStringValue("230.001"));

// Binary
final Map<String, MessageAttributeValue> messageAttributes = new HashMap<>();
messageAttributes.put(
  "ByteArray",
  new MessageAttributeValue().withDataType("Binary").withBinaryValue(ByteBuffer.wrap(new byte[10])));

// String (custom)
final Map<String, MessageAttributeValue> messageAttributes = new HashMap<>();
messageAttributes.put(
  "EmployeeId",
  new MessageAttributeValue().withDataType("String.EmployeeId").withStringValue("ABC123456"));

// Number (custom)
final Map<String, MessageAttributeValue> messageAttributes = new HashMap<>();
messageAttributes.put(
  "AccountId",
  new MessageAttributeValue().withDataType("Number.AccountId").withStringValue("000123456"));
// Because the base data type is Number, the ReceiveMessage method returns 123456.

// Binary (custom)
final Map<String, MessageAttributeValue> messageAttributes = new HashMap<>();
messageAttributes.put(
  "ApplicationIcon",
  new MessageAttributeValue().withDataType("Binary.JPEG").withBinaryValue(ByteBuffer.wrap(new byte[10])));

Sending message with attributes

This example adds the attributes to the SendMessageRequest before sending the message.

  
// Send a message with an attribute.
final SendMessageRequest sendMessageRequest = new SendMessageRequest();

sendMessageRequest.withMessageBody("This is my message text.");
sendMessageRequest.withQueueUrl(myQueueUrl);
sendMessageRequest.withMessageAttributes(messageAttributes);

sqs.sendMessage(sendMessageRequest);

example

  
SqsClient sqsClient = SqsClient.builder().region(Region.of("ap-southeast-2")).build();

CreateQueueRequest request = CreateQueueRequest.builder().queueName("MyQueue").build();
CreateQueueResponse response = sqsClient.createQueue(request);

// generates a UUID as the traceId
String traceId = UUID.randomUUID().toString();

final Map<String, MessageAttributeValue> messageAttributes = new HashMap<>();

// add traceId as a message attribute
messageAttributes.put(
  TRACE_ID_NAME,
  MessageAttributeValue.builder().dataType("String").stringValue(traceId).build());

SendMessageRequest.Builder builder = SendMessageRequest.builder()
        .queueUrl(queueUrl)
        .messageBody(message)
        .messageAttributes(messageAttributes);

// send the message
logger.info("Sending message to queue {} with {}={}", this.queueName, TRACE_ID_NAME, traceId);
this.sqsClient.sendMessage(builder.build());



// Extracting attribute

// long polling and wait for waitTimeSeconds before timed out
ReceiveMessageRequest receiveMessageRequest = ReceiveMessageRequest.builder()
        .queueUrl(queueUrl)
        .waitTimeSeconds(20)
        .messageAttributeNames("trace-id") // returns the trace Id
        .build();
List<Message> messages = this.sqsClient.receiveMessage(receiveMessageRequest).messages();



/**
 * Extract message attribute.
 *
 * @param message       The message
 * @param attributeName The attribute name
 * @return The attribute value
 */
private static String extractAttribute(Message message, String attributeName) {
    if (message.hasMessageAttributes()) {
        Map<String, MessageAttributeValue> messageAttributes = message.messageAttributes();
        MessageAttributeValue attributeValue = messageAttributes.get(attributeName);
        if (attributeValue != null) {
            return attributeValue.stringValue();
        }
    }
    return null;
}

/**
 * Process message.
 *
 * @param message the message
 */
private void processMessage(Message message) {
    logger.info("Processing message {}", message.messageId());

    // extract traceId
    String traceId = extractAttribute(message, TRACE_ID_NAME);

    // special handling before parsing the message body
    if (traceId == null || traceId.length() != 36) {
        logger.error("{} is compromised, message {} abandoned", TRACE_ID_NAME, message.messageId());
        return;
    }

    // deserialise message body
    MyEvent myEvent = null;
    try {
        myEvent = objectMapper.readValue(message.body(), MyEvent.class);
    } catch (JsonProcessingException e) {
        logger.error(e);
    }

    logger.info("Message processed: {}={}, MyEvent={}", TRACE_ID_NAME, traceId, myEvent == null ? null : myEvent.toString());
}



gradlew messageAttributesExample


// The output would be:
aws.demo.MessageAttributesExample:135 [main] - Create queue MyQueue in region ap-southeast-2
aws.demo.MessageAttributesExample:139 [main] - Queue URL: https://sqs.ap-southeast-2.amazonaws.com/000000000000/MyQueue

aws.demo.MessageAttributesExample$MyProducer:276 [pool-3-thread-1] - Sending message to queue MyQueue with trace-id=abc

aws.demo.MessageAttributesExample$MyConsumer:334 [pool-4-thread-1] - Receiving messages from MyQueue...
aws.demo.MessageAttributesExample$MyConsumer:345 [pool-4-thread-1] - 1 messages received.
aws.demo.MessageAttributesExample$MyConsumer:360 [pool-4-thread-1] - Processing message xxx
aws.demo.MessageAttributesExample$MyConsumer:379 [pool-4-thread-1] - Message processed: trace-id=abc, MyEvent=MyEvent[id='434d8016-b06c-4424-8ab1-1140f6fc87b3', timeStamp=2020-08-25T13:11:24.101739200Z, source='pool-3-thread-1', payload='AWS SQS message attributes example.']
aws.demo.MessageAttributesExample$MyConsumer:390 [pool-4-thread-1] - Deleting message 9cfcde98-841d-415e-b795-7c9396c7a8ea from queue: MyQueue

aws.demo.MessageAttributesExample:149 [main] - Delete queue MyQueue
aws.demo.MessageAttributesExample:154 [main] - Queue MyQueue deleted.

send a message to a First-In-First-Out (FIFO) queue, make sure that the sendMessage method executes after provide the message group ID.
If use the SendMessageBatch method instead of SendMessage, you must specify message attributes for each message in the batch.

  
aws sqs send-message \
  --queue-url https://sqs.us-east-1.amazonaws.com/80398EXAMPLE/MyQueue \
  --message-body "Information about the largest city in Any Region." \
  --delay-seconds 10 \
  --message-attributes file://send-message.json
# send-message.json:
{
  "City": {
    "DataType": "String",
    "StringValue": "Any City"
  },
  "Greeting": {
    "DataType": "Binary",
    "BinaryValue": "Hello, World!"
  },
  "Population": {
    "DataType": "Number",
    "StringValue": "1250800"
  }
}


aws sqs send-message-batch \
  --queue-url https://sqs.us-east-1.amazonaws.com/80398EXAMPLE/MyQueue \
  --entries file://send-message-batch.json
# send-message-batch.json :
[
  {
    "Id": "FuelReport-0001-2015-09-16T140731Z",
    "MessageBody": "Fuel report for account 0001 on 2015-09-16 at 02:07:31 PM.",
    "DelaySeconds": 10,
    "MessageAttributes": {
      "SellerName": {
            "DataType": "String",
                "StringValue": "Example Store"
      },
      "City": {
        "DataType": "String",
        "StringValue": "Any City"
      }
    }
  },
  {
    "Id": "FuelReport-0002-2015-09-16T140930Z",
    "MessageBody": "Fuel report for account 0002 on 2015-09-16 at 02:09:30 PM.",
    "DelaySeconds": 10,
    "MessageAttributes": {
      "SellerName": {
            "DataType": "String",
                "StringValue": "Example Fuels"
      },
      "City": {
        "DataType": "String",
        "StringValue": "North Town"
      }
    }
  }
]

Amazon SQS message system attributes

Whereas you can use message attributes to attach custom metadata to SQS messages for applications,

you can use message system attributes to store metadata for other AWS services, such as AWS X-Ray.

For more information, see

the MessageSystemAttribute request parameter of the SendMessage and SendMessageBatch API actions,
the AWSTraceHeader attribute of the ReceiveMessage API action,
the MessageSystemAttributeValue data type in the Amazon Simple Queue Service API Reference.

Message system attributes are structured exactly like message attributes, with the following exceptions:

Currently, the only supported message system attribute is AWSTraceHeader. Its type must be String and its value must be a correctly formatted AWS X-Ray trace header string.
The size of a message system attribute doesn’t count towards the total size of a message.

SCALABILITY AND DURABILITY

can have multiple queues with different priorities.
Scaling is performed by creating more queues.
SQS stores all message queues and messages within a single, highly-available AWS region with multiple redundant AZs.

SECURITY

can use IAM policies to control who can read/write messages.
- Authentication can be used to secure messages within queues
- (who can send and receive).
SQS supports HTTPS and supports TLS versions 1.0, 1.1, 1.2.
SQS is PCI DSS level 1 compliant and HIPAA eligible.
Server-side encryption (SSE) lets you transmit sensitive data in encrypted queues (AWS KMS):
- SSE encrypts messages as soon as SQS receives them.
- The messages are stored in encrypted form and SQS decrypts messages only when they are sent to an authorized consumer.
- Uses AES 256 bit encryption.
- Not available in all regions.
- Standard and FIFO queues.
- Body of message is encrypted.
- The following is not encrypted:
  - Queue metadata.
  - Message metadata.
  - Per-queue metrics.

Monitoring

CloudWatch is integrated with SQS
- can view and monitor queue metrics.
- metrics are automatically collected every 5 minutes.
CloudWatch considers a queue to be active for up to 6 hours if it contains any messages or if any API action accesses it.
No charge for CloudWatch (no detailed monitoring).
CloudTrail captures API calls from SQS and logs to a specified S3 bucket.

01AWS, MQ

AWS

This post is licensed under CC BY 4.0 by the author.