SQS’s "exactly-once processing" is a bit of a misnomer; it’s really about idempotent consumers that can tolerate duplicate messages without causing issues.
Let’s see SQS in action. Imagine a simple order processing system.
import boto3
import json
import time
sqs = boto3.client('sqs', region_name='us-east-1')
sns = boto3.client('sns', region_name='us-east-1')
QUEUE_URL = 'YOUR_SQS_QUEUE_URL' # Replace with your actual SQS queue URL
TOPIC_ARN = 'YOUR_SNS_TOPIC_ARN' # Replace with your actual SNS topic ARN
def send_order_message(order_id, item, quantity):
message_body = {
'order_id': order_id,
'item': item,
'quantity': quantity,
'timestamp': int(time.time())
}
response = sqs.send_message(
QueueUrl=QUEUE_URL,
MessageBody=json.dumps(message_body),
MessageAttributes={
'MessageType': {
'DataType': 'String',
'StringValue': 'Order'
}
}
)
print(f"Sent order {order_id} message: {response['MessageId']}")
def process_order_message(message):
receipt_handle = message['ReceiptHandle']
body = json.loads(message['Body'])
order_id = body['order_id']
item = body['item']
quantity = body['quantity']
timestamp = body['timestamp']
print(f"Processing order {order_id} for {quantity} of {item} (sent at {timestamp})...")
# Simulate order processing logic
# In a real-world scenario, this might involve:
# - Updating a database
# - Calling another service
# - Sending a notification via SNS
# Example: Simulate a successful database update
time.sleep(1)
print(f"Order {order_id} processed successfully.")
# Delete the message from SQS after successful processing
sqs.delete_message(
QueueUrl=QUEUE_URL,
ReceiptHandle=receipt_handle
)
print(f"Deleted message for order {order_id}.")
def receive_messages():
response = sqs.receive_message(
QueueUrl=QUEUE_URL,
MaxNumberOfMessages=10,
WaitTimeSeconds=20, # Long polling
MessageAttributeNames=['MessageType']
)
messages = response.get('Messages', [])
for message in messages:
if message['MessageAttributes']['MessageType']['StringValue'] == 'Order':
process_order_message(message)
else:
print(f"Received unknown message type: {message['MessageAttributes']['MessageType']['StringValue']}")
# Optionally delete or move to a dead-letter queue
if __name__ == "__main__":
# Example usage: Send some messages
send_order_message('ORD1001', 'Laptop', 1)
send_order_message('ORD1002', 'Keyboard', 2)
send_order_message('ORD1001', 'Laptop', 1) # Duplicate message for ORD1001
print("\n--- Starting message consumption ---")
# In a real application, this would run continuously
for _ in range(5): # Simulate processing for a few cycles
receive_messages()
time.sleep(5)
This code demonstrates sending messages to SQS and then consuming them. The crucial part is process_order_message and how it interacts with SQS. When a message is received, the consumer processes it and then deletes it from the queue using its ReceiptHandle. If the consumer crashes after processing but before deleting, SQS will eventually make the message visible again. This is where duplicates can happen.
The mental model for SQS processing, particularly with "exactly-once" in mind, revolves around these key components:
- Message Producer: Your application or service sends messages to an SQS queue. Each message has a body and optional attributes.
- SQS Queue: The central hub. It stores messages. SQS guarantees at-least-once delivery by default. When a consumer reads a message, it becomes invisible for a configurable "visibility timeout." If the consumer doesn’t delete the message within this timeout, it reappears in the queue.
- Message Consumer: Your application reads messages from the queue. It processes the message’s content.
- Visibility Timeout: This is the duration a message remains invisible to other consumers after being received. It’s your primary defense against processing duplicates during consumption.
- Message Deletion: Crucially, a consumer must explicitly delete a message from the queue after it has been successfully processed. This is the signal to SQS that the message is handled.
- Idempotency: This is the consumer’s responsibility. An idempotent operation is one that can be performed multiple times with the same effect as performing it once. For example, setting a user’s status to "active" is idempotent. If you do it once or ten times, the status is still "active."
The "exactly-once" promise is achieved by making your consumer idempotent. SQS guarantees that a message will be delivered at least once. If your consumer can safely process the same message multiple times without adverse side effects, then you’ve effectively achieved "exactly-once" for your application’s business logic.
The core challenge is handling the scenario where a message is delivered, processed, but not deleted before the visibility timeout expires, leading to a redelivery.
The way to build an idempotent consumer is to ensure that the effect of processing a message is unique and repeatable without side effects. A common pattern is to use a unique identifier from the message itself (like order_id in the example) and store the state of its processing.
Consider processing an Order message. The order_id is the key. You want to ensure that an order is only created once.
A robust idempotent consumer would involve:
- A unique identifier: Extract a unique identifier from the message (e.g.,
order_id,transaction_id). - A state store: Use a database (like DynamoDB, RDS, or even Redis) to track which unique identifiers have already been processed.
- Atomic check-and-set: Before processing, check if the identifier exists in your state store. If it does, the message is a duplicate; skip processing and delete it. If it doesn’t, atomically add the identifier to the state store and then proceed with the actual business logic. The "atomically" part is key; you don’t want to add the ID and then crash before processing, or process and then crash before adding the ID.
Here’s a conceptual look at how an idempotent consumer might check a database before processing:
# Conceptual idempotent processing logic
def process_idempotent_order(message):
receipt_handle = message['ReceiptHandle']
body = json.loads(message['Body'])
order_id = body['order_id']
# Assume 'processed_orders_db' is a database table/collection
# that stores 'order_id' as a primary key.
# Check if order_id already exists in our processed records
existing_order = processed_orders_db.get_item(Key={'order_id': order_id})
if existing_order['Item']:
print(f"Duplicate detected: Order {order_id} already processed. Deleting message.")
sqs.delete_message(QueueUrl=QUEUE_URL, ReceiptHandle=receipt_handle)
return
# If not found, proceed with processing and mark as processed
try:
print(f"Processing new order: {order_id}...")
# --- Actual business logic here ---
# e.g., create order in your main database
create_order_in_main_db(order_id, body)
# ---------------------------------
# Mark as processed in our state store *after* successful business logic
processed_orders_db.put_item(Item={'order_id': order_id, 'processed_at': int(time.time())})
print(f"Order {order_id} processed and marked. Deleting message.")
sqs.delete_message(QueueUrl=QUEUE_URL, ReceiptHandle=receipt_handle)
except Exception as e:
print(f"Error processing order {order_id}: {e}. Message will be redelivered.")
# Do NOT delete the message. It will become visible again after visibility timeout.
# Handle error, e.g., send to DLQ if retries exhausted.
The most surprising truth about SQS exactly-once processing is that the "exactly-once" guarantee is entirely on the consumer’s shoulders, not SQS’s. SQS provides the foundation by guaranteeing "at-least-once" delivery and offering mechanisms like visibility timeouts and dead-letter queues, but the actual prevention of side effects from duplicate deliveries is a design pattern you implement in your consuming application.
The next hurdle you’ll face is managing message ordering when it matters for your business logic.