Amazon SQS #

Amazon Simple Queue Service (SQS) is AWS’s fully-managed message queue service that requires no installation or infrastructure management whatsoever. Unlike RabbitMQ or Kafka, which must be set up and maintained yourself, SQS just needs to be created via the AWS console or Terraform and it’s ready to use — no broker servers, no connection pools, no patching. SQS is a great fit for applications already running in the AWS ecosystem that want to decouple components in the simplest way possible. In Ruby, the official AWS SDK (aws-sdk-sqs) provides a complete client, and Shoryuken provides a Sidekiq-like worker framework for SQS.

Standard Queue vs FIFO Queue #

SQS has two queue types with different characteristics:

Standard Queue:
  ✓ Nearly unlimited throughput (unlimited TPS)
  ✓ At-least-once delivery — messages are guaranteed to be sent, but can be duplicated
  ✗ Best-effort ordering — order is not guaranteed
  ✓ Cheaper pricing
  Suitable for: tasks tolerant of duplicates, high-volume processing

FIFO Queue (First-In-First-Out):
  ✓ Exactly-once processing — no duplicates
  ✓ Ordering is guaranteed — messages are processed in the order they arrive
  ✗ Limited throughput: 300 TPS (or 3000 with batching)
  ✗ More expensive
  Suitable for: order processing, financial transactions, event sourcing
  FIFO queue names must end with ".fifo": "orders.fifo"

flowchart LR
    P[Producer\nRuby App] -->|send_message| SQS

    subgraph AWS SQS
        SQS[Queue: new-orders]
        DLQ[DLQ: new-orders-dlq]
        SQS -->|max_receive_count = 3| DLQ
    end

    SQS -->|receive_message| C1[Consumer 1\nEC2 / Lambda]
    SQS -->|receive_message| C2[Consumer 2\nEC2 / Lambda]
    SQS -->|receive_message| C3[Consumer 3\nEC2 / Lambda]
    DLQ --> Alert[CloudWatch Alarm\n→ Notification]

Installation and Configuration #

gem install aws-sdk-sqs

# Or in the Gemfile

# Gemfile
gem 'aws-sdk-sqs',  '~> 1.70'   # SQS client
gem 'shoryuken',    '~> 6.2'    # worker framework (optional)

AWS Authentication #

There are several ways to authenticate to AWS — choose the one that best fits your environment:

require 'aws-sdk-sqs'

# Method 1: Environment variables (most common for development)
# AWS_ACCESS_KEY_ID=AKIAIOSFODNN7EXAMPLE
# AWS_SECRET_ACCESS_KEY=wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
# AWS_REGION=ap-southeast-1
sqs = Aws::SQS::Client.new   # automatically reads from ENV

# Method 2: Explicit in code (avoid in production)
sqs = Aws::SQS::Client.new(
  region:            "ap-southeast-1",
  access_key_id:     ENV["AWS_ACCESS_KEY_ID"],
  secret_access_key: ENV["AWS_SECRET_ACCESS_KEY"]
)

# Method 3: IAM Role (best for EC2, ECS, Lambda)
# No credentials needed at all — the AWS SDK automatically grabs them from instance metadata
sqs = Aws::SQS::Client.new(region: "ap-southeast-1")

# Method 4: AWS Profile from ~/.aws/credentials
sqs = Aws::SQS::Client.new(
  region:  "ap-southeast-1",
  profile: "production"   # profile name in ~/.aws/credentials
)

# Method 5: LocalStack for local development (no connection to real AWS)
sqs = Aws::SQS::Client.new(
  region:            "us-east-1",
  endpoint:          "http://localhost:4566",
  access_key_id:     "test",
  secret_access_key: "test"
)

Queue Management #

# Create a Standard Queue
resp = sqs.create_queue(
  queue_name:  "new-orders",
  attributes: {
    "VisibilityTimeout"             => "30",       # seconds
    "MessageRetentionPeriod"        => "86400",    # 1 day
    "ReceiveMessageWaitTimeSeconds" => "20",       # long polling
    "RedrivePolicy" => JSON.generate({
      deadLetterTargetArn: "arn:aws:sqs:ap-southeast-1:123456789:new-orders-dlq",
      maxReceiveCount:     3                       # after 3 failures → DLQ
    })
  }
)
queue_url = resp.queue_url

# Create a FIFO Queue (the name must end in .fifo)
sqs.create_queue(
  queue_name: "payments.fifo",
  attributes: {
    "FifoQueue"                     => "true",
    "ContentBasedDeduplication"     => "true",     # automatic deduplication based on content
    "VisibilityTimeout"             => "60",
    "MessageRetentionPeriod"        => "86400",
    "ReceiveMessageWaitTimeSeconds" => "20"
  }
)

