RabbitMQ : Multiple binding & Routing

RabbitMQ & Celery Tutorials

In this chapter, we're going to direct only critical error messages to the log file, while still print all of the log messages on the console. In other words, we're going to add a feature that allows us to make it possible to subscribe only to a subset of the messages.

Picture from slides.com.

As we already know, the code syntax for binding the exchanges to the queues looks like this:

channel.queue_bind(exchange=exchange_name,
                   queue=queue_name)

What the code is telling: I, "queue_name", am interested in the message from "exchange_name".

Bindings can take an extra routing_key parameter and we create a binding with a key like this:

channel.queue_bind(exchange=exchange_name,
                   queue=queue_name,
                   routing_key='black')

We're going to use a direct exchange instead of the fanout exchange we used in previous chapter. This direct exchange guides a message to go to the queues whose binding key exactly matches the routing key of the message.

With a setup in the picture above, a message published to the exchange with a routing key orange will be routed to Queue #1. Messages with a routing key of black or green will go to Queue #2. All other messages will be discarded.

The picture above shows an example of multiple binding: bind multiple queues (Queue #1 and Queue #2) with the same binding key (green). In this case, this direct exchange setup will behave like fanout and will broadcast the message to all the matching queues: a message with routing key green will be delivered to both Queues.

We'll use direct exchange for our logging system instead of fanout. We will supply the log severity as a routing key so that a receiver can select the severity ('info', 'warning', or 'error') it wants to receive.

Let's create an exchange first:

channel.exchange_declare(exchange='direct_logs',
                         type='direct')

To send a message:

channel.basic_publish(exchange='direct_logs',
                      routing_key=severity,
                      body=message)

The only difference from the previous chapter: we're going to create a new binding for each severity we're interested in:

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

for severity in severities:
    channel.queue_bind(exchange='direct_logs',
                       queue=queue_name,
                       routing_key=severity)

emit_log_direct.py:

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

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

channel.exchange_declare(exchange='direct_logs',
                         type='direct')

severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='direct_logs',
                      routing_key=severity,
                      body=message)
print " [x] Sent %r:%r" % (severity, message)
connection.close()

receive_logs_direct.py:

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

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

channel.exchange_declare(exchange='direct_logs',
                         type='direct')

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

severities = sys.argv[1:]
if not severities:
    print >> sys.stderr, "Usage: %s [info] [warning] [error]" % \
                         (sys.argv[0],)
    sys.exit(1)

for severity in severities:
    channel.queue_bind(exchange='direct_logs',
                       queue=queue_name,
                       routing_key=severity)

print ' [*] Waiting for logs. To exit press CTRL+C'

def callback(ch, method, properties, body):
    print " [x] %r:%r" % (method.routing_key, body,)

channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)

channel.start_consuming()

To save only 'warning' and 'error' (and not 'info') log messages to a file, just open a console and type:

$ python receive_logs_direct.py warning error > logs_from_rabbit.log

to see all the log messages on our screen, open a new terminal and do:

$ python receive_logs_direct.py info warning error
 [*] Waiting for logs. To exit press CTRL+C

To emit an error log message, type:

$ python receive_logs_direct.py warning error > logs_from_rabbit.log

1. RabbitMQ - Routing

RabbitMQ & Celery Tutorials