RabbitMQ Setup

You are a messaging infrastructure engineer. The user wants to set up RabbitMQ with exchanges, queues, and bindings for a pub/sub or work queue pattern.

What to check first

• Run rabbitmq-server --version to confirm RabbitMQ is installed and accessible
• Verify the RabbitMQ management plugin is enabled: rabbitmq-plugins list | grep management
• Check that the RabbitMQ broker is running on localhost:5672 (or your target host)

Steps

1. Install the pika library (Python RabbitMQ client): pip install pika
2. Connect to the RabbitMQ broker using pika.BlockingConnection() with your host, port, and credentials
3. Create a channel from the connection using connection.channel()
4. Declare an exchange with channel.exchange_declare() — specify type (direct, fanout, or topic) and durable=True for persistence
5. Declare a queue with channel.queue_declare() — set durable=True and arguments for TTL or max length if needed
6. Bind the queue to the exchange using channel.queue_bind() with the routing key pattern
7. Set up a consumer callback with channel.basic_consume() and start consuming with channel.start_consuming()
8. For publishing, use channel.basic_publish() with the exchange name and routing key

Code

import pika
import json

# Connection parameters
credentials = pika.PlainCredentials('guest', 'guest')
parameters = pika.ConnectionParameters(
    host='localhost',
    port=5672,
    credentials=credentials,
    heartbeat=600
)

# Establish connection
connection = pika.BlockingConnection(parameters)
channel = connection.channel()

# Declare exchange (fanout for broadcasting, direct for routing)
channel.exchange_declare(
    exchange='user_events',
    exchange_type='direct',
    durable=True
)

# Declare queue with TTL and max length
channel.queue_declare(
    queue='user_notifications',
    durable=True,
    arguments={
        'x-message-ttl': 86400000,  # 24 hours in ms
        'x-max-length': 10000
    }
)

# Bind queue to exchange with routing key
channel.queue_bind(
    exchange='user_events',
    queue='user_notifications',
    routing_key='user.created'
)

# Consumer callback function
def on_message(ch, method, properties, body):
    try:
        message = json.loads(body)
        print(f"Received: {message}")
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        print(f"Error: {e}")
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True

Note: this example was truncated in the source. See the GitHub repo for the latest full version.

Common Pitfalls

• Treating this skill as a one-shot solution — most workflows need iteration and verification
• Skipping the verification steps — you don't know it worked until you measure
• Applying this skill without understanding the underlying problem — read the related docs first

When NOT to Use This Skill

• When a simpler manual approach would take less than 10 minutes
• On critical production systems without testing in staging first
• When you don't have permission or authorization to make these changes

How to Verify It Worked

• Run the verification steps documented above
• Compare the output against your expected baseline
• Check logs for any warnings or errors — silent failures are the worst kind

Production Considerations

• Test in staging before deploying to production
• Have a rollback plan — every change should be reversible
• Monitor the affected systems for at least 24 hours after the change