Message Queues

Overview

Message queues provide asynchronous communication between distributed system components by storing messages in an intermediate buffer until the receiving component can process them. This decoupling allows producers to send messages without waiting for consumers to be available, and consumers to process messages at their own pace without blocking producers.

The message queue pattern emerged from the need to handle workloads that don't require immediate processing and to prevent system failures from cascading across components. When a web application needs to send an email, resize an image, or update a recommendation engine, executing these tasks synchronously would slow down the user's request. Message queues move these operations into background jobs that execute independently of the main application flow.

A message queue system contains three primary components: producers that generate messages, a broker that stores and routes messages, and consumers that process messages. The broker ensures messages persist even if consumers are temporarily unavailable, providing durability guarantees that prevent data loss during system failures.

# Producer sends a message
MessageQueue.publish('email.send', {
  to: 'user@example.com',
  subject: 'Welcome',
  body: 'Thank you for signing up'
})

# Consumer processes messages
MessageQueue.subscribe('email.send') do |message|
  EmailService.deliver(message[:to], message[:subject], message[:body])
end

Message queues solve several distributed systems problems: they absorb traffic spikes by buffering incoming work, enable horizontal scaling of workers, provide fault tolerance through message persistence, and allow independent deployment and scaling of system components. The pattern transforms synchronous dependencies into asynchronous workflows, reducing coupling and increasing system resilience.

Key Principles

Message queues operate on a producer-consumer model where producers generate messages without knowledge of who will process them, and consumers process messages without knowledge of who generated them. This separation creates loose coupling between system components, allowing each to evolve independently.

Message Structure: Each message contains a payload (the data to process) and metadata (routing information, priority, timestamps). Messages are immutable once created. The payload typically uses a serialization format like JSON, MessagePack, or Protocol Buffers. Metadata includes the routing key that determines which queue receives the message, delivery mode flags, expiration times, and correlation identifiers for tracking related messages.

Delivery Guarantees: Message queues provide different delivery semantics. At-most-once delivery sends each message once without confirmation, accepting potential message loss for lower latency. At-least-once delivery confirms receipt and retries on failure, ensuring no messages are lost but potentially delivering duplicates. Exactly-once delivery guarantees each message processes once through distributed transactions or idempotency mechanisms, though implementing true exactly-once semantics requires careful coordination between the broker and consumers.

Acknowledgment Pattern: Consumers explicitly acknowledge messages after successful processing. The broker maintains messages in an unacknowledged state until receiving confirmation, allowing message redelivery if a consumer fails. This acknowledgment mechanism provides fault tolerance without requiring producers to track consumer state.

# Manual acknowledgment pattern
consumer.subscribe('image.resize', manual_ack: true) do |delivery_info, properties, payload|
  begin
    image = JSON.parse(payload)
    ImageProcessor.resize(image['path'], image['dimensions'])
    consumer.ack(delivery_info.delivery_tag)
  rescue => e
    consumer.nack(delivery_info.delivery_tag, false, true) # reject and requeue
  end
end

Message Ordering: Queues typically preserve FIFO ordering for messages sent to the same queue, but ordering across multiple queues or consumers requires additional coordination. Partitioned queues assign messages to specific partitions based on a key, ensuring messages with the same key process in order. Global ordering across all messages often requires single-consumer processing, sacrificing parallel execution for ordering guarantees.

Durability and Persistence: Messages can exist only in memory for maximum throughput or persist to disk for durability. Durable messages survive broker restarts but incur write latency. Transient messages provide lower latency but risk loss during failures. The durability choice depends on whether the application can tolerate message loss or requires guaranteed delivery.

Dead Letter Queues: When messages fail processing repeatedly, queues route them to a dead letter queue rather than retrying indefinitely. This prevents poison messages from blocking queue processing and provides a mechanism for manual inspection and remediation of problematic messages.

