Build a RabbitMQ Consumer with Python and Pika

Overview

RabbitMQ is a message broker that routes messages between producers and consumers. Pika is the standard Python client. Below: building a producer and consumer with durable queues, manual acknowledgments, dead-letter exchanges for failed messages, prefetch tuning for fair dispatch, and connection recovery.

When to Use This

• Distributing background work across multiple workers (image processing, email sending)
• Decoupling producers from consumers in a microservice architecture
• Work queues where each message must be processed exactly once
• Systems that need message persistence and guaranteed delivery

Prerequisites

• Python 3.10+
• RabbitMQ server (local or cloud, e.g., CloudAMQP)
• pika package

Solution

1. Producer

import pika
import json
import uuid

def get_connection():
    credentials = pika.PlainCredentials('guest', 'guest')
    params = pika.ConnectionParameters(
        host='localhost',
        port=5672,
        credentials=credentials,
        heartbeat=30,
        blocked_connection_timeout=7200,
    )
    return pika.BlockingConnection(params)

def publish_message(queue_name: str, message: dict):
    connection = get_connection()
    channel = connection.channel()

    # Declare a durable queue — survives RabbitMQ restarts
    channel.queue_declare(queue=queue_name, durable=True)

    channel.basic_publish(
        exchange='',
        routing_key=queue_name,
        body=json.dumps(message),
        properties=pika.BasicProperties(
            delivery_mode=2,  # Persistent message
            message_id=str(uuid.uuid4()),
            content_type='application/json',
            timestamp=int(__import__('time').time()),
        ),
    )
    print(f"Published to {queue_name}: {message['id']}")
    connection.close()

# Usage
publish_message('task_queue', {
    'id': 'task-001',
    'type': 'email',
    'payload': {'to': 'user@example.com', 'subject': 'Welcome'},
})

2. Consumer with Manual Acknowledgments

import pika
import json
import time

def get_connection():
    credentials = pika.PlainCredentials('guest', 'guest')
    params = pika.ConnectionParameters(
        host='localhost',
        port=5672,
        credentials=credentials,
        heartbeat=30,
    )
    return pika.BlockingConnection(params)

def process_message(channel, method, properties, body):
    message = json.loads(body)
    print(f"Received: {message['id']} — {message['type']}")

    try:
        # Simulate work
        time.sleep(1)
        process_task(message)
        print(f"Done: {message['id']}")

        # Acknowledge only after successful processing
        channel.basic_ack(delivery_tag=method.delivery_tag)

    except Exception as e:
        print(f"Failed: {message['id']} — {e}")
        # Reject and requeue — message goes back to the queue
        channel.basic_nack(
            delivery_tag=method.delivery_tag,
            requeue=True,
        )

def process_task(message: dict):
    if message['type'] == 'email':
        send_email(message['payload'])
    elif message['type'] == 'report':
        generate_report(message['payload'])
    else:
        raise ValueError(f"Unknown task type: {message['type']}")

def start_consumer(queue_name: str):
    connection = get_connection()
    channel = connection.channel()

    # Declare queue as durable (must match producer)
    channel.queue_declare(queue=queue_name, durable=True)

    # Fair dispatch — don't dispatch a new message to a worker
    # until it has processed and acknowledged the previous one
    channel.basic_qos(prefetch_count=1)

    channel.basic_consume(
        queue=queue_name,
        on_message_callback=process_message,
    )

    print(f"Waiting for messages on {queue_name}. To exit press CTRL+C")
    channel.start_consuming()

start_consumer('task_queue')

3. Dead-Letter Exchange

import pika

def declare_queues_with_dlq(channel):
    # Dead-letter exchange
    channel.exchange_declare(exchange='dlx', exchange_type='direct', durable=True)

    # Dead-letter queue
    channel.queue_declare(queue='task_queue.dlq', durable=True)
    channel.queue_bind(queue='task_queue.dlq', exchange='dlx', routing_key='task_queue.dlq')

    # Main queue with dead-letter arguments
    args = {
        'x-dead-letter-exchange': 'dlx',
        'x-dead-letter-routing-key': 'task_queue.dlq',
        'x-max-retries': 3,  # Custom header-based retry limit
    }
    channel.queue_declare(queue='task_queue', durable=True, arguments=args)

