Intermediate
Your web app needs to send a thousand email notifications, resize uploaded images, and kick off reports — all without making the user wait. If you are handling this with threads or background tasks inside the same process, you are one busy period away from a frozen server. Message queues solve this by decoupling the code that produces work from the code that does the work. RabbitMQ is the most widely deployed open-source message broker, and pika is Python’s official client for it.
Install pika with pip install pika. You also need a running RabbitMQ instance — the fastest way to get one locally is docker run -d -p 5672:5672 -p 15672:15672 rabbitmq:3-management, which starts RabbitMQ with the management UI at http://localhost:15672 (login: guest/guest). Once that is running, every example in this tutorial will work as-is.
This tutorial covers the core pika workflow: connecting to RabbitMQ, declaring queues and exchanges, publishing messages, consuming them with worker processes, acknowledging delivery, handling errors, and building a real parallel task queue. By the end, you will be able to distribute work across multiple Python workers reliably.
Publishing and Consuming a Message: Quick Example
Here is the smallest possible producer-consumer pair. Two scripts — one that sends, one that receives — showing the entire lifecycle of a RabbitMQ message:
# producer.py
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello', durable=True)
channel.basic_publish(
exchange='',
routing_key='hello',
body='Hello, World!',
properties=pika.BasicProperties(delivery_mode=2) # make message persistent
)
print("[x] Sent 'Hello, World!'")
connection.close()# consumer.py
import pika
def callback(ch, method, properties, body):
print(f"[x] Received: {body.decode()}")
ch.basic_ack(delivery_tag=method.delivery_tag) # acknowledge receipt
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello', durable=True)
channel.basic_consume(queue='hello', on_message_callback=callback)
print("[*] Waiting for messages. CTRL+C to exit")
channel.start_consuming()Run producer.py, then consumer.py in a separate terminal. Output:
# producer.py output:
[x] Sent 'Hello, World!'
# consumer.py output:
[*] Waiting for messages. CTRL+C to exit
[x] Received: Hello, World!Three things to notice: queue_declare(durable=True) survives RabbitMQ restarts, delivery_mode=2 persists the message to disk before acknowledging, and basic_ack() tells the broker the message was processed successfully. Without the ack, RabbitMQ will re-queue the message and redeliver it to another consumer when the connection drops.
Core Concepts: Exchanges, Queues, and Routing
RabbitMQ’s routing model is more flexible than a simple queue. Messages are always published to an exchange, and the exchange decides which queues receive copies based on routing rules.
| Exchange Type | How Routing Works | Use Case |
|---|---|---|
| direct | Routes to queues whose binding key exactly matches the routing key | Task distribution, worker pools |
| fanout | Routes to all bound queues, ignoring routing key | Broadcast notifications, pub/sub |
| topic | Routes using wildcard patterns (* = one word, # = zero or more) | Log routing, category-based filtering |
| headers | Routes based on message header attributes | Complex multi-attribute filtering |
The empty string exchange='' in the quick example is the default direct exchange — it routes messages directly to the queue named in routing_key. For most task queue use cases, the default exchange is all you need. The other exchange types shine when you have multiple consumers that need different subsets of the same message stream.
Publishing Messages
For production use, always declare the queue as durable and messages as persistent. Here is a more complete publisher with error handling and connection retry:
# robust_publisher.py
import pika
import json
import time
def get_channel(host='localhost', retries=5):
"""Connect with retry logic for container startup delays."""
for attempt in range(retries):
try:
conn = pika.BlockingConnection(pika.ConnectionParameters(
host=host,
heartbeat=600,
blocked_connection_timeout=300,
))
return conn, conn.channel()
except pika.exceptions.AMQPConnectionError as e:
if attempt < retries - 1:
print(f"Connection failed ({e}), retrying in 2s...")
time.sleep(2)
else:
raise
connection, channel = get_channel()
# Declare a durable queue -- safe to call multiple times (idempotent)
channel.queue_declare(queue='tasks', durable=True)
# Publish 10 tasks
for i in range(10):
task = {'id': i, 'action': 'resize_image', 'path': f'/uploads/img_{i}.jpg'}
channel.basic_publish(
exchange='',
routing_key='tasks',
body=json.dumps(task),
properties=pika.BasicProperties(
delivery_mode=2, # persist to disk
content_type='application/json',
)
)
print(f"[x] Published task {i}")
connection.close()
print("[x] Done")Output:
[x] Published task 0
[x] Published task 1
...
[x] Published task 9
[x] DoneThe heartbeat=600 parameter keeps the connection alive during long operations by sending periodic heartbeat frames. Without it, the broker may close idle connections after 60 seconds -- a common source of confusing ConnectionResetError bugs in long-running services.
Consuming Messages with Workers
A worker process consumes messages one at a time, acknowledges each after successful processing, and rejects (nacks) messages that fail so they get requeued:
# worker.py
import pika
import json
import time
import random
def process_task(task):
"""Simulate work that takes a variable amount of time."""
delay = random.uniform(0.1, 0.5)
time.sleep(delay)
return {'status': 'ok', 'processed': task['id'], 'time': round(delay, 2)}
def on_message(channel, method, properties, body):
task = json.loads(body)
print(f"[>] Processing task {task['id']}: {task['action']}")
try:
result = process_task(task)
print(f"[+] Done: {result}")
channel.basic_ack(delivery_tag=method.delivery_tag)
except Exception as e:
print(f"[!] Failed: {e} -- requeuing")
channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='tasks', durable=True)
# basic_qos limits to 1 unacked message per worker -- prevents overload
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='tasks', on_message_callback=on_message)
print("[*] Worker ready. CTRL+C to exit")
try:
channel.start_consuming()
except KeyboardInterrupt:
channel.stop_consuming()
connection.close()Output (running two worker processes simultaneously):
[*] Worker ready. CTRL+C to exit
[>] Processing task 0: resize_image
[+] Done: {'status': 'ok', 'processed': 0, 'time': 0.23}
[>] Processing task 2: resize_image
[+] Done: {'status': 'ok', 'processed': 2, 'time': 0.41}basic_qos(prefetch_count=1) is critical for fair work distribution. Without it, RabbitMQ sends all pending messages to workers at once -- the first worker that connects gets everything queued, and other workers sit idle. With prefetch_count=1, each worker receives the next message only after acknowledging the previous one, spreading load evenly.
Real-Life Example: Parallel Image Resizer
Let us build a practical task queue that resizes uploaded images in parallel. A publisher queues resize jobs, and multiple worker processes consume them concurrently:
# image_queue_publisher.py
import pika
import json
import os
QUEUE = 'image_resize'
# Sample jobs -- in production these come from your web app
jobs = [
{'input': f'uploads/photo_{i}.jpg', 'width': 800, 'height': 600}
for i in range(20)
]
conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue=QUEUE, durable=True)
for job in jobs:
ch.basic_publish(
exchange='',
routing_key=QUEUE,
body=json.dumps(job),
properties=pika.BasicProperties(delivery_mode=2)
)
print(f"Queued: {job['input']}")
conn.close()
print(f"Published {len(jobs)} resize jobs")# image_queue_worker.py
import pika
import json
import time
import random
QUEUE = 'image_resize'
def resize_image(job):
"""Simulate image resize (replace with Pillow in real use)."""
time.sleep(random.uniform(0.05, 0.2))
return f"resized/{os.path.basename(job['input'])}"
import os
def on_job(ch, method, properties, body):
job = json.loads(body)
try:
output = resize_image(job)
print(f"[worker] Resized {job['input']} -> {output} ({job['width']}x{job['height']})")
ch.basic_ack(delivery_tag=method.delivery_tag)
except Exception as e:
print(f"[worker] ERROR on {job['input']}: {e}")
ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False) # dead-letter
conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue=QUEUE, durable=True)
ch.basic_qos(prefetch_count=1)
ch.basic_consume(queue=QUEUE, on_message_callback=on_job)
print(f"[worker] Listening on '{QUEUE}'...")
channel_name = conn.server_properties.get(b'cluster_name', b'rabbitmq').decode()
try:
ch.start_consuming()
except KeyboardInterrupt:
ch.stop_consuming()
conn.close()Start 3--5 worker processes in separate terminals, then run the publisher. Each worker picks up jobs as they become available. Because RabbitMQ persists both the queue and the messages, you can kill and restart workers at any time without losing jobs. This is the key advantage over in-process background threads -- the work queue survives process crashes, deployments, and restarts.
Frequently Asked Questions
What happens if a worker crashes before acknowledging a message?
RabbitMQ tracks which messages are unacknowledged. If the consumer's connection closes without sending an ack, the broker automatically requeues the message and delivers it to the next available worker. This is the core durability guarantee of the AMQP protocol. The risk is that if your worker processes a message but crashes before the ack, the message gets processed twice -- so your task handlers should be idempotent (safe to run multiple times with the same input).
How do I handle messages that keep failing?
Configure a dead-letter exchange (DLX) on the queue. When a message is nacked with requeue=False or exceeds its retry count, RabbitMQ forwards it to the DLX rather than discarding it. You can then inspect failed messages separately, alert on them, or replay them manually after fixing the bug. Set it up via queue_declare(arguments={'x-dead-letter-exchange': 'dlx_name'}).
How do I connect to a cloud RabbitMQ instance?
Use pika.URLParameters('amqps://user:password@host:5671/vhost') for TLS connections (cloud providers use port 5671 for TLS, not 5672). CloudAMQP, Amazon MQ, and other hosted services provide the full connection URL in their dashboard. For credential security, load the URL from an environment variable: pika.URLParameters(os.environ['RABBITMQ_URL']).
How do I broadcast a message to all consumers?
Declare a fanout exchange and bind each consumer's queue to it. The publisher sends to the exchange name with an empty routing key: basic_publish(exchange='notifications', routing_key='', body=msg). Each bound queue receives a copy regardless of routing key. This is the pattern for broadcasting events like "cache invalidated" or "config updated" to all running instances of a service.
Can I use pika with asyncio or threading?
Yes. For asyncio applications, use aio-pika (pip install aio-pika) which provides an async/await API over the same AMQP protocol. For threaded applications, create one channel per thread -- channels are not thread-safe but connections can be shared. Alternatively, pika.SelectConnection provides an event-driven callback API that works with existing event loops.
Conclusion
Python's pika library gives you direct access to RabbitMQ's AMQP protocol for building reliable, scalable message pipelines. In this tutorial you connected to a broker with retry logic, published persistent messages, consumed them with fair dispatch using prefetch_count=1, handled acknowledgements and nacks, and built a parallel image-resize task queue that survives crashes and restarts.
The image resizer is ready to extend -- add a dead-letter queue for failed jobs, implement an exponential backoff retry policy, or wire the publisher into a Flask endpoint so web requests enqueue jobs instead of processing them synchronously. Each of these extensions follows the same publish/consume/ack pattern you learned here.
Official documentation: pika -- AMQP 0-9-1 client library.