# Dead letter queue configuration
queue = channel.queue('orders.process', arguments: {
  'x-dead-letter-exchange' => 'orders.dlx',
  'x-message-ttl' => 86400000, # 24 hours
  'x-max-length' => 10000
})

Backpressure: When consumers cannot keep pace with incoming messages, queues implement backpressure through prefetch limits that control how many unacknowledged messages a consumer can hold. This prevents overwhelming slow consumers and provides time-based flow control across the system.

Implementation Approaches

Message queue implementations follow several architectural patterns, each optimized for different use cases and consistency requirements.

Work Queue Pattern: Multiple workers consume from a single queue, distributing tasks across available processors. The broker assigns each message to exactly one consumer using round-robin or weighted distribution. This pattern scales horizontal processing capacity and provides automatic load balancing. Workers compete for messages, with faster workers processing more tasks. The pattern suits CPU-intensive work that can execute independently, like image processing, report generation, or data transformation.

Publish-Subscribe Pattern: Producers publish messages to topics, and multiple subscribers receive copies of each message. The broker maintains separate queues for each subscriber, copying messages to all interested consumers. This pattern distributes events to multiple systems that need to react to the same occurrence, like updating a cache, sending notifications, and logging analytics when a user signs up.

# Publish-subscribe with topic exchange
exchange = channel.topic('user.events')

# Multiple consumers receive the same messages
analytics_queue = channel.queue('analytics.user_events')
analytics_queue.bind(exchange, routing_key: 'user.#')

notification_queue = channel.queue('notifications.user_events')
notification_queue.bind(exchange, routing_key: 'user.signup')

cache_queue = channel.queue('cache.user_events')
cache_queue.bind(exchange, routing_key: 'user.*')

Request-Reply Pattern: A client sends a message and waits for a response, implementing synchronous RPC semantics over asynchronous messaging. The request includes a reply-to queue and correlation identifier. The server processes the request and sends the response to the specified reply queue with the matching correlation ID. This pattern combines asynchronous messaging benefits with synchronous semantics when the client needs the result before continuing.

Priority Queue Pattern: Messages carry priority values, and the broker delivers high-priority messages before low-priority ones within the same queue. This ensures time-sensitive operations process before routine background tasks. Priority queues require careful design to prevent starvation of low-priority messages.

Competing Consumers Pattern: Multiple consumer instances process messages from the same queue, providing fault tolerance and load distribution. If one consumer fails, others continue processing. The broker distributes messages across consumers and redelivers messages from failed consumers. This pattern combines redundancy with parallel processing.

Message Routing: Brokers route messages based on patterns matching the routing key. Direct routing delivers messages to queues with exact key matches. Topic routing uses wildcard patterns for flexible subscription. Fanout routing sends messages to all bound queues regardless of routing keys. Header routing examines message headers for routing decisions.

# Topic-based routing with patterns
exchange = channel.topic('logs')

# Receives all error logs from any service
error_queue = channel.queue('errors')
error_queue.bind(exchange, routing_key: '*.error')

# Receives all logs from the payment service
payment_queue = channel.queue('payment_logs')
payment_queue.bind(exchange, routing_key: 'payment.*')

# Receives only critical errors
critical_queue = channel.queue('critical')
critical_queue.bind(exchange, routing_key: '*.error.critical')

Ruby Implementation

Ruby applications interact with message queues through client libraries that abstract broker-specific protocols. The ecosystem includes gems for job processing, broker clients, and complete background job frameworks.

Sidekiq: The most popular Ruby background job processor uses Redis as its message broker. Sidekiq provides job queuing, scheduling, retries, and a web interface for monitoring. Jobs are Ruby classes with a perform method. Sidekiq serializes job arguments to JSON and stores them in Redis lists.

# Define a job
class ImageResizeWorker
  include Sidekiq::Worker
  sidekiq_options queue: 'images', retry: 5

  def perform(image_id, width, height)
    image = Image.find(image_id)
    resized = ImageProcessor.resize(image.path, width, height)
    image.update(resized_path: resized)
  end
end

