RabbitMQ with Python (Pika) — Complete Guide
In this tutorial, you will learn about RabbitMQ with Python (Pika). We cover key concepts, practical examples, and best practices to help you master this topic.
Use Pika to integrate RabbitMQ with Python for publishing and consuming messages with exchange types, acknowledgements, and connection management.
What You Learn
You will learn how to connect to RabbitMQ with Pika, publish and consume messages, work with different exchange types, handle reconnections, and implement common patterns.
Why It Matters
Python is the most popular language for RabbitMQ applications. Pika is the officially recommended Python client. Mastering Pika allows you to build reliable messaging systems for data pipelines, task queues, and Microservices.
Real-World Use
Doda Browser's malware analysis pipeline uses Pika for all RabbitMQ communication. Python workers consume scan requests, publish results, and handle retries. The entire system runs on Pika with manual acknowledgements.
Connecting with Pika
import pika
# Basic connection
params = pika.ConnectionParameters(
host='localhost',
port=5672,
credentials=pika.PlainCredentials('guest', 'guest'),
virtual_host='/',
heartbeat=600,
blocked_connection_timeout=300
)
connection = pika.BlockingConnection(params)
channel = connection.channel()
print(f"Connected: {connection.is_open}")
print(f"Channel: {channel.is_open}")
channel.close()
connection.close()
Expected output:
Connected: True
Channel: True
Publishing Messages
import pika
import json
conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.exchange_declare(exchange='python_exchange', exchange_type='topic', durable=True)
ch.queue_declare(queue='python_queue', durable=True)
ch.queue_bind(exchange='python_exchange', queue='python_queue', routing_key='python.#')
message = {
'action': 'scan_file',
'file_path': '/tmp/suspicious.exe',
'user_id': 42
}
ch.basic_publish(
exchange='python_exchange',
routing_key='python.scan.request',
body=json.dumps(message),
properties=pika.BasicProperties(
delivery_mode=2,
content_type='application/json',
message_id='msg_001'
)
)
print(f"Published: {message}")
conn.close()
Expected output:
Published: {'action': 'scan_file', 'file_path': '/tmp/suspicious.exe', 'user_id': 42}
Consuming Messages
import pika
import json
conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue='python_queue', durable=True)
ch.basic_qos(prefetch_count=1)
def callback(ch, method, properties, body):
data = json.loads(body)
print(f"Message ID: {properties.message_id}")
print(f"Action: {data['action']}")
print(f"File: {data['file_path']}")
print(f"Processing scan...")
import time
time.sleep(0.5)
print("Scan complete")
ch.basic_ack(delivery_tag=method.delivery_tag)
ch.basic_consume(queue='python_queue', on_message_callback=callback)
print("Waiting for messages...")
ch.start_consuming()
Expected output:
Waiting for messages...
Message ID: msg_001
Action: scan_file
File: /tmp/suspicious.exe
Processing scan...
Scan complete
Robust Connection with Auto-Recovery
import pika
import logging
logging.basicConfig(level=logging.INFO)
class RabbitMQClient:
def __init__(self):
self.connection = None
self.channel = None
self.connect()
def connect(self):
params = pika.ConnectionParameters(
host='localhost',
port=5672,
credentials=pika.PlainCredentials('guest', 'guest'),
heartbeat=600,
blocked_connection_timeout=300
)
self.connection = pika.BlockingConnection(params)
self.channel = self.connection.channel()
self.channel.queue_declare(queue='robust_queue', durable=True)
logging.info("Connected to RabbitMQ")
def publish(self, message):
try:
self.channel.basic_publish(
exchange='',
routing_key='robust_queue',
body=message,
properties=pika.BasicProperties(delivery_mode=2)
)
logging.info(f"Published: {message}")
except (pika.exceptions.ConnectionClosed,
pika.exceptions.ChannelClosed) as e:
logging.error(f"Connection lost: {e}")
self.connect()
self.publish(message)
def close(self):
if self.connection and self.connection.is_open:
self.connection.close()
client = RabbitMQClient()
client.publish("message 1")
client.publish("message 2")
client.close()
Expected output:
INFO:root:Connected to RabbitMQ
INFO:root:Published: message 1
INFO:root:Published: message 2
Common Patterns: RPC with Pika
import pika
import uuid
import json
class RPCClient:
def __init__(self):
self.conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
self.ch = self.conn.channel()
result = self.ch.queue_declare(queue='', exclusive=True)
self.callback_queue = result.method.queue
self.ch.basic_consume(
queue=self.callback_queue,
on_message_callback=self.on_response,
auto_ack=True
)
self.response = None
self.corr_id = None
def on_response(self, ch, method, props, body):
if self.corr_id == props.correlation_id:
self.response = body
def call(self, request):
self.corr_id = str(uuid.uuid4())
self.ch.basic_publish(
exchange='',
routing_key='rpc_queue',
properties=pika.BasicProperties(
reply_to=self.callback_queue,
correlation_id=self.corr_id
),
body=json.dumps(request)
)
while self.response is None:
self.conn.process_data_events()
return json.loads(self.response)
rpc = RPCClient()
result = rpc.call({'operation': 'scan', 'file': 'test.exe'})
print(f"RPC result: {result}")
Expected output:
RPC result: {'status': 'completed', 'threat_level': 'low'}
Common Mistakes
1. Not Handling Connection Errors
Network failures happen. Without reconnection logic, the application crashes when RabbitMQ restarts. Use connection parameters with heartbeat and implement retry.
2. Creating a New Connection Per Message
Connection creation is expensive. Reuse connections and channels. One connection per Process with one channel per thread is the standard pattern.
3. Not Setting prefetch_count
Without QoS prefetch, Pika consumers receive all messages at once. Memory grows unbounded. Always set basic_qos(prefetch_count=1).
4. Using BlockingConnection in Async Code
BlockingConnection blocks the event loop. Use pika.adapters.asyncio_connection.AsyncioConnection or aio_pika for async Python applications.
5. Forgetting to Close Connections
Unclosed connections eventually exhaust RabbitMQ connection limits. Use context managers or try/finally blocks to ensure cleanup.
Practice Questions
1. What is Pika?
Pika is the official Python client library for RabbitMQ. It supports both blocking and asynchronous connection adapters.
2. How do you set up a consumer with manual acknowledgements?
Set auto_ack=False in basic_consume, then call basic_ack(delivery_tag=tag) in the callback after processing.
3. What does channel.basic_qos do?
It limits the number of unacknowledged messages delivered to a consumer. prefetch_count=1 means one message at a time.
4. How do you implement reconnection in Pika?
Catch ConnectionClosed/ChannelClosed exceptions, then recreate the connection and channel. Use a retry loop with exponential backoff.
Challenge
Build a Python application using Pika that implements a priority task queue. Messages with higher priority should be consumed first. Use queue arguments and multiple queues with topic exchange routing.
FAQ
Mini Project: Task Producer and Worker
import pika
import json
import time
import threading
import random
# Producer
def producer():
conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue='python_tasks', durable=True)
tasks = [
{'type': 'scan', 'target': 'file.exe'},
{'type': 'compress', 'target': 'data.zip'},
{'type': 'analyze', 'target': 'log.txt'},
{'type': 'scan', 'target': 'document.pdf'},
{'type': 'compress', 'target': 'photos.tar'},
]
for task in tasks:
ch.basic_publish(
exchange='',
routing_key='python_tasks',
body=json.dumps(task),
properties=pika.BasicProperties(
delivery_mode=2,
content_type='application/json'
)
)
print(f"Produced: {task['type']} - {task['target']}")
time.sleep(0.5)
conn.close()
# Worker
def worker(name):
conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue='python_tasks', durable=True)
ch.basic_qos(prefetch_count=1)
def callback(ch, method, properties, body):
task = json.loads(body)
print(f"[{name}] Processing {task['type']} on {task['target']}")
time.sleep(random.uniform(0.5, 2.0))
ch.basic_ack(delivery_tag=method.delivery_tag)
print(f"[{name}] Completed {task['type']}")
ch.basic_consume(queue='python_tasks', on_message_callback=callback)
ch.start_consuming()
# Start workers and producer
workers = []
for i in range(2):
t = threading.Thread(target=worker, args=(f'W{i}',), daemon=True)
t.start()
workers.append(t)
time.sleep(0.5)
producer()
time.sleep(3)
Expected output:
[W0] Processing scan on file.exe
[W1] Processing compress on data.zip
[W0] Completed scan
[W0] Processing analyze on log.txt
[W1] Completed compress
[W1] Processing scan on document.pdf
[W0] Completed analyze
[W0] Processing compress on photos.tar
[W1] Completed scan
[W0] Completed compress
What's Next
Now that you understand RabbitMQ with Python, explore RabbitMQ with Node.js for JavaScript applications, then learn about RabbitMQ security for production hardening.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro