RabbitMQ Pub/Sub in Python

Hello! Welcome to the third lesson in our module on Asynchronous Messaging with Message Brokers.

Introduction

In our last lesson, we built our first distributed application using RabbitMQ. You learned how to programmatically declare a fanout exchange, create temporary queues, and bind them together to implement the publish/subscribe pattern. This allowed you to broadcast a single message to multiple, independent subscribers—a foundational pattern for systems that need to distribute information widely, like a market data feed.

Today, we will build on that foundation to implement more sophisticated communication. While broadcasting is useful, many real-world scenarios require more targeted message delivery. For instance, you might want to send trade execution reports only to the settlement system, while risk calculation updates go only to the analytics engine.

This lesson will teach you how to implement publishers and subscribers that communicate using more advanced routing topologies. Specifically, we will cover:

• Direct exchanges for routing messages based on an exact-match routing key.
• Topic exchanges for routing messages based on flexible, wildcard-based patterns.

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

Implement publishers and subscribers in Python that communicate via the configured RabbitMQ topology.

This will involve writing Python code that not only publishes and consumes messages but does so in a way that leverages RabbitMQ's powerful routing capabilities to deliver the right messages to the right services.

---

1. Beyond Broadcasting: Selective Routing

The fanout exchange from our previous lesson is powerful but indiscriminate; it sends every message to every bound queue. To build more complex systems, we need more control. This is where direct and topic exchanges come in. They use a routing_key provided by the publisher to decide which queues should receive the message.

• A publisher sends a message to an exchange with a specific routing_key.
• A consumer binds its queue to that exchange with a binding_key.
• The exchange routes the message to queues where the routing_key matches the binding_key.

How the "match" is determined depends on the exchange type.

Direct Exchanges: One-to-One Routing

A direct exchange delivers messages to queues whose binding_key exactly matches the message's routing_key.

This is ideal for unicast routing of tasks. For example, in a payment system, you might publish a message with the routing key payment.gbp to ensure it is handled by the consumer responsible for processing British Pound payments, and not by the one handling payment.usd.

The following video demonstrates how to implement this pattern.

As you saw, the key changes from our previous fanout implementation are:

• Publisher: The call to basic_publish now includes a meaningful routing_key.

  channel.basic_publish(exchange='routing_exchange',
                        routing_key='payments_only', # This key matters now
                        body=message)
  

• Subscriber: The call to queue_bind specifies the routing_key it is interested in.

  channel.queue_bind(exchange='routing_exchange',
                     queue=queue_name,
                     routing_key='payments_only') # This is the binding key
  

• Exchange Declaration: The exchange_type is set to 'direct'.

---

2. Flexible Subscriptions with Topic Exchanges

Topic exchanges take routing a step further by allowing pattern-based matching on the routing key. This is extremely powerful for multicast scenarios where consumers may be interested in related, but not identical, categories of messages.

Routing keys in topic exchanges are typically a list of words delimited by dots (e.g., trades.fx.eurusd). The binding keys can then use wildcards:

• * (star) can substitute for exactly one word.
• # (hash) can substitute for zero or more words.

Example Scenario:
Imagine a system publishing financial events with routing keys like country.asset_class.instrument.

• A message with routing key uk.fx.gbpusd would be matched by:
  • A binding key uk.fx.gbpusd (exact match).
  • A binding key uk.fx.* (any FX instrument in the UK).
  • A binding key *.fx.* (any FX instrument in any country).
  • A binding key uk.# (all events originating from the UK).

This allows for a very flexible and decoupled architecture. A publisher can emit highly specific events, and various consumers can subscribe to broad or narrow streams of data as needed, without the publisher needing to know who is listening.

The same video you just watched also has an excellent section on implementing topic exchanges.

---

3. Your Turn: Implementing a Financial Event Router

Now it's time to apply these concepts. You will build a small system that simulates the distribution of financial events using a topic exchange.

You will create:

• One publisher that emits different types of events.
• Three distinct subscribers, each interested in a different subset of those events.

Setup

