protocols-amqp-rabbitmq

RabbitMQ and AMQP 0-9-1 message broker implementation

AMQP and RabbitMQ Implementation

Scope: AMQP 0-9-1 protocol, RabbitMQ broker, exchanges, queues, routing patterns, clustering, high availability Lines: ~350 Last Updated: 2025-10-27

When to Use This Skill

Activate this skill when:

Core Concepts

AMQP 0-9-1 Architecture

AMQP (Advanced Message Queuing Protocol): Open standard application layer protocol for message-oriented middleware.

Key characteristics:

Message flow:

Publisher → Exchange → Binding → Queue → Consumer
            ↓                      ↓
        Routing Key           Delivery Ack

Exchange Types

1. Direct Exchange

Routes messages with a specific routing key to queues bound with that exact key.

Use cases:

Example binding:

Exchange: tasks
Queue: email_tasks   Binding: email
Queue: sms_tasks     Binding: sms
Queue: push_tasks    Binding: push

Message with routing_key="email" → email_tasks queue

2. Fanout Exchange

Routes messages to all bound queues (ignores routing key).

Use cases:

Example:

Exchange: notifications
Queue: slack_notifier
Queue: email_notifier
Queue: sms_notifier

All queues receive every message

3. Topic Exchange

Routes based on wildcard pattern matching of routing keys.

Use cases:

Patterns:

Example:

Routing key: "order.created.us"
Binding: "order.*.*"       → Matches
Binding: "order.created.#" → Matches
Binding: "order.#"         → Matches
Binding: "*.created.*"     → Matches
Binding: "invoice.#"       → No match

4. Headers Exchange

Routes based on message header attributes instead of routing key.

Use cases:

Example:

Headers: {type: "order", region: "us", priority: "high"}
Binding: {type: "order", region: "us"}  → Matches if x-match=all

Common Patterns

Pattern 1: Work Queue

Goal: Distribute tasks among multiple workers.

Setup:

Python example:

import pika

# Publisher
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='tasks', durable=True)

channel.basic_publish(
    exchange='',
    routing_key='tasks',
    body='Task data',
    properties=pika.BasicProperties(delivery_mode=2)  # Persistent
)

# Consumer
def callback(ch, method, properties, body):
    print(f"Processing: {body}")
    # Do work...
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)  # Fair dispatch
channel.basic_consume(queue='tasks', on_message_callback=callback)
channel.start_consuming()

Pattern 2: Pub-Sub

Goal: Broadcast messages to multiple subscribers.

Setup:

Python example:

# Publisher
channel.exchange_declare(exchange='logs', exchange_type='fanout')
channel.basic_publish(exchange='logs', routing_key='', body='Log message')

# Subscriber
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
channel.queue_bind(exchange='logs', queue=queue_name)
channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)

Pattern 3: Routing

Goal: Subscribe to subset of messages.

Setup:

Python example:

# Publisher
channel.exchange_declare(exchange='logs', exchange_type='direct')
channel.basic_publish(exchange='logs', routing_key='error', body='Error log')

# Consumer (only error logs)
channel.queue_bind(exchange='logs', queue=queue_name, routing_key='error')

Pattern 4: RPC

Goal: Remote procedure calls with response.

Setup:

Python example:

# Client
import uuid

class RpcClient:
    def __init__(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        self.channel = self.connection.channel()
        result = self.channel.queue_declare(queue='', exclusive=True)
        self.callback_queue = result.method.queue
        self.channel.basic_consume(
            queue=self.callback_queue,
            on_message_callback=self.on_response,
            auto_ack=True
        )
        self.response = None
        self.corr_id = None

    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body

    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4())
        self.channel.basic_publish(
            exchange='',
            routing_key='rpc_queue',
            properties=pika.BasicProperties(
                reply_to=self.callback_queue,
                correlation_id=self.corr_id,
            ),
            body=str(n)
        )
        while self.response is None:
            self.connection.process_data_events()
        return int(self.response)

Message Durability

Making Messages Persistent

Requirements:

  1. Queue must be durable
  2. Message must be marked persistent
  3. Exchange durability (recommended)

Python example:

# Durable queue
channel.queue_declare(queue='tasks', durable=True)

# Persistent message
channel.basic_publish(
    exchange='',
    routing_key='tasks',
    body='Important task',
    properties=pika.BasicProperties(
        delivery_mode=2,  # Persistent
    )
)