# Consumer with retry tracking
def process_message_with_retry(channel, method, properties, body):
    message = json.loads(body)
    headers = properties.headers or {}
    retry_count = headers.get('x-retry-count', 0)

    try:
        process_task(message)
        channel.basic_ack(delivery_tag=method.delivery_tag)

    except Exception as e:
        if retry_count < 3:
            print(f"Retry {retry_count + 1}/3 for {message['id']}")
            # Re-publish with incremented retry count
            channel.basic_publish(
                exchange='',
                routing_key=method.routing_key,
                body=body,
                properties=pika.BasicProperties(
                    delivery_mode=2,
                    headers={'x-retry-count': retry_count + 1},
                    content_type='application/json',
                ),
            )
            channel.basic_ack(delivery_tag=method.delivery_tag)
        else:
            print(f"Max retries exceeded for {message['id']}, sending to DLQ")
            # Let RabbitMQ dead-letter it
            channel.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

4. Work Queue with Multiple Workers

# Run multiple instances of the consumer — RabbitMQ distributes messages
# in round-robin with prefetch_count=1 for fair dispatch

# worker.py
import pika
import json
import sys

def start_worker(worker_id: str, queue_name: str):
    connection = get_connection()
    channel = connection.channel()
    channel.queue_declare(queue=queue_name, durable=True)
    channel.basic_qos(prefetch_count=1)

    def callback(ch, method, properties, body):
        message = json.loads(body)
        print(f"[Worker {worker_id}] Processing: {message['id']}")
        time.sleep(1)
        ch.basic_ack(delivery_tag=method.delivery_tag)

    channel.basic_consume(queue=queue_name, on_message_callback=callback)
    print(f"[Worker {worker_id}] Waiting for tasks...")
    channel.start_consuming()

worker_id = sys.argv[1] if len(sys.argv) > 1 else '1'
start_worker(worker_id, 'task_queue')

5. Topic Exchange for Routing

import pika
import json

def setup_topic_exchange():
    connection = get_connection()
    channel = connection.channel()

    # Topic exchange — routes by pattern (e.g., 'orders.*', 'logs.error')
    channel.exchange_declare(exchange='topic_logs', exchange_type='topic', durable=True)

    # Queues for different routing keys
    channel.queue_declare(queue='all_orders', durable=True)
    channel.queue_declare(queue='error_logs', durable=True)
    channel.queue_declare(queue='all_logs', durable=True)

    channel.queue_bind(queue='all_orders', exchange='topic_logs', routing_key='order.#')
    channel.queue_bind(queue='error_logs', exchange='topic_logs', routing_key='log.error')
    channel.queue_bind(queue='all_logs', exchange='topic_logs', routing_key='log.#')

    return channel

def publish_topic(routing_key: str, message: dict):
    channel = setup_topic_exchange()
    channel.basic_publish(
        exchange='topic_logs',
        routing_key=routing_key,
        body=json.dumps(message),
        properties=pika.BasicProperties(delivery_mode=2, content_type='application/json'),
    )

# Routes to 'all_orders' queue
publish_topic('order.created', {'orderId': '123'})
publish_topic('order.cancelled', {'orderId': '456'})

# Routes to 'error_logs' AND 'all_logs' queues
publish_topic('log.error', {'service': 'api', 'msg': 'timeout'})

6. Connection Recovery

import pika
import json
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ResilientConsumer:
    def __init__(self, queue_name: str, host: str = 'localhost'):
        self.queue_name = queue_name
        self.host = host
        self._connection = None
        self._channel = None

    def connect(self):
        while True:
            try:
                self._connection = pika.BlockingConnection(
                    pika.ConnectionParameters(
                        host=self.host,
                        heartbeat=30,
                        blocked_connection_timeout=7200,
                    )
                )
                self._channel = self._connection.channel()
                self._channel.queue_declare(queue=self.queue_name, durable=True)
                self._channel.basic_qos(prefetch_count=1)
                self._channel.basic_consume(
                    queue=self.queue_name,
                    on_message_callback=self.on_message,
                )
                logger.info("Connected and consuming...")
                self._channel.start_consuming()
                break
            except pika.exceptions.AMQPConnectionError:
                logger.warning("Connection lost, retrying in 5s...")
                time.sleep(5)
            except Exception as e:
                logger.error(f"Unexpected error: {e}, retrying in 5s...")
                time.sleep(5)

    def on_message(self, channel, method, properties, body):
        try:
            message = json.loads(body)
            self.process(message)
            channel.basic_ack(delivery_tag=method.delivery_tag)
        except Exception as e:
            logger.error(f"Processing failed: {e}")
            channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

    def process(self, message: dict):
        logger.info(f"Processing: {message.get('id')}")
        # Business logic here

