Implementing Pub/Sub with RabbitMQ in Python

Hello! Welcome back to our module on Asynchronous Messaging with Message Brokers.

Introduction

In our last lesson, you successfully set up a RabbitMQ instance using Docker. You now have a running message broker, the central component of our asynchronous systems.

Today, we will move from infrastructure to application code. We'll connect to your RabbitMQ instance using a Python client and programmatically build the components for a fundamental messaging pattern: Publish/Subscribe (Pub/Sub). This pattern is essential for broadcasting information to multiple interested parties. For instance, in an FX trading system like the one you've worked on, a pub/sub mechanism could be used to broadcast real-time price updates from a market data feed to various consumers simultaneously—such as a pricing engine, a risk management dashboard, and multiple trading algorithms.

By the end of this lesson, you will be able to:

Programmatically declare a RabbitMQ exchange and queue, and bind them to implement a pub/sub pattern using a Python client library.

We will use pika, the most common Python library for interacting with RabbitMQ.

---

1. The RabbitMQ Messaging Model

Before we write code, it's crucial to understand RabbitMQ's core messaging model, which is more nuanced than simply sending a message to a queue.

In RabbitMQ, a producer (your application that sends messages) does not send messages directly to a queue. Instead, it sends messages to an exchange.

• Exchange: An exchange receives messages from producers and is responsible for routing them to queues. Think of it as a mail sorting facility.
• Queue: A queue is a buffer that stores messages. It's the "mailbox" where messages wait to be picked up by a consumer.
• Binding: A binding is a rule that connects an exchange to a queue. It tells the exchange which queues it should route messages to.

The logic for how an exchange routes messages is determined by its type. The main types are direct, topic, headers, and fanout. For the pub/sub pattern, we are interested in the fanout exchange.

The official RabbitMQ documentation provides an excellent explanation of these concepts.

As you've read, a fanout exchange is the simplest type. It broadcasts all the messages it receives to all the queues it knows about, completely ignoring any routing keys. This is exactly what we need for a pub/sub system where one message should be delivered to every subscriber.

---

2. Implementing the Pub/Sub Pattern in Python

To implement this pattern, we need two main components:

1. A publisher that connects to RabbitMQ, declares a fanout exchange, and sends messages to it.
2. A subscriber that connects, declares its own unique queue, binds that queue to the fanout exchange, and waits for messages.

The following video provides a practical walkthrough of the code for both the publisher (producer) and the subscriber (consumer).

Let's break down the key steps and the corresponding pika code.

The Publisher (publisher.py)

The publisher's job is to send a message to a named exchange.

1. Connect to RabbitMQ: Establish a connection and open a channel.

   import pika
   connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
   channel = connection.channel()
   

2. Declare the Exchange: We declare a fanout exchange. If the exchange doesn't exist, it will be created. If it already exists, this command does nothing. This makes the operation idempotent.

   channel.exchange_declare(exchange='market_data', exchange_type='fanout')
   

   I've named our exchange market_data to fit our financial example.

3. Publish the Message: We publish the message to our named exchange. For a fanout exchange, the routing_key is ignored, so we can leave it as an empty string.

   message = "EUR/USD: 1.0855"
   channel.basic_publish(exchange='market_data', routing_key='', body=message)
   

The Subscriber (subscriber.py)

The subscriber's setup is more involved. Each subscriber needs its own private queue to receive a copy of the broadcasted message.

1. Connect and Declare the Exchange: Same as the publisher, the subscriber should also declare the exchange to ensure it exists before trying to bind to it.

   import pika
   connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
   channel = connection.channel()
   channel.exchange_declare(exchange='market_data', exchange_type='fanout')
   

2. Declare a Temporary Queue: We need a unique queue for each subscriber instance. We achieve this by asking RabbitMQ to create a queue with a random name.

   result = channel.queue_declare(queue='', exclusive=True)
   queue_name = result.method.queue
   

   • queue='': Tells RabbitMQ to create a non-durable queue with a generated name.
   • exclusive=True: Ensures that once the consumer's connection is closed, the queue is automatically deleted. This is perfect for temporary subscribers.
   • result.method.queue gives us the randomly generated queue name.

