Publish and Consume RabbitMQ with pika

RabbitMQ is a reliable message broker. It lets services talk to each other without waiting on each other.

pika is the official Python client for RabbitMQ. It implements the AMQP 0-9-1 protocol cleanly and predictably.

This guide shows you how to publish and consume messages with pika. It covers exchanges, queues, acknowledgements, retries, and error handling.

Prerequisites

You need Python 3.7 or newer and a running RabbitMQ broker. A local instance is perfect for learning.

Install the client inside a virtual environment. If you have not done this yet, follow a guide to install pika in Python first.

Confirm the broker is reachable on port 5672. The management UI usually runs on port 15672.

Understand the Core Concepts

A producer publishes messages. A consumer receives them.

A queue stores messages until a consumer reads them. Queues are the buffer between services.

An exchange routes messages to queues. The default exchange routes by queue name directly.

A channel is a lightweight connection inside a connection. Most AMQP operations happen on a channel.

Start simple. Use the default exchange and one durable queue.

Create a Connection

The pika.BlockingConnection class opens a synchronous connection. Use it for scripts and workers.

Pass a host and port with pika.ConnectionParameters. Keep one connection per process and reuse it.

import pika

# Connect to a local RabbitMQ broker
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host="localhost", port=5672)
)

# Open a channel for AMQP operations
channel = connection.channel()

print("Connected to RabbitMQ.")
connection.close()

Connected to RabbitMQ.

Always close the connection when done. A context manager handles this automatically.

Publish Messages to a Queue

Publishing needs a queue, a routing key, and a message body.

Declare the queue with channel.queue_declare. Set durable=True so the queue survives broker restarts.

Publish with channel.basic_publish. Use an empty exchange and the queue name as the routing key.

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()

# Create a durable queue
channel.queue_declare(queue="orders", durable=True)

# Publish a persistent message
channel.basic_publish(
    exchange="",
    routing_key="orders",
    body="order:12345",
    properties=pika.BasicProperties(delivery_mode=2),
)

print("Order message published.")
connection.close()

Order message published.

The delivery_mode=2 value marks the message as persistent. This protects data if the broker restarts.

Declare the queue before publishing. A missing queue silently drops messages.

Publish to a Direct Exchange

Direct exchanges route messages by an exact routing key. This pattern fits command routing and task types.

Declare the exchange with channel.exchange_declare. Then bind a queue to it.

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()

# Declare a direct exchange and bind a queue
channel.exchange_declare(exchange="logs", exchange_type="direct", durable=True)
channel.queue_declare(queue="error_logs", durable=True)
channel.queue_bind(exchange="logs", queue="error_logs", routing_key="error")

# Publish to the exchange with a routing key
channel.basic_publish(
    exchange="logs",
    routing_key="error",
    body="Database timeout",
    properties=pika.BasicProperties(delivery_mode=2),
)

print("Message routed to error_logs.")
connection.close()

Message routed to error_logs.

Messages published with other routing keys are ignored by this queue. That is the point of direct routing.

Consume Messages

Consuming requires a callback. pika calls it for each delivered message.

Register the callback with channel.basic_consume. Then block with channel.start_consuming.

import pika

def handle_order(channel, method, properties, body):
    print(f"Processing: {body.decode()}")
    # Acknowledge after successful processing
    channel.basic_ack(delivery_tag=method.delivery_tag)

connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()

channel.queue_declare(queue="orders", durable=True)
channel.basic_consume(
    queue="orders",
    on_message_callback=handle_order,
    auto_ack=False,
)

print("Waiting for orders. Press CTRL+C to stop.")
channel.start_consuming()

Waiting for orders. Press CTRL+C to stop.
Processing: order:12345

The channel.basic_ack method confirms the message was handled. Without it, RabbitMQ redelivers the message.

Acknowledge only after your work succeeds. A crash before acknowledgement triggers a safe redelivery.

Use Prefetch to Control Load

Consumers should not grab too many messages at once. A large backlog can overwhelm a single worker.

The channel.basic_qos method limits unacknowledged messages per consumer.

channel.basic_qos(prefetch_count=1)
channel.basic_consume(
    queue="orders",
    on_message_callback=handle_order,
    auto_ack=False,
)

A prefetch of 1 gives fair dispatch across multiple consumers. Tune it based on message processing time.

Handle Failures and Retries

Messages can fail. A bad payload, a downstream timeout, or a missing record will happen.

On failure, reject with channel.basic_nack. Set requeue=False to send the message to a dead-letter queue.

import json
import pika

def handle_order(channel, method, properties, body):
    try:
        payload = json.loads(body)
        process_order(payload)
        channel.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as exc:
        print(f"Failed to process message: {exc}")
        # Reject without requeue to avoid an infinite loop
        channel.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

A dead-letter queue captures failed messages for later inspection. This keeps your main queue healthy.

Wrap the connection in a retry loop. Catch pika.exceptions.AMQPConnectionError and reconnect with a delay.

import time
import pika
from pika.exceptions import AMQPConnectionError

while True:
    try:
        connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
        channel = connection.channel()
        channel.queue_declare(queue="orders", durable=True)
        channel.basic_qos(prefetch_count=1)
        channel.basic_consume(queue="orders", on_message_callback=handle_order, auto_ack=False)
        channel.start_consuming()
    except AMQPConnectionError:
        print("Broker unreachable. Retrying in 5 seconds.")
        time.sleep(5)

Broker unreachable. Retrying in 5 seconds.

If you also persist message results to a database, you can install mysqlclient in Python and write processed orders from the same consumer.

For analytics or reporting that reads from another system, you can query Oracle Database from Python inside a worker, as long as the callback stays fast.

Close Connections Cleanly

Long-running consumers should shut down gracefully. The channel.stop_consuming method exits the blocking loop.

Then close the connection with connection.close. This releases broker resources cleanly.

try:
    channel.start_consuming()
except KeyboardInterrupt:
    print("Shutting down consumer.")
    channel.stop_consuming()
finally:
    connection.close()
    print("Connection closed.")

Shutting down consumer.
Connection closed.

A clean shutdown avoids redelivery storms and stuck connections.

Best Practices for Reliable Messaging

Use durable queues and persistent messages for important work. This protects data during broker restarts.

Set auto_ack=False and acknowledge only after success. Automatic acknowledgement can lose messages on failure.

Set a prefetch limit. It protects your consumers from message floods.

Use dead-letter queues for poison messages. Infinite requeues block progress.

Keep callbacks short. Offload heavy work to dedicated workers or background threads.

Reuse one connection per process. Opening a connection per message wastes broker resources.

Monitor broker health and consumer lag. A quiet queue often means a broken consumer.

Test failure paths locally. Stop the broker and confirm the retry loop recovers.

Pin the pika version in your dependency file. Reproducible deployments avoid surprise behavior changes.

Conclusion

Publishing and consuming RabbitMQ messages with pika is simple once the patterns are clear. Use channel.basic_publish to send and channel.basic_consume to receive.

Use durable queues and persistent messages for reliability. Acknowledge after success and reject failures to a dead-letter queue.

Add retries for connection failures and shut down consumers cleanly.

With these patterns, your Python services can exchange messages reliably and stay resilient under real production load.