consumer = ResilientConsumer('task_queue')
consumer.connect()

How It Works

1. Queues: Messages are placed on a queue. Consumers subscribe to the queue. RabbitMQ delivers messages in FIFO order. Durable queues persist to disk, surviving broker restarts.
2. Acknowledgments: With manual ack, a message is only removed from the queue after the consumer acknowledges it. If the consumer dies (connection lost, crash), RabbitMQ re-delivers the message to another consumer.
3. Prefetch (QoS): basic_qos(prefetch_count=1) tells RabbitMQ not to send a new message to a worker until it has acknowledged the previous one. This prevents one slow worker from hoarding messages.
4. Dead-letter exchange: When a message is rejected with requeue=False or expires, RabbitMQ routes it to the dead-letter exchange. The DLQ holds failed messages for inspection or replay.
5. Exchanges: Direct exchanges route by exact routing key. Topic exchanges route by pattern (* = one word, # = zero or more words). Fanout exchanges broadcast to all bound queues.

Variants

Fanout Exchange (Broadcast)

channel.exchange_declare(exchange='broadcast', exchange_type='fanout', durable=True)
# All queues bound to this exchange receive every message
channel.queue_bind(queue='worker1_queue', exchange='broadcast', routing_key='')
channel.queue_bind(queue='worker2_queue', exchange='broadcast', routing_key='')

Priority Queue

# Declare queue with max priority
channel.queue_declare(
    queue='priority_tasks',
    durable=True,
    arguments={'x-max-priority': 10},
)

# Publish with priority
channel.basic_publish(
    exchange='',
    routing_key='priority_tasks',
    body=json.dumps(message),
    properties=pika.BasicProperties(
        delivery_mode=2,
        priority=5,  # Higher number = higher priority
    ),
)

Delayed Messages (TTL + DLQ)

# Declare a queue with TTL — messages expire after N ms and go to DLQ
channel.queue_declare(
    queue='delayed_tasks',
    durable=True,
    arguments={
        'x-message-ttl': 60000,  # 60 seconds
        'x-dead-letter-exchange': '',
        'x-dead-letter-routing-key': 'task_queue',
    },
)

# Publish to the delayed queue — after 60s, the message
# expires and is routed to 'task_queue' for processing
channel.basic_publish(
    exchange='',
    routing_key='delayed_tasks',
    body=json.dumps(message),
    properties=pika.BasicProperties(delivery_mode=2),
)

Best Practices

• For a deeper guide, see Configure Dead-Letter Queues in RabbitMQ for Failed Messages.

• Always use durable queues and persistent messages: Without durability, messages are lost on broker restart. Set delivery_mode=2 on every message.

• Use manual acknowledgments: Auto-ack removes messages before processing completes. If the consumer crashes, the message is lost. Always use manual ack.

• Set prefetch_count=1 for fair dispatch: Without prefetch, RabbitMQ dispatches all messages to the first available consumer. Prefetch=1 ensures round-robin distribution.

• Use dead-letter queues: Don’t requeue failed messages indefinitely — they become poison pills. Use a retry counter and send to DLQ after max retries.

• Set heartbeat: Without heartbeat, dead connections aren’t detected for hours. Set heartbeat=30 to detect network failures quickly.

• Handle connection recovery: Network failures happen. Wrap the consumer in a reconnect loop that retries with backoff.

Common Mistakes

• Using auto-ack: auto_ack=True removes the message before processing. If the consumer crashes, the message is lost. Always use manual ack.
• Not setting prefetch: Without basic_qos, one consumer can receive all messages while others sit idle. Set prefetch_count=1 for fair dispatch.
• Requeuing poison pills: A message that always fails gets requeued and reprocessed forever. Use a retry counter and dead-letter after max attempts.
• Not declaring queues on both sides: Both producer and consumer must declare the queue with the same parameters (durable, arguments). Mismatched declarations cause errors.
• Blocking the callback: Long-running tasks in the callback block the heartbeat, causing RabbitMQ to close the connection. Offload heavy work to a thread pool.