1. Create a new directory for this project.
2. Make sure your Docker container for RabbitMQ is running.
3. Install the pika library if you haven't already: pip install pika.

The Publisher (publisher.py)

Create a file named publisher.py and add the following code. This script will publish three distinct messages to a topic exchange named financial_events.

#!/usr/bin/env python
import pika

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

# Declare a topic exchange
exchange_name = 'financial_events'
channel.exchange_declare(exchange=exchange_name, exchange_type='topic')

# Define messages and their routing keys
messages = {
    'trade.fx.eurusd': 'Executed EUR/USD buy @ 1.0855',
    'analytics.volatility.eurusd': 'EUR/USD 1-day volatility is high',
    'trade.fx.gbpjpy': 'Executed GBP/JPY sell @ 199.80'
}

# Publish each message
for routing_key, message in messages.items():
    channel.basic_publish(exchange=exchange_name, routing_key=routing_key, body=message)
    print(f" [x] Sent '{routing_key}':'{message}'")

connection.close()

The Subscribers

Now, you will create the three subscribers. Each will have its own script and will listen for a specific pattern of messages.

Subscriber 1: FX Trade Blotter (fx_blotter.py)

This service is interested in all FX trades, regardless of the currency pair.
Create a file named fx_blotter.py with the following code.

#!/usr/bin/env python
import pika

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

exchange_name = 'financial_events'
channel.exchange_declare(exchange=exchange_name, exchange_type='topic')

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

# This binding key captures all messages starting with 'trade.fx'
binding_key = 'trade.fx.*'
channel.queue_bind(exchange=exchange_name, queue=queue_name, routing_key=binding_key)

print(' [*] FX Blotter waiting for FX trades. To exit press CTRL+C')

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

channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
channel.start_consuming()

Subscriber 2 & 3: Your Task

Following the pattern of fx_blotter.py, create two more subscriber files:

1. eur_auditor.py: This service needs to receive all events related to the eurusd pair, whether they are trades or analytics.
   • Hint: What binding key would match any message containing .eurusd? A key like *.*.eurusd would work perfectly.
2. analytics_dashboard.py: This service is interested in all analytics messages, regardless of the instrument.
   • Hint: The # wildcard is useful here. A binding key like analytics.# will match analytics.volatility.eurusd and any other keys that start with analytics..

Running the System

1. Open three separate terminal windows. In each one, start one of your subscriber scripts:

   # Terminal 1
   python fx_blotter.py
   
   # Terminal 2
   python eur_auditor.py
   
   # Terminal 3
   python analytics_dashboard.py
   

2. Open a fourth terminal and run the publisher:

   python publisher.py
   

3. Observe the output in your subscriber terminals. You should see the following:

   • fx_blotter received the two trade messages (trade.fx.eurusd and trade.fx.gbpjpy).
   • eur_auditor received the two eurusd messages (trade.fx.eurusd and analytics.volatility.eurusd).
   • analytics_dashboard received the one analytics message (analytics.volatility.eurusd).

This exercise demonstrates the power of topic exchanges. The publisher simply describes its data with routing keys, and the consumers independently decide which data they are interested in.

---

Conclusion

In this lesson, you've moved beyond simple broadcasting and implemented sophisticated message routing. You have learned how to write Python publishers and subscribers that use direct and topic exchanges to ensure messages are delivered precisely where they need to go.

Key Takeaways:

• Direct exchanges route messages based on an exact match between the routing key and the binding key, ideal for targeted, one-to-one delivery.
• Topic exchanges use wildcard patterns (* and #) for flexible, pattern-based routing, enabling complex one-to-many or many-to-many communication.
• The publisher remains decoupled from the consumers. It only needs to provide a descriptive routing key, not a list of recipients.
• This routing logic is central to building scalable and maintainable distributed systems, allowing services to evolve independently.

Preview of the Next Lesson:

So far, we've focused on routing messages to different types of consumers. But what if one type of consumer can't keep up with the volume of messages? In the next lesson, we will explore the competing consumers pattern, which allows you to scale out processing by having multiple instances of the same consumer process messages from a single queue in parallel.