# Enqueue a job
ImageResizeWorker.perform_async(image_id, 800, 600)

# Schedule for later
ImageResizeWorker.perform_in(1.hour, image_id, 800, 600)
ImageResizeWorker.perform_at(tomorrow_noon, image_id, 800, 600)

Sidekiq workers poll Redis for jobs, deserialize the arguments, and execute the perform method. Failed jobs automatically retry with exponential backoff. The retry count, delay, and dead queue behavior are configurable per job class. Sidekiq uses connection pooling and multi-threading to maximize throughput from a single process.

Bunny: A RabbitMQ client for Ruby that implements AMQP 0.9.1. Bunny provides low-level control over exchanges, queues, bindings, and message properties. Applications create channels, declare queues and exchanges, publish messages, and consume with callbacks or blocking iteration.

require 'bunny'

connection = Bunny.new(host: 'localhost', port: 5672)
connection.start
channel = connection.create_channel

# Declare durable queue
queue = channel.queue('orders', durable: true)

# Publish with persistence
channel.default_exchange.publish(
  order_data.to_json,
  routing_key: queue.name,
  persistent: true,
  content_type: 'application/json',
  timestamp: Time.now.to_i
)

# Consume with manual acknowledgment
queue.subscribe(manual_ack: true, block: true) do |delivery_info, properties, body|
  order = JSON.parse(body)
  begin
    OrderProcessor.process(order)
    channel.ack(delivery_info.delivery_tag)
  rescue => error
    channel.nack(delivery_info.delivery_tag, false, false)
    ErrorLogger.log(error, order)
  end
end

Bunny supports publisher confirms for guaranteed delivery, consumer prefetch for flow control, and message TTL for automatic expiration. The library handles connection recovery, provides thread-safe channel operations, and exposes RabbitMQ-specific features like exchange-to-exchange bindings and alternate exchanges.

Sneakers: A background processing framework built on RabbitMQ and Bunny. Sneakers provides a Sidekiq-like API for defining workers while using RabbitMQ as the broker. Workers declare which queues they consume from, and Sneakers handles thread pools, connection management, and graceful shutdown.

class EmailWorker
  include Sneakers::Worker
  from_queue 'emails',
    threads: 10,
    prefetch: 10,
    timeout_job_after: 60,
    ack: true

  def work(msg)
    data = JSON.parse(msg)
    EmailService.deliver(data['to'], data['subject'], data['body'])
    ack!
  rescue => error
    ErrorLogger.log(error, msg)
    reject!
  end
end

Resque: Redis-backed job processing inspired by DelayedJob. Resque uses Redis lists for queues and Ruby processes for workers. Each job is a Ruby class with a perform class method. Resque prioritizes reliability over throughput, processing one job per worker process.

class ReportGenerator
  @queue = :reports

  def self.perform(user_id, report_type, start_date, end_date)
    user = User.find(user_id)
    report = ReportBuilder.generate(report_type, start_date, end_date)
    ReportMailer.deliver(user.email, report)
  end
end

# Enqueue
Resque.enqueue(ReportGenerator, user.id, 'sales', Date.today - 30, Date.today)

AWS SDK for SQS: Amazon's simple queue service client provides managed message queuing without running infrastructure. The AWS SDK handles authentication, request signing, and retry logic. SQS offers standard queues with high throughput and best-effort ordering, and FIFO queues with guaranteed ordering and exactly-once processing.

require 'aws-sdk-sqs'

sqs = Aws::SQS::Client.new(region: 'us-east-1')
queue_url = sqs.get_queue_url(queue_name: 'orders').queue_url

# Send message
sqs.send_message(
  queue_url: queue_url,
  message_body: order.to_json,
  message_attributes: {
    'Priority' => { string_value: 'high', data_type: 'String' },
    'Source' => { string_value: 'web', data_type: 'String' }
  }
)