# Get the URL of an existing queue
resp = sqs.get_queue_url(queue_name: "new-orders")
queue_url = resp.queue_url
# => "https://sqs.ap-southeast-1.amazonaws.com/123456789/new-orders"

# Get queue attributes (including message counts)
attrs = sqs.get_queue_attributes(
  queue_url:       queue_url,
  attribute_names: ["All"]
).attributes

puts "Messages waiting:    #{attrs['ApproximateNumberOfMessages']}"
puts "Messages not visible: #{attrs['ApproximateNumberOfMessagesNotVisible']}"
puts "Messages delayed:    #{attrs['ApproximateNumberOfMessagesDelayed']}"

# Delete a queue (be careful!)
sqs.delete_queue(queue_url: queue_url)

# List all queues
sqs.list_queues(queue_name_prefix: "orders").queue_urls.each { |url| puts url }

Sending Messages #

queue_url = sqs.get_queue_url(queue_name: "new-orders").queue_url

# Send a simple message
sqs.send_message(
  queue_url:    queue_url,
  message_body: JSON.generate({
    event:     "order_created",
    order_id:  12345,
    total:     150_000,
    timestamp: Time.now.iso8601
  })
)

# Send a message with additional options
resp = sqs.send_message(
  queue_url:     queue_url,
  message_body:  JSON.generate({ order_id: 12346, total: 250_000 }),
  delay_seconds: 5,          # delay delivery by 5 seconds
  message_attributes: {
    "event_type" => {
      data_type:    "String",
      string_value: "new_order"
    },
    "priority" => {
      data_type:    "Number",
      string_value: "10"
    },
    "schema_version" => {
      data_type:    "String",
      string_value: "v2"
    }
  }
)

puts "Message ID: #{resp.message_id}"

# For FIFO Queues — MessageGroupId and MessageDeduplicationId are required
sqs.send_message(
  queue_url:              "https://sqs.ap-southeast-1.amazonaws.com/123/payments.fifo",
  message_body:           JSON.generate({ payment_id: 9999, amount: 500_000 }),
  message_group_id:       "user-#{user_id}",   # all messages in this group are processed in order
  message_deduplication_id: SecureRandom.uuid  # unique ID to prevent duplicates
)

Batch Send — Sending Many at Once #

# Batch send — max 10 messages per batch, saves cost and latency
entries = order_list.each_with_index.map do |order, i|
  {
    id:           "msg-#{i}",   # unique ID within the batch (not the SQS Message ID)
    message_body: JSON.generate({
      order_id: order.id,
      total:    order.total,
      timestamp: Time.now.iso8601
    })
  }
end

# Send in batches of 10
entries.each_slice(10) do |batch|
  resp = sqs.send_message_batch(
    queue_url: queue_url,
    entries:   batch
  )

  # Check for failed messages
  resp.failed.each do |failure|
    puts "Failed to send #{failure.id}: #{failure.message} (#{failure.code})"
  end

  puts "Successfully sent #{resp.successful.length} messages"
end

Receiving and Processing Messages #

Long Polling — The Efficient Way #

# Long polling — wait up to 20 seconds for messages to arrive
# More cost-efficient than short polling (no constant polling)
loop do
  resp = sqs.receive_message(
    queue_url:              queue_url,
    max_number_of_messages: 10,           # max 10 messages per receive
    wait_time_seconds:      20,           # long polling: wait max 20 seconds
    visibility_timeout:     30,           # hide messages from other consumers for 30 seconds
    message_attribute_names: ["All"],     # fetch all message attributes
    attribute_names:         ["All"]      # fetch all system attributes
  )

  resp.messages.each do |message|
    puts "Message ID: #{message.message_id}"
    puts "Body: #{message.body}"
    puts "Attributes: #{message.message_attributes}"
    puts "Received #{message.attributes['ApproximateReceiveCount']} times"

    begin
      data = JSON.parse(message.body, symbolize_names: true)
      process_order(data)

      # Delete the message after successful processing
      sqs.delete_message(
        queue_url:      queue_url,
        receipt_handle: message.receipt_handle
      )
      puts "Message #{message.message_id} processed and deleted successfully"

    rescue JSON::ParserError => e
      puts "Invalid JSON: #{e.message}"
      # Delete invalid messages — don't leave them in the queue
      sqs.delete_message(queue_url: queue_url, receipt_handle: message.receipt_handle)

    rescue => e
      puts "Error: #{e.message}"
      # Don't delete — the visibility timeout will expire and the message returns to the queue
      # After maxReceiveCount times → goes to the DLQ
    end
  end
