RabbitMQ : Exchanges & Publish/Subscribe

RabbitMQ & Celery Tutorials

In this chapter, we're going to build a simple logging system, and we'll able to broadcast log messages to many receivers.

Picture from slides.com.

The producer never sends any messages directly to a queue. Instead, the producer can only send messages to an exchange. It not only receives messages from producers but also pushes them to queues. The exchange must know exactly what to do with a message it receives: whether it should be appended to a particular queue or appended to many queues. The answer depends on the exchange type.

There are 4 few exchange types: direct, topic, headers and fanout. In this chapter, we'll use the fanout type. Let's create an exchange of that type, and call it logs. The fanout just broadcasts all the messages it receives to all the queues it knows:

channel.exchange_declare(exchange='logs', type='fanout'

To list the exchanges on the server, we run rabbitmqctl:

$ sudo rabbitmqctl list_exchanges
Listing exchanges ...
	direct
amq.direct	direct
amq.fanout	fanout
amq.headers	headers
amq.match	headers
amq.rabbitmq.log	topic
amq.rabbitmq.trace	topic
amq.topic	topic
...done.

As we can see from the list, there are amq.* exchanges and the default (unnamed) exchange which are created by default.

Recall, in our previous chapter, we were able to send messages to queues using a default exchange, which we identify by the empty string (""):

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

However, we can now publish to our named exchange instead:

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

So far, we've been using named queue such as "hello" or "task_queue", but now we want to use "randomly named" queues. How? By not supplying the queue parameter to queue_declare:

result = channel.queue_declare()

the result.method.queue contains a random queue name.

Also, once we disconnect the consumer, the queue should be deleted:

result = channel.queue_declare(exclusive=True)

We need to tell the exchange to send messages to our queue. This process is called binding.

channel.queue_bind(exchange='logs',
                   queue=result.method.queue)

Now the exchange (logs) will append messages to our queue.

The producer program emits log messages. Since we now want to publish messages to our logs exchange instead of the nameless one, we supply a routing_key when sending, but its value is ignored for fanout exchanges. Since publishing to a non-existing exchange is forbidden, we declared the exchange after establishing the connection.

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

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

channel.exchange_declare(exchange='logs',
                         type='fanout')

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

#!/usr/bin/env python
import pika

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

channel.exchange_declare(exchange='logs',
                         type='fanout')

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

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

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

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

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

channel.start_consuming()

To save logs to a file, just open a console and type:

$ python receive_logs.py > logs_from_rabbit.log

To see the logs on your screen, spawn a new terminal and run:

$ python receive_logs.py

To see the logs on your screen, spawn a new terminal and run:

$ python emit_log.py

Then, on each console that's running receive_logs.py, we can see the following:

$ python receive_logs.py
 [*] Waiting for logs. To exit press CTRL+C
 [x] 'info: Hello World!'
 [x] 'info: Hello World!'

Using rabbitmqctl list_bindings. we can verify that the code actually creates bindings and queues as we want:

$ sudo rabbitmqctl list_bindings
Listing bindings ...
	exchange	amq.gen-gq_vRxMnVA2rn2_rMlqi-A	queue	amq.gen-gq_vRxMnVA2rn2_rMlqi-A	[]
	exchange	amq.gen-pMYkFEP6cLAsLc1W1OA8tg	queue	amq.gen-pMYkFEP6cLAsLc1W1OA8tg	[]
	exchange	task_queue	queue	task_queue	[]
logs	exchange	amq.gen-gq_vRxMnVA2rn2_rMlqi-A	queue	amq.gen-gq_vRxMnVA2rn2_rMlqi-A	[]
logs	exchange	amq.gen-pMYkFEP6cLAsLc1W1OA8tg	queue	amq.gen-pMYkFEP6cLAsLc1W1OA8tg	[]
...done.

From the output, we see logs exchanges including the exchanges we used in previous chapters.

1. RabbitMQ - Publish/Subscribe

RabbitMQ & Celery Tutorials