Python’s pika library lets your applications talk to RabbitMQ, a powerful message broker.
Let’s see pika in action. Imagine a simple producer that sends a greeting and a consumer that receives it.
Producer (producer.py):
import pika
import sys
# Establish connection
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
# Declare a queue for messages
channel.queue_declare(queue='hello')
# Message to send
message = 'Hello, RabbitMQ!'
# Publish the message
channel.basic_publish(exchange='',
routing_key='hello',
body=message)
print(f" [x] Sent '{message}'")
# Close connection
connection.close()
Consumer (consumer.py):
import pika
import sys
# Establish connection
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
# Ensure the queue exists (idempotent)
channel.queue_declare(queue='hello')
# Define a callback function to process messages
def callback(ch, method, properties, body):
print(f" [x] Received {body.decode()}")
# Set up the consumer to listen to the 'hello' queue
channel.basic_consume(queue='hello',
on_message_callback=callback,
auto_ack=True) # Acknowledge message automatically
print(' [*] Waiting for messages. To exit press CTRL+C')
# Start consuming messages
channel.start_consuming()
To run this:
- Ensure RabbitMQ is running on
localhost. - Open two terminals.
- In the first, run
python consumer.py. You’ll see[*] Waiting for messages. To exit press CTRL+C. - In the second, run
python producer.py. You’ll see[x] Sent 'Hello, RabbitMQ!'. - Switch back to the consumer terminal. You’ll see
[x] Received 'Hello, RabbitMQ!'.
This illustrates the core exchange: the producer sends a message to a named queue (hello), and the consumer listens to that same queue, processing messages as they arrive. pika handles the low-level AMQP protocol details, letting you focus on your application logic.
The power of RabbitMQ comes from its routing capabilities. While we used a direct exchange='' (the "default exchange") which routes messages directly to queues named by routing_key, RabbitMQ supports various exchange types: direct, fanout, topic, and headers. Each type dictates how messages are routed from producers to queues.
For example, with a fanout exchange, a message published to the exchange is broadcast to all queues bound to it, regardless of the routing_key. A topic exchange allows for more complex routing based on patterns in the routing_key.
The basic_publish method has several optional arguments that unlock advanced features. delivery_mode=2 marks a message as persistent. When the broker restarts, persistent messages are not lost. properties=pika.BasicProperties(delivery_mode=2) would achieve this. Without persistence, messages are stored in memory and lost if the broker crashes.
When you auto_ack=True in basic_consume, the message is removed from the queue as soon as it’s delivered to the consumer. If the consumer crashes after receiving the message but before processing it, that message is lost. For reliable processing, you should set auto_ack=False and manually acknowledge messages using channel.basic_ack(delivery_tag=method.delivery_tag) after successful processing. This ensures messages are only removed from the queue once they’ve been fully handled.
The pika.ConnectionParameters object is highly configurable. You can specify host, port (defaulting to 5672), username, password, virtual_host, and even SSL/TLS options for secure connections.
The most surprising true thing about pika is that its BlockingConnection is actually a thin wrapper around an asynchronous I/O loop. You’re not truly blocking the entire Python process; you’re blocking the current thread from doing further AMQP operations until the current one completes. This is why you can run multiple consumers in separate threads within the same Python process, each with its own BlockingConnection and channel, and they won’t interfere with each other’s I/O.
You’ll next want to explore how to handle message acknowledgments for guaranteed delivery.