end

Dynamically Changing the Visibility Timeout #

# If processing takes longer than the visibility timeout,
# extend it before it expires
message = sqs.receive_message(
  queue_url:          queue_url,
  visibility_timeout: 30
).messages.first

Thread.new do
  loop do
    sleep 20   # extend every 20 seconds (before the 30 seconds expire)
    sqs.change_message_visibility(
      queue_url:          queue_url,
      receipt_handle:     message.receipt_handle,
      visibility_timeout: 30   # add another 30 seconds
    )
  end
end

# A processing task that takes a long time
heavy_process(JSON.parse(message.body))
sqs.delete_message(queue_url: queue_url, receipt_handle: message.receipt_handle)

Batch Delete — Deleting Many at Once #

# Batch delete — more efficient than deleting one by one
message_list = sqs.receive_message(
  queue_url:              queue_url,
  max_number_of_messages: 10,
  wait_time_seconds:      20
).messages

# Process all messages
processed = []
message_list.each do |message|
  begin
    process_order(JSON.parse(message.body))
    processed << { id: message.message_id, receipt_handle: message.receipt_handle }
  rescue => e
    puts "Failed to process #{message.message_id}: #{e.message}"
    # Don't add to the delete list
  end
end

# Delete all successful ones in a single batch
unless processed.empty?
  resp = sqs.delete_message_batch(
    queue_url: queue_url,
    entries:   processed
  )

  resp.failed.each do |f|
    puts "Failed to delete #{f.id}: #{f.message}"
  end
end

Dead Letter Queues (DLQ) #

A DLQ holds messages that failed to process after maxReceiveCount times:

# 1. Create the DLQ first
dlq_resp = sqs.create_queue(queue_name: "new-orders-dlq")
dlq_url  = dlq_resp.queue_url

# Get the DLQ's ARN
dlq_arn = sqs.get_queue_attributes(
  queue_url:       dlq_url,
  attribute_names: ["QueueArn"]
).attributes["QueueArn"]

# 2. Create the main queue with DLQ configuration
sqs.create_queue(
  queue_name: "new-orders",
  attributes: {
    "RedrivePolicy" => JSON.generate({
      deadLetterTargetArn: dlq_arn,
      maxReceiveCount:     3   # after 3 receives without deletion → DLQ
    })
  }
)

# 3. Monitor and process messages in the DLQ
def monitor_dlq(sqs, dlq_url)
  loop do
    resp = sqs.receive_message(
      queue_url:              dlq_url,
      max_number_of_messages: 10,
      wait_time_seconds:      20,
      attribute_names:        ["All"]
    )

    resp.messages.each do |message|
      receive_count = message.attributes["ApproximateReceiveCount"]
      first_sent    = message.attributes["SentTimestamp"].to_i / 1000

      puts "DLQ: #{message.message_id} (received #{receive_count}x)"
      puts "Sent at: #{Time.at(first_sent)}"
      puts "Body: #{message.body}"

      # Save to a database for analysis
      FailedOrder.create!(
        message_id:   message.message_id,
        payload:      message.body,
        receive_count: receive_count.to_i,
        failed_at:    Time.now
      )

      # Notify the team
      Monitoring.alert("Message in DLQ: #{message.message_id}")

      # Delete from the DLQ after processing
      sqs.delete_message(queue_url: dlq_url, receipt_handle: message.receipt_handle)
    end

    sleep 60   # check the DLQ every 1 minute
  end
end

The SNS → SQS pattern is very common on AWS — one publish to an SNS topic is forwarded to many SQS queues:

require 'aws-sdk-sns'
require 'aws-sdk-sqs'

sns = Aws::SNS::Client.new(region: "ap-southeast-1")
sqs = Aws::SQS::Client.new(region: "ap-southeast-1")

# 1. Create an SNS Topic
topic_arn = sns.create_topic(name: "order_events").topic_arn

# 2. Create several SQS Queues
["order_notifications", "order_analytics", "order_inventory"].each do |name|
  url = sqs.create_queue(queue_name: name).queue_url
  arn = sqs.get_queue_attributes(
    queue_url: url, attribute_names: ["QueueArn"]
  ).attributes["QueueArn"]

  # 3. Subscribe the queue to the SNS topic
  sns.subscribe(
    topic_arn: topic_arn,
    protocol:  "sqs",
    endpoint:  arn
  )

  # 4. Set the queue policy — allow SNS to send messages to this queue
  sqs.set_queue_attributes(
    queue_url:  url,
    attributes: {
      "Policy" => JSON.generate({
        Version: "2012-10-17",
        Statement: [{
          Effect:    "Allow",
          Principal: { Service: "sns.amazonaws.com" },
          Action:    "SQS:SendMessage",
          Resource:  arn,
          Condition: { ArnEquals: { "aws:SourceArn" => topic_arn } }
        }]
      })
    }
  )
end

# 5. Publish to SNS — automatically forwarded to all subscribed queues
sns.publish(
  topic_arn: topic_arn,
  message:   JSON.generate({
    event:     "order_created",
    order_id:  12345,
    total:     150_000
  }),
  subject:   "order_created",
  message_attributes: {
    "event_type" => { data_type: "String", string_value: "order_created" }
  }
)

Shoryuken — Background Jobs for Rails #

Shoryuken is an SQS worker framework integrated with Rails ActiveJob:

# Gemfile
# gem 'shoryuken', '~> 6.2'

# config/shoryuken.yml
aws:
  region: ap-southeast-1
  # Credentials from an IAM Role or environment variables

concurrency: 25   # number of concurrent worker threads
delay: 0          # default delay between polls

queues:
  - [new_orders,   6]    # high priority (weight 6)
  - [payments,     4]    # medium priority
  - [notifications, 2]   # low priority
  - [reports,      1]

polling_strategy: WeightedRoundRobin   # distribution based on weight

# app/workers/order_shoryuken_worker.rb
class OrderShoryukenWorker
  include Shoryuken::Worker

  shoryuken_options(
    queue:       "new_orders",
    auto_delete: true,        # auto-delete after success
    body_parser: :json        # parse the body as JSON automatically
  )

  def perform(sqs_msg, body)
    order_id = body["order_id"]
    Rails.logger.info "Processing order #{order_id}"

    # body is already parsed as a Hash (because body_parser: :json)
    order = Order.find(order_id)
    order.update!(status: :processed)
    NotificationMailer.order_confirmed(order).deliver_later

  rescue ActiveRecord::RecordNotFound => e
    Rails.logger.error "Order not found: #{e.message}"
    # auto_delete: true → the message is still deleted even with an error
    # For retries: raise an error so the message isn't deleted and returns to the queue
  end
end

# Enqueue a message to SQS
OrderShoryukenWorker.perform_async(
  JSON.generate({ order_id: 123, total: 150_000 })
)

# Or with a delay
OrderShoryukenWorker.perform_in(5.minutes, JSON.generate({ order_id: 456 }))

# ActiveJob integration (without a Shoryuken worker class)
# config/application.rb
config.active_job.queue_adapter = :shoryuken

# app/jobs/send_notification_job.rb
class SendNotificationJob < ApplicationJob
  queue_as :notifications

  def perform(user_id, message)
    user = User.find(user_id)
    NotificationService.send(user, message)
  end
end

# Enqueue
SendNotificationJob.perform_later(user.id, "Order confirmed!")
SendNotificationJob.set(wait: 5.minutes).perform_later(user.id, "Reminder")

# Run the Shoryuken worker
bundle exec shoryuken -r ./config/environment.rb -C config/shoryuken.yml

# For Rails
bundle exec shoryuken -r ./config/environment.rb -C config/shoryuken.yml -R

At-Least-Once vs Exactly-Once Patterns #

SQS Standard doesn’t guarantee exactly-once — messages can be received more than once:

# AT-LEAST-ONCE — Standard Queue
# Messages can be duplicated, make processing idempotent

class PaymentProcessor
  def process(message_id, data)
    # ANTI-PATTERN: not idempotent — could charge twice
    card.charge(data[:amount])
    Payment.create!(amount: data[:amount])

    # CORRECT: idempotent — check whether it was already processed
    return if Payment.exists?(message_id: message_id)

    ActiveRecord::Base.transaction do
      card.charge(data[:amount])
      Payment.create!(
        message_id: message_id,   # store the SQS Message ID for idempotency
        amount:     data[:amount],
        status:     "success"
      )
    end
  end
end

# EXACTLY-ONCE — FIFO Queue
# Use ContentBasedDeduplication or MessageDeduplicationId
sqs.send_message(
  queue_url:                "https://sqs.../payments.fifo",
  message_body:             JSON.generate({ payment_id: 999 }),
  message_group_id:         "payment-#{user_id}",
  message_deduplication_id: "payment-#{payment_id}"
  # SQS will reject messages with the same deduplication_id within 5 minutes
)

Monitoring with CloudWatch #

require 'aws-sdk-cloudwatch'

cloudwatch = Aws::CloudWatch::Client.new(region: "ap-southeast-1")