# Receive and process
loop do
  response = sqs.receive_message(
    queue_url: queue_url,
    max_number_of_messages: 10,
    wait_time_seconds: 20,
    visibility_timeout: 30
  )

  response.messages.each do |message|
    process_order(JSON.parse(message.body))
    sqs.delete_message(queue_url: queue_url, receipt_handle: message.receipt_handle)
  end
end

Shoryuken: An SQS-based background processing library that provides a Sidekiq-like interface for AWS SQS. Shoryuken handles polling, visibility timeout extension, and automatic deletion of processed messages.

Design Considerations

Message queues introduce asynchronous complexity that requires careful evaluation against application requirements. The decision to adopt message queues depends on whether the benefits of decoupling and scalability outweigh the operational complexity.

Synchronous vs Asynchronous Processing: Synchronous processing provides immediate feedback and simpler error handling but couples components tightly and limits scalability. Asynchronous processing through queues decouples components and improves throughput but complicates error handling and makes debugging harder. Use synchronous processing when the user needs immediate results or when operations must complete in a specific order. Use message queues when operations can execute independently, when traffic varies significantly, or when processing time exceeds acceptable request latency.

Message Queue vs Direct API Calls: Direct API calls provide strong consistency and immediate error feedback but create tight coupling and fail when the called service is unavailable. Message queues provide loose coupling and buffer capacity spikes but require eventual consistency thinking and more complex error tracking. Choose direct calls for operations requiring immediate confirmation, like payment processing or inventory reservation. Choose message queues for operations that can complete asynchronously, like sending emails, generating reports, or updating analytics.

Broker Selection Criteria: Different message brokers optimize for different characteristics. Redis-based queues (Sidekiq, Resque) provide simple deployment and high throughput but limited durability guarantees and routing features. RabbitMQ provides strong durability, sophisticated routing, and clustering but requires more operational knowledge. Cloud-managed services (SQS, Cloud Pub/Sub) eliminate infrastructure management but increase per-message costs and introduce vendor lock-in. Apache Kafka provides high-throughput stream processing but adds significant operational complexity.

Choose Redis for job queues in applications already using Redis, when message loss during failures is acceptable, or when throughput is the primary concern. Choose RabbitMQ when durability and complex routing are required, when running on-premises infrastructure, or when needing sophisticated queue features. Choose managed services when minimizing operational burden justifies the cost, when scaling unpredictably, or when integrating with cloud-native applications.

Queue Granularity: Applications must decide between few queues with high throughput and many specialized queues with targeted processing. Few queues simplify operations and reduce connection overhead but make priority handling and scaling more difficult. Many queues enable fine-grained priority and scaling but increase complexity and broker resource usage.

Create separate queues for different priority levels, different processing characteristics (CPU-bound vs I/O-bound), and different retry behaviors. Avoid creating queues per tenant or per entity, which scales poorly. Balance queue count against operational complexity and broker capabilities.

Message Size Constraints: Large messages increase broker memory usage, network bandwidth, and processing latency. When messages exceed a few kilobytes, consider storing the data externally and passing only a reference through the queue. This pattern reduces broker load and enables processing large datasets without queue size limits.

# Store large payload externally
s3_key = S3Service.upload(large_dataset)
VideoTranscodeWorker.perform_async(s3_key)

# Worker retrieves payload
def perform(s3_key)
  dataset = S3Service.download(s3_key)
  process(dataset)
  S3Service.delete(s3_key)
end

Idempotency Requirements: At-least-once delivery means consumers may process messages multiple times. Operations must either be naturally idempotent or implement idempotency explicitly. Design message handlers to produce the same result when executed multiple times with the same input. Store message identifiers in a database to detect and skip duplicate processing, or use upsert operations that produce the same final state regardless of execution count.

Monitoring and Observability: Message queues hide processing behind asynchronous boundaries, making observability critical. Track queue depth, message age, processing time, error rates, and consumer lag. Set alerts for queue depth exceeding thresholds, old messages indicating stalled processing, and elevated error rates signaling systemic issues. Include correlation IDs in messages to trace work across queue boundaries.