Note: Persistence doesn't guarantee 100% durability. For stronger guarantees, use publisher confirms.

Publisher Confirms

Ensure message was written to disk before considering it sent.

Python example:

channel.confirm_delivery()

try:
    channel.basic_publish(
        exchange='',
        routing_key='tasks',
        body='Task',
        properties=pika.BasicProperties(delivery_mode=2),
        mandatory=True
    )
    print("Message delivered")
except pika.exceptions.UnroutableError:
    print("Message could not be routed")

Acknowledgments

Manual Acknowledgments

Consumer should acknowledge after processing completes.

def callback(ch, method, properties, body):
    try:
        process_message(body)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        # Requeue on failure
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

channel.basic_consume(
    queue='tasks',
    on_message_callback=callback,
    auto_ack=False  # Manual ack
)

Prefetch Count

Control how many messages delivered to consumer before acknowledgment.

channel.basic_qos(prefetch_count=1)  # One at a time (fair dispatch)
channel.basic_qos(prefetch_count=10) # Batch processing

Best practice: Use prefetch_count=1 for long-running tasks to ensure fair distribution.

Dead Letter Exchanges

Route failed messages to another exchange for retry or inspection.

Setup:

# Main queue with DLX
channel.queue_declare(
    queue='tasks',
    durable=True,
    arguments={
        'x-dead-letter-exchange': 'dlx',
        'x-dead-letter-routing-key': 'failed.tasks',
        'x-message-ttl': 60000  # Optional: TTL before DLX
    }
)

# Dead letter exchange
channel.exchange_declare(exchange='dlx', exchange_type='direct')

# Dead letter queue
channel.queue_declare(queue='failed_tasks', durable=True)
channel.queue_bind(exchange='dlx', queue='failed_tasks', routing_key='failed.tasks')

Messages sent to DLX when:

High Availability

Clustering

Setup (3-node cluster):

# Node 1
rabbitmq-server -detached
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl start_app

# Node 2
rabbitmq-server -detached
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app

# Node 3
rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app

Quorum Queues

Replicated queue type with high availability and data safety.

Python example:

channel.queue_declare(
    queue='ha_tasks',
    durable=True,
    arguments={
        'x-queue-type': 'quorum'
    }
)

Characteristics:

Performance Optimization

Connection Pooling

Reuse connections and channels.

import pika.pool

params = pika.ConnectionParameters('localhost')
pool = pika.pool.ConnectionPool(
    lambda: pika.BlockingConnection(params),
    max_size=10,
    max_overflow=20,
    timeout=10,
    recycle=3600
)

with pool.acquire() as connection:
    channel = connection.channel()
    channel.basic_publish(...)

Batch Publishing

Publish multiple messages in one go.

for i in range(1000):
    channel.basic_publish(
        exchange='',
        routing_key='tasks',
        body=f'Task {i}'
    )
# All published in one network round trip

Lazy Queues

Keep messages on disk instead of RAM.

channel.queue_declare(
    queue='large_queue',
    durable=True,
    arguments={'x-queue-mode': 'lazy'}
)

Use when: Queue has many messages (millions) or messages are large.

Monitoring

Key Metrics

Monitor these via management API or Prometheus:

Queue metrics:

Connection metrics:

Node metrics:

Python monitoring example:

import requests

response = requests.get(
    'http://localhost:15672/api/queues',
    auth=('guest', 'guest')
)
queues = response.json()

for queue in queues:
    print(f"{queue['name']}: {queue['messages']} messages")

Best Practices

  1. Use manual acknowledgments for important messages
  2. Set prefetch_count=1 for long-running tasks
  3. Enable publisher confirms for critical data
  4. Use quorum queues for high availability
  5. Monitor queue depths and set alerts
  6. Use dead letter exchanges for failed message handling
  7. Set message TTL to prevent queue buildup
  8. Limit queue length with x-max-length
  9. Use connection pooling in high-throughput scenarios
  10. Test failure scenarios (node crashes, network partitions)

Common Pitfalls

  1. Not using durable queues: Messages lost on broker restart
  2. Auto-acknowledgment: Messages lost if consumer crashes
  3. No prefetch limit: All messages delivered to one consumer
  4. Not handling connection errors: Application crashes on network issues
  5. Publishing to non-existent exchange: Messages silently dropped
  6. Not monitoring: Queue buildup goes unnoticed
  7. Ignoring memory alarms: Broker blocks publishers
  8. Too many connections: File descriptor exhaustion

Resources