3. Bind the Queue to the Exchange: This is the crucial step that connects our subscriber's queue to the broadcast.

   channel.queue_bind(exchange='market_data', queue=queue_name)
   

   Now, any message sent to the market_data exchange will be routed to our new queue.

4. Consume Messages: Finally, we register a callback function to process messages from our queue.

   def callback(ch, method, properties, body):
       print(f" [x] Received {body.decode()}")
   
   channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
   channel.start_consuming()
   

---

3. Your Turn: Practical Implementation

Now, let's put this all together. Create a new directory for this lesson's code. Inside it, create two files: publisher.py and subscriber.py.

publisher.py

#!/usr/bin/env python
import pika
import sys

# Establish connection to RabbitMQ server
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# Declare a fanout exchange named 'market_data'
channel.exchange_declare(exchange='market_data', exchange_type='fanout')

# Construct the message from command-line arguments or use a default
message = ' '.join(sys.argv[1:]) or "EUR/USD: 1.0855"

# Publish the message to the 'market_data' exchange
channel.basic_publish(exchange='market_data', routing_key='', body=message)

print(f" [x] Sent '{message}'")
connection.close()

subscriber.py

#!/usr/bin/env python
import pika

# Establish connection to RabbitMQ server
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# Declare the same fanout exchange
channel.exchange_declare(exchange='market_data', exchange_type='fanout')

# Declare an exclusive queue with a random name
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue

# Bind the queue to the 'market_data' exchange
channel.queue_bind(exchange='market_data', queue=queue_name)

print(' [*] Waiting for market data. To exit press CTRL+C')

def callback(ch, method, properties, body):
    print(f" [x] Received price update: {body.decode()}")

# Start consuming messages from the queue
channel.basic_consume(
    queue=queue_name, on_message_callback=callback, auto_ack=True)

channel.start_consuming()

Running the Code

1. Open two separate terminal windows. In each, navigate to your project directory and start a subscriber instance:

   # Terminal 1
   python subscriber.py
   
   # Terminal 2
   python subscriber.py
   

   You should see [*] Waiting for market data. in both terminals.

2. Open a third terminal and run the publisher. You can send a custom message from the command line.

   # Terminal 3
   python publisher.py "GBP/USD: 1.2730"
   

You should immediately see the message [x] Received price update: GBP/USD: 1.2730 appear in both subscriber terminals. You have successfully implemented a pub/sub system!

Verification with the Management UI

Open your browser and navigate to the RabbitMQ Management UI at http://localhost:15672 (user: guest, pass: guest).

• Click on the Exchanges tab. You will see your market_data exchange of type fanout.
• Click on the Queues tab. You will see two queues with random-looking names (e.g., amq.gen-...). These are the exclusive queues created by your two subscribers.
• Click on the market_data exchange name. You will see the bindings that link the exchange to your two queues.

This visual confirmation is a powerful way to solidify your understanding of the underlying topology you've just created with code.

---

Conclusion

In this lesson, you've bridged the gap between having a running message broker and using it to implement a core distributed systems pattern. You've learned how to use the pika library to programmatically define the necessary RabbitMQ components for a publish/subscribe workflow.

Key Takeaways:

• Producers publish messages to exchanges, not directly to queues.
• A fanout exchange is used for the pub/sub pattern, as it broadcasts messages to all queues bound to it.
• A binding is the link that tells an exchange where to send messages.
• For pub/sub, each subscriber creates its own temporary, exclusive queue to ensure it receives a copy of every message.
• The entire topology—exchanges, queues, and bindings—can be declared programmatically, making your application setup reproducible and self-contained.

Preview of the Next Lesson:

Now that you have successfully implemented a basic pub/sub system, the next lesson, "Implement publishers and subscribers in Python that communicate via the configured RabbitMQ topology," will build upon this foundation. We will delve deeper into the implementation details, exploring how to structure the code for more robust and maintainable applications.