Tools & Ecosystem

The message queue ecosystem includes brokers, client libraries, monitoring tools, and managed services that handle different aspects of asynchronous messaging.

RabbitMQ: An open-source message broker implementing AMQP with plugins for MQTT, STOMP, and other protocols. RabbitMQ provides durable queues, flexible routing through exchanges, clustering for high availability, and federation for distributed deployments. The management plugin offers a web UI for monitoring and administration. RabbitMQ excels at reliable message delivery with complex routing requirements and moderate throughput needs.

Apache Kafka: A distributed streaming platform designed for high-throughput event streaming. Kafka stores messages in partitioned, replicated logs rather than transient queues. Consumers track their offset in the log, enabling message replay and multiple consumer groups reading the same data at different rates. Kafka fits event sourcing, log aggregation, and stream processing use cases requiring message retention and replay capabilities.

Redis: An in-memory data store that provides list and pub/sub operations used for message queuing. Redis offers exceptional throughput and simple operations but limited durability and no complex routing. Redis-based queues work for job queues where occasional message loss during failures is acceptable and when Redis is already part of the infrastructure.

Amazon SQS: A managed message queue service that provides standard queues with high throughput and FIFO queues with ordering guarantees. SQS handles scaling, replication, and failure recovery automatically. The service charges per message, making it cost-effective for variable workloads and expensive for high-volume applications. SQS integrates natively with other AWS services and eliminates operational overhead.

Google Cloud Pub/Sub: A managed publish-subscribe service providing global message delivery with at-least-once semantics. Pub/Sub automatically scales and replicates messages across regions. The service supports push subscriptions that deliver messages to HTTP endpoints and pull subscriptions where consumers poll for messages. Pub/Sub fits cloud-native applications requiring global distribution and automatic scaling.

ActiveMQ: A Java-based message broker supporting JMS, AMQP, MQTT, and STOMP protocols. ActiveMQ provides clustering, master-slave failover, and message groups for ordered processing. The broker suits enterprise environments requiring JMS compatibility and multiple protocol support.

NATS: A lightweight message broker focused on simplicity and performance. NATS provides pub/sub messaging with optional persistence and streaming through NATS Streaming. The broker emphasizes low latency and high throughput for cloud-native microservices communication.

Monitoring Tools: Queue monitoring requires visibility into message flow, processing latency, and error rates. RabbitMQ provides a management plugin with metrics and a HTTP API. Prometheus exporters collect metrics from various brokers for time-series analysis. DataDog, New Relic, and AWS CloudWatch offer managed monitoring with alerting. Application Performance Monitoring tools trace requests across queue boundaries using correlation identifiers.

Ruby Gems: The Ruby ecosystem includes gems for different brokers and patterns. Bunny and March Hare (JRuby) provide RabbitMQ clients. Sidekiq, Resque, and DelayedJob offer background job processing. Sneakers combines RabbitMQ with background processing. Shoryuken and ActiveJob's SQS adapter integrate with Amazon SQS. Karafka provides Kafka consumer framework. Wisper implements in-process pub/sub for decoupled Ruby code.

Common Patterns

Message queue implementations follow established patterns that solve recurring distributed messaging problems.

Retry Pattern: Failed messages retry with exponential backoff to handle transient failures without overwhelming the system. The pattern increases delay between retries, giving temporary problems time to resolve. After exceeding maximum retries, messages move to a dead letter queue for manual investigation.

class DocumentProcessor
  include Sidekiq::Worker
  sidekiq_options retry: 10

  sidekiq_retry_in do |count|
    10 * (count + 1) # 10, 20, 30, 40... seconds
  end

  sidekiq_retries_exhausted do |msg, ex|
    Sidekiq.logger.warn "Failed #{msg['class']} with #{msg['args']}: #{ex.message}"
    ErrorNotifier.alert(msg, ex)
  end

  def perform(document_id)
    DocumentService.process(document_id)
  end
end

