RabbitMQ: Message Broker Guide for Developers

# Using Docker (recommended for development)
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4-management

# Verify it's running
docker logs rabbitmq --tail 5

# connect.py
# Connect to RabbitMQ and declare a queue
import pika

# Create connection
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost', 5672)
)
channel = connection.channel()

# Declare a queue (idempotent — creates only if not exists)
channel.queue_declare(queue='hello', durable=True)

print("Connected to RabbitMQ. Queue 'hello' is ready.")

connection.close()

# direct_exchange.py
# Direct exchange: routing key must match binding key exactly

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Declare exchange
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')

# Declare queues
channel.queue_declare(queue='error_queue', durable=True)
channel.queue_declare(queue='warning_queue', durable=True)

# Bind queues with routing keys
channel.queue_bind(exchange='direct_logs', queue='error_queue', routing_key='error')
channel.queue_bind(exchange='direct_logs', queue='warning_queue', routing_key='warning')

# Publish messages
channel.basic_publish(
    exchange='direct_logs',
    routing_key='error',
    body='[ERROR] Disk space critical'
)
channel.basic_publish(
    exchange='direct_logs',
    routing_key='warning',
    body='[WARNING] Memory usage 85%'
)

print("Sent: [ERROR] Disk space critical → error_queue")
print("Sent: [WARNING] Memory usage 85% → warning_queue")

connection.close()

# topic_exchange.py
# Topic exchange: routing key patterns with wildcards

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
channel.queue_declare(queue='all_logs', durable=True)
channel.queue_declare(queue='error_logs', durable=True)

# Bind with patterns
channel.queue_bind(exchange='topic_logs', queue='all_logs', routing_key='#')
channel.queue_bind(exchange='topic_logs', queue='error_logs', routing_key='*.error')

# Publish messages with topic routing keys
channel.basic_publish(
    exchange='topic_logs',
    routing_key='app1.error',
    body='App1 crashed: OOM'
)
channel.basic_publish(
    exchange='topic_logs',
    routing_key='app2.info',
    body='App2 started successfully'
)
channel.basic_publish(
    exchange='topic_logs',
    routing_key='db.warning',
    body='DB connection pool > 80%'
)

print("Published messages with topic routing keys")

connection.close()

# fanout_exchange.py
# Fanout exchange: broadcast to all queues

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='broadcast', exchange_type='fanout')

# Create exclusive queues (auto-deleted when consumer disconnects)
result = channel.queue_declare(queue='', exclusive=True)
queue_name1 = result.method.queue
result = channel.queue_declare(queue='', exclusive=True)
queue_name2 = result.method.queue

# Bind both queues to the fanout exchange
channel.queue_bind(exchange='broadcast', queue=queue_name1)
channel.queue_bind(exchange='broadcast', queue=queue_name2)

# Broadcast message
channel.basic_publish(
    exchange='broadcast',
    routing_key='',
    body='System maintenance in 5 minutes'
)

print(f"Broadcast message sent to {queue_name1} and {queue_name2}")

connection.close()

# consumer.py
# Consume messages from a queue with acknowledgment

import pika
import time

def callback(ch, method, properties, body):
    """Process a received message."""
    print(f"Received: {body.decode()}")
    print(f"Routing key: {method.routing_key}")
    print(f"Delivery tag: {method.delivery_tag}")

    # Simulate processing
    time.sleep(0.5)

    # Acknowledge the message
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print(f"Acknowledged: {method.delivery_tag}")

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='task_queue', durable=True)

# Fair dispatch — don't give more than one message at a time
channel.basic_qos(prefetch_count=1)

channel.basic_consume(
    queue='task_queue',
    on_message_callback=callback,
    auto_ack=False  # Manual acknowledgment
)

print('Waiting for messages. Press Ctrl+C to exit.')
channel.start_consuming()

# reliable_publisher.py
# Durable queues, persistent messages, and publisher confirms

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Enable publisher confirms
channel.confirm_delivery()

# Durable queue survives broker restarts
channel.queue_declare(queue='critical_tasks', durable=True)

# Publish persistent message
try:
    channel.basic_publish(
        exchange='',
        routing_key='critical_tasks',
        body='Process payment: Order #1234',
        properties=pika.BasicProperties(
            delivery_mode=2,  # Persistent message
            content_type='text/plain',
        ),
        mandatory=True,  # Return if no queue bound
    )
    print("Message published and confirmed")
except pika.exceptions.UnroutableError:
    print("Message could not be routed")
except pika.exceptions.NackError:
    print("Publisher negative acknowledgment — message lost")

# Join a node to a cluster
docker run -d --name rabbitmq2 -p 5673:5672 --link rabbitmq \
  -e CLUSTERED=true \
  -e CLUSTER_WITH=rabbitmq \
  -e RABBITMQ_ERLANG_COOKIE=secret_cookie \
  rabbitmq:4-management

rabbitmqctl set_policy ha-all ".*" '{"ha-mode":"all","ha-sync-mode":"automatic"}'

# dead_letter_setup.py
# Configure a dead letter queue for failed messages

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Dead letter exchange and queue
channel.exchange_declare(exchange='dlx', exchange_type='fanout', durable=True)
channel.queue_declare(queue='dead_letter_queue', durable=True)
channel.queue_bind(exchange='dlx', queue='dead_letter_queue')

# Main queue with dead letter config
args = {
    'x-dead-letter-exchange': 'dlx',
    'x-message-ttl': 60000,  # 60 seconds
    'x-max-retries': 3,
}
channel.queue_declare(
    queue='main_orders',
    durable=True,
    arguments=args
)

# Publish a message that will fail
channel.basic_publish(
    exchange='',
    routing_key='main_orders',
    body='Process order: #9999',
    properties=pika.BasicProperties(delivery_mode=2)
)
print("Published order to main_orders. If processing fails, it moves to dead_letter_queue.")

connection.close()