RabbitMQ Python Cheat-Sheet

March 21, 2025•2 min read

RabbitMQ Docker setup

Run RabbitMQ with the management plugin enabled:

docker run -d \
  --name rabbitmq \
  -p 5672:5672 \
  -p 15672:15672 \
  rabbitmq:management

The connection string is amqp://localhost:5672

You can connect to your instance using RabbitGUI at http://localhost:15672 (guest/guest)

Docker compose setup for RabbitMQ

Create a docker-compose.yml file with custom credentials and persistent storage:

version: '3'
services:
  rabbitmq:
    image: rabbitmq:management
    container_name: rabbitmq
    ports:
      - "5672:5672"
      - "15672:15672"
    environment:
      RABBITMQ_DEFAULT_USER: user
      RABBITMQ_DEFAULT_PASS: password
    volumes:
      - rabbitmq_data:/var/lib/rabbitmq
 
volumes:
  rabbitmq_data:

The connection string is amqp://user:password@localhost:5672

You can connect to your instance using RabbitGUI at http://localhost:15672 (user/password)

Installation in Python

Add the pika package to your project:

pip install pika

Producer example

You can publish directly to a queue using the basic_publish method:

import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='my_queue')
 
channel.basic_publish(exchange='', routing_key='my_queue', body='Hello RabbitGUI!')
 
connection.close()

Consumer example

Your consumer can listen to messages from a queue using the basic_consume method:

import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='my_queue')
 
def callback(ch, method, properties, body):
    print(f"Message: {body.decode()}")
 
channel.basic_consume(queue='my_queue', on_message_callback=callback, auto_ack=True)
channel.start_consuming()

Exchanges examples

Direct

A direct exchange routes messages to queues based on exact routing key matches:

channel.exchange_declare(exchange='my_exchange', exchange_type='direct')
channel.queue_declare(queue='my_queue')
channel.queue_bind(exchange='my_exchange', queue='my_queue', routing_key='my.routing.key')
 
channel.basic_publish(exchange='my_exchange', routing_key='my.routing.key', body='Hello RabbitGUI!')
 
def callback(ch, method, properties, body):
    print(f"Message: {body.decode()}")
 
channel.basic_consume(queue='my_queue', on_message_callback=callback)

Fanout

A fanout exchange broadcasts messages to all bound queues, ignoring routing keys:

channel.exchange_declare(exchange='my_exchange', exchange_type='fanout')
channel.queue_declare(queue='my_queue')
channel.queue_bind(exchange='my_exchange', queue='my_queue')
 
channel.basic_publish(exchange='my_exchange', routing_key='', body='Hello RabbitGUI!')
 
def callback(ch, method, properties, body):
    print(f"Message: {body.decode()}")
 
channel.basic_consume(queue='my_queue', on_message_callback=callback)

Topic

A topic exchange routes messages based on wildcard pattern matching (* for one word, # for zero or more words):

channel.exchange_declare(exchange='my_exchange', exchange_type='topic')
channel.queue_declare(queue='my_queue')
channel.queue_bind(exchange='my_exchange', queue='my_queue', routing_key='*.routing.*')
 
channel.basic_publish(exchange='my_exchange', routing_key='my.routing.key', body='Hello RabbitGUI!')
 
def callback(ch, method, properties, body):
    print(f"Message: {body.decode()}")
 
channel.basic_consume(queue='my_queue', on_message_callback=callback)

Message acknowledgements

Automatic acknowledgement

RabbitMQ will dequeue messages as soon as they've been sent down the wire.

channel.basic_consume(queue='my_queue', on_message_callback=callback, auto_ack=True)

Manual acknowledgement

Manually acknowledge messages after processing to ensure they are removed from the queue:

def callback(ch, method, properties, body):
    # ...
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
channel.basic_consume(queue='my_queue', on_message_callback=callback)

Reject and requeue

Reject a message and put it back in the queue for reprocessing:

def callback(ch, method, properties, body):
    # ...
    ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
 
channel.basic_consume(queue='my_queue', on_message_callback=callback)

Reject and do not requeue

Reject a message and discard it (sends to dead letter queue if configured):

def callback(ch, method, properties, body):
    # ...
    ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
 
channel.basic_consume(queue='my_queue', on_message_callback=callback)

Dead Letter Queues

Configure a queue with a dead letter exchange to catch rejected or expired messages:

import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
 
# Create the dead letter exchange and queue
channel.exchange_declare(exchange='dlx_exchange', exchange_type='direct')
channel.queue_declare(queue='dead_letter_queue')
channel.queue_bind(exchange='dlx_exchange', queue='dead_letter_queue', routing_key='')
 
# Create main queue with dead letter configuration
channel.queue_declare(
    queue='my_queue',
    arguments={
        'x-dead-letter-exchange': 'dlx_exchange',
        'x-dead-letter-routing-key': ''
    }
)