RabbitMQ Client with pika

RabbitMQ is an open-source message broker that implements the Advanced Message Queuing Protocol (AMQP). It acts as an intermediary for messages, allowing different applications (producers and consumers) to communicate asynchronously and reliably. This decoupling enhances system resilience, scalability, and maintainability. Producers send messages to RabbitMQ, which then routes them to queues, and consumers retrieve messages from these queues.

pika is the official Python client library for RabbitMQ. It provides a comprehensive interface for Python applications to interact with RabbitMQ brokers. pika supports both synchronous (blocking) and asynchronous (non-blocking) communication patterns, catering to various application architectures.

Key concepts when working with RabbitMQ and pika:
- Producer: An application that creates and sends messages to RabbitMQ.
- Consumer: An application that receives and processes messages from RabbitMQ.
- Queue: A named buffer within RabbitMQ where messages are stored until they are consumed.
- Exchange: Receives messages from producers and routes them to one or more queues based on predefined rules (e.g., direct, topic, fanout).
- Binding: A relationship between an exchange and a queue, often defined by a routing key, which determines how messages are routed.
- Connection: A TCP connection between your pika-enabled Python application and the RabbitMQ broker.
- Channel: A lightweight virtual connection within a single AMQP connection. All AMQP commands are issued over a channel, allowing multiple concurrent operations over a single connection.

The `pika.BlockingConnection` is the simplest way to interact with RabbitMQ for synchronous operations. It blocks the execution of your program until a given operation completes. This is suitable for straightforward scripts or applications where complex asynchronous event loops are not required.

Example Code

 Producer (send.py)
import pika
import sys

 Establish a connection to RabbitMQ server
 Default host is 'localhost'. Change if RabbitMQ is on a different machine.
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

 Create a channel
channel = connection.channel()

 Declare a queue named 'hello'. This is idempotent; it will only be created if it doesn't exist.
channel.queue_declare(queue='hello')

 Get message from command line arguments or use a default
message = ' '.join(sys.argv[1:]) or "Hello World!"

 Publish the message to the 'hello' queue
 exchange='' means we are using the default exchange
 routing_key='hello' specifies the queue to which the message should be delivered
channel.basic_publish(exchange='',
                      routing_key='hello',
                      body=message.encode())
print(f" [x] Sent '{message}'")

 Close the connection
connection.close()


 Consumer (receive.py)
import pika
import time

 Establish a connection to RabbitMQ server
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

 Create a channel
channel = connection.channel()

 Declare the same queue named 'hello'. Ensures the queue exists.
channel.queue_declare(queue='hello')

print(' [-] Waiting for messages. To exit press CTRL+C')

 Define a callback function to be executed when a message is received
def callback(ch, method, properties, body):
    print(f" [x] Received {body.decode()}")
     Simulate some work
    time.sleep(body.count(b'.'))
    print(" [x] Done")
     Acknowledge the message (important for reliability)
    ch.basic_ack(delivery_tag=method.delivery_tag)

 Start consuming messages from the 'hello' queue
 on_message_callback specifies the function to call when a message arrives
 auto_ack=False means we will manually acknowledge messages after processing them
channel.basic_consume(queue='hello',
                      on_message_callback=callback,
                      auto_ack=False)  Changed to False for demonstrating manual ack

 Keep the consumer running, waiting for messages
channel.start_consuming()

 Note:
 To run these examples:
 1. Ensure RabbitMQ server is running (e.g., via Docker: docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management)
 2. Install pika: pip install pika
 3. Run the consumer first in one terminal: python receive.py
 4. Run the producer in another terminal: python send.py "Your message here..."

RabbitMQ Client with pika

Example Code

Related Topics