# Create an alarm for high queue depth
cloudwatch.put_metric_alarm(
  alarm_name:          "sqs-new-orders-high-depth",
  alarm_description:   "The new-orders queue has too many messages",
  actions_enabled:     true,
  alarm_actions:       ["arn:aws:sns:ap-southeast-1:123:ops-team"],
  metric_name:         "ApproximateNumberOfMessagesVisible",
  namespace:           "AWS/SQS",
  statistic:           "Average",
  dimensions: [{
    name:  "QueueName",
    value: "new-orders"
  }],
  period:              300,     # evaluate every 5 minutes
  evaluation_periods:  1,
  threshold:           1000.0,  # alert if > 1000 messages
  comparison_operator: "GreaterThanThreshold",
  treat_missing_data:  "notBreaching"
)

# Create an alarm for the DLQ
cloudwatch.put_metric_alarm(
  alarm_name:          "sqs-orders-dlq-not-empty",
  alarm_description:   "There are messages in the orders DLQ!",
  actions_enabled:     true,
  alarm_actions:       ["arn:aws:sns:ap-southeast-1:123:ops-team"],
  metric_name:         "ApproximateNumberOfMessagesVisible",
  namespace:           "AWS/SQS",
  statistic:           "Sum",
  dimensions: [{ name: "QueueName", value: "new-orders-dlq" }],
  period:              60,
  evaluation_periods:  1,
  threshold:           0,
  comparison_operator: "GreaterThanThreshold",
  treat_missing_data:  "notBreaching"
)

Security — Proper IAM Policies #

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ProducerPermissions",
      "Effect": "Allow",
      "Action": [
        "sqs:SendMessage",
        "sqs:SendMessageBatch",
        "sqs:GetQueueUrl",
        "sqs:GetQueueAttributes"
      ],
      "Resource": "arn:aws:sqs:ap-southeast-1:123456789:orders-*"
    },
    {
      "Sid": "ConsumerPermissions",
      "Effect": "Allow",
      "Action": [
        "sqs:ReceiveMessage",
        "sqs:DeleteMessage",
        "sqs:DeleteMessageBatch",
        "sqs:ChangeMessageVisibility",
        "sqs:GetQueueUrl",
        "sqs:GetQueueAttributes"
      ],
      "Resource": "arn:aws:sqs:ap-southeast-1:123456789:orders-*"
    }
  ]
}

# SQS security best practices:
# 1. Use IAM Roles, not Access Keys, for EC2/ECS/Lambda
# 2. Principle of least privilege — producers don't need receive, consumers don't need create
# 3. Server-side encryption with KMS for sensitive data
# 4. VPC Endpoints so traffic doesn't leave the internet
# 5. Audit with CloudTrail

# Create a queue with KMS encryption
sqs.create_queue(
  queue_name: "encrypted-payments",
  attributes: {
    "KmsMasterKeyId"               => "arn:aws:kms:ap-southeast-1:123:key/xxx",
    "KmsDataKeyReusePeriodSeconds" => "300"
  }
)

Summary #

Standard Queue for high throughput, FIFO for ordering — Standard tolerates duplicates but has nearly unlimited throughput; FIFO has no duplicates but is limited to 300 TPS (3000 with batching).
Long polling saves money — set wait_time_seconds: 20 to reduce the number of empty poll requests; without it, SQS costs can rise dramatically for quiet queues.
The visibility timeout must be longer than the process — if processing takes 60 seconds but the timeout is 30 seconds, the message reappears and gets processed twice; set the timeout to 2-3x the estimated processing time.
Always delete messages after success — SQS doesn’t auto-delete after receive; call delete_message on success, or let the visibility timeout expire to retry.
A DLQ is mandatory for production queues — configure RedrivePolicy with maxReceiveCount: 3; persistently failing messages won’t block the queue and can be analyzed in the DLQ.
Processing must be idempotent for Standard Queues — store the SQS message_id in the database and check before processing to prevent unwanted duplicate effects.
Batch operations save cost and latency — send_message_batch and delete_message_batch send/delete up to 10 messages in a single API call.
IAM Roles, not Access Keys, for production — EC2, ECS, and Lambda can use IAM Roles automatically without storing credentials in code or the environment.
SNS → SQS for fan-out — publish once to SNS, automatically delivered to many SQS queues; ideal for events that need processing by several different systems.
CloudWatch alarms for the DLQ — create an alarm that triggers when the DLQ isn’t empty; a message in the DLQ always means a bug or invalid data needing immediate attention.

← Previous: RabbitMQ Next: Google Pub/Sub →