Python Pika RabbitMQ Producer Consumer

python pika rabbitmq producer consumer requires a clear understanding of the core syntax, runtime behavior, and practical implementation patterns demonstrated in the examples below.

When you need to move work out of a request path or distribute tasks across workers, RabbitMQ with the pika client is a common choice in Python. This article walks through a minimal producer and consumer using pika, 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 a simple direct exchange, the default exchange ("") routes messages 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

By default, pika uses manual acknowledgment when you supply a callback and do not set auto_ack=True. Manual ack gives you control over when a message is considered processed.

• basic_ack confirms successful processing.
• basic_nack rejects a message and optionally requeues it.
• basic_reject rejects a message without the option to requeue 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 batching publishes. If you send many small messages, consider collecting them and publishing in a loop with a single channel, and use basic_publish with mandatory and immediate only when you need to detect unroutable messages.