Message Deduplication Pattern: When at-least-once delivery causes duplicate messages, track processed message identifiers to skip duplicates. Store message IDs in Redis or a database table with an expiration matching the maximum redelivery window. Before processing, check if the message ID exists; if so, acknowledge without processing.

def perform(message_id, data)
  return if REDIS.get("processed:#{message_id}")

  process_data(data)

  REDIS.setex("processed:#{message_id}", 86400, "1")
end

Saga Pattern: Coordinate multi-step processes across services using message choreography. Each service publishes events about its actions, and other services react by performing their steps and publishing their own events. This pattern implements distributed transactions without a central coordinator. If a step fails, compensating actions undo previous steps.

# Order service publishes order created event
EventBus.publish('order.created', {
  order_id: order.id,
  user_id: order.user_id,
  items: order.items
})

# Inventory service reserves items
EventBus.subscribe('order.created') do |event|
  if InventoryService.reserve(event[:items])
    EventBus.publish('inventory.reserved', event)
  else
    EventBus.publish('inventory.reservation_failed', event)
  end
end

# Payment service charges customer
EventBus.subscribe('inventory.reserved') do |event|
  if PaymentService.charge(event[:user_id], event[:total])
    EventBus.publish('payment.completed', event)
  else
    EventBus.publish('payment.failed', event)
  end
end

# Compensate on failure
EventBus.subscribe('payment.failed') do |event|
  InventoryService.release(event[:items])
  EventBus.publish('order.cancelled', event)
end

Outbox Pattern: Ensure messages publish when database changes commit by writing both the database record and the message to send within the same transaction. A separate process polls the outbox table and publishes messages to the queue, then marks them as sent. This guarantees messages send exactly once per database change without distributed transactions.

# Write both database record and outbox entry in transaction
ActiveRecord::Base.transaction do
  user = User.create!(email: email, name: name)
  
  Outbox.create!(
    aggregate_id: user.id,
    aggregate_type: 'User',
    event_type: 'user.created',
    payload: { id: user.id, email: user.email }
  )
end

# Separate process publishes outbox entries
loop do
  Outbox.unpublished.find_each do |entry|
    MessageQueue.publish(entry.event_type, entry.payload)
    entry.update!(published_at: Time.current)
  end
  sleep 1
end

Circuit Breaker Pattern: Protect failing services by stopping message processing when error rates exceed thresholds. The circuit breaker tracks failure rates and transitions between closed (normal), open (failing), and half-open (testing) states. When open, messages are rejected immediately without attempting processing. After a timeout, the circuit enters half-open state to test if the service recovered.

Message Filtering Pattern: Consumers filter messages based on content or metadata to process only relevant messages. Broker-side filtering using message properties reduces network traffic and consumer load. Client-side filtering provides more flexible logic but requires receiving all messages.

# Broker-side filtering with header matching
queue.bind(exchange, routing_key: 'notification.#', arguments: {
  'x-match' => 'all',
  'region' => 'us-east-1',
  'priority' => 'high'
})

# Client-side filtering
queue.subscribe do |delivery_info, properties, body|
  message = JSON.parse(body)
  next unless message['user_id'] == current_user_id
  next unless message['type'].in?(['urgent', 'important'])
  
  process_notification(message)
end

Claim Check Pattern: Handle large messages by storing the payload externally and passing a reference through the queue. The consumer uses the reference to retrieve the full payload from the storage system. This pattern reduces queue message size while enabling processing of arbitrarily large data.

Message Aggregation Pattern: Combine multiple related messages into a single batch before processing. An aggregator collects messages until reaching a count threshold or time window, then processes the batch. This pattern improves throughput for operations with high fixed costs per invocation.

Reference

Message Properties

Delivery Guarantees

Queue Configuration Options

Consumer Configuration

Ruby Queue Libraries Comparison

Message Broker Feature Comparison

Monitoring Metrics

Common Message Patterns

Sidekiq Job Options

RabbitMQ Exchange Types