Python Pika RabbitMQ Producer Consumer

To build a producer and consumer with Python Pika and RabbitMQ, you need a broker connection, a declared queue, a publishing call, and a consuming callback that acknowledges each message. This article walks through a minimal producer and consumer, covering connection setup, queue declaration, publishing, consuming, and acknowledgment behavior.

Installing pika and Running RabbitMQ

Install the pika client with pip:

pip install pika

RabbitMQ itself runs as a separate service. On a local machine, you can start it with the official Docker image:

docker run -d --name rabbitmq -p 5672:5672 rabbitmq:3

The default guest user works for local development, but in production you should create a dedicated user with a restricted set of permissions.

Connecting to RabbitMQ with pika

The pika client uses a connection object that wraps the AMQP protocol. The most common approach for a simple script is BlockingConnection, which keeps the calling thread blocked while it waits for responses from the broker.

import pika

params = pika.URLParameters("amqp://guest:guest@localhost:5672/%2F")
connection = pika.BlockingConnection(params)
channel = connection.channel()

The URLParameters object parses the AMQP URI and passes the host, port, credentials, and virtual host to the connection. The virtual host is encoded as %2F for the default /.

Declaring a Queue and Publishing Messages

Before publishing, you need a queue to hold the messages. The queue_declare call creates the queue if it does not exist, or verifies that it does.

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

The durable flag tells RabbitMQ to persist the queue definition across broker restarts. It does not by itself make messages durable; that is controlled at publish time.

To publish a message, use basic_publish with an exchange and a routing key. For the default direct exchange (""), messages are routed to a queue whose name matches the routing key.

channel.basic_publish(
    exchange="",
    routing_key="task_queue",
    body="Hello, worker",
    properties=pika.BasicProperties(delivery_mode=2)
)

delivery_mode=2 marks the message as persistent, so RabbitMQ writes it to disk before acknowledging receipt. This trades some throughput for durability.

After publishing, close the connection to free the socket:

connection.close()

Consuming Messages with a Callback

A consumer registers a callback function that pika invokes for each delivered message. The callback receives the channel, method frame, properties, and body.

def on_message(channel, method, properties, body):
    print(f"Received: {body.decode()}")
    channel.basic_ack(delivery_tag=method.delivery_tag)

The basic_ack call confirms that the message has been processed. Without it, RabbitMQ will redeliver the message when the channel closes or the connection drops.

Set up the consumer and start the event loop:

channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue="task_queue", on_message_callback=on_message)
channel.start_consuming()

basic_qos(prefetch_count=1) tells RabbitMQ not to send more than one message to this consumer at a time, which spreads work evenly across multiple consumers.

How Acknowledgments Affect Delivery

The examples use manual acknowledgment: auto_ack is left at its default False, so the callback is responsible for telling RabbitMQ when a message has been handled.

• basic_ack confirms successful processing.
• basic_nack rejects one or more messages and can requeue them.
• basic_reject rejects a single message and can requeue that one; use basic_nack when you need to reject multiple messages.

If your consumer crashes before acking, RabbitMQ redelivers the message to another consumer. This at-least-once delivery means your handler should be idempotent.

Managing Connection Lifecycle and Reconnection

BlockingConnection is convenient for short-lived scripts, but it does not automatically reconnect. If the broker restarts, the connection object becomes unusable. For a long-running consumer, you need a reconnection loop.

A common pattern is to wrap the connection setup in a while loop that retries after a delay:

import time

while True:
    try:
        connection = pika.BlockingConnection(params)
        channel = connection.channel()
        channel.queue_declare(queue="task_queue", durable=True)
        channel.basic_qos(prefetch_count=1)
        channel.basic_consume(queue="task_queue", on_message_callback=on_message)
        channel.start_consuming()
    except pika.exceptions.AMQPConnectionError:
        time.sleep(5)

The SelectConnection adapter offers an asynchronous event loop, which is useful when you need to integrate with other I/O, but it requires a more complex state machine and is not necessary for most producer/consumer scripts.

Tuning Prefetch and Throughput

The prefetch_count value controls how many unacknowledged messages the broker sends to a consumer. A value of 1 is safest for long-running tasks because it prevents one consumer from hoarding work. Higher values increase throughput when message processing is fast and the worker can handle multiple in-flight messages.

There is no universal number; the right setting depends on the average processing time, memory usage, and the number of consumers. Start with 1 and increase only after measuring how your worker behaves under load.

For a producer, the main throughput lever is to avoid per-message setup overhead. If you send many small messages, keep using one channel and publish them in a loop instead of opening a new connection or channel for each message. Use the mandatory flag only when you need unroutable messages returned to the publisher, and handle them with a basic_return callback.