RabbitMQ : Work Queues (Task Queues)

RabbitMQ & Celery Tutorials

In this chapter, we'll create a Work Queues (Task Queues) that will be used to distribute time-consuming tasks among multiple workers. By creating the Work Queues, we can avoid starting a resource-intensive task immediately and having to wait for it to complete. We may want to schedule the heavy task to be done later.

We'll simulate the busy work loads using the sleep() function.

The code doing task schedule is task.py, and it looks like this:

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

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

channel.queue_declare(queue='task_queue', durable=True)

message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(exchange='',
                      routing_key='task_queue',
                      body=message,
                      properties=pika.BasicProperties(
                         delivery_mode = 2, # make message persistent
                      ))
print " [x] Sent %r" % (message,)
connection.close()

Note that we can now send arbitrary messages from the command line.

This worker.py will display the work in the message body. We simulate the task load as doc('.'), each accounts for 5 seconds duration. It will pop messages from the queue and perform the task, so let's call it worker.py:It will pop messages from the queue and perform the task:

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

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

channel.queue_declare(queue='task_queue', durable=True)
print ' [*] Waiting for messages. To exit press CTRL+C'

def callback(ch, method, properties, body):
    print " [x] Received %r" % (body,)
    time.sleep( body.count('.') )
    print " [x] Done"
    ch.basic_ack(delivery_tag = method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
                      queue='task_queue')

channel.start_consuming()

Let's have three workers perform their tasks at the same time. We can simulate this with three console terminals each running worker.py and the 4th console, we run task.py to create works for our workers.

Let the three worker in waiting mode:

W1$ python worker.py
 [*] Waiting for messages. To exit press CTRL+C

W2$ python worker.py
 [*] Waiting for messages. To exit press CTRL+C

W3$ python worker.py
 [*] Waiting for messages. To exit press CTRL+C

Then, we run the task.py with several(6) tasks:

$ python task.py taks1.;python task.py taks2..;python task.py taks3...;python task.py taks4....;python task.py taks5.....;python task.py taks6......

We can see how the workers are doing on each console:

W1$ python worker.py
 [*] Waiting for messages. To exit press CTRL+C
 [x] Received 'taks1.'
 [x] Done
 [x] Received 'taks4...'
 [x] Done


W2$ python worker.py
 [*] Waiting for messages. To exit press CTRL+C
 [x] Received 'taks2..'
 [x] Done
 [x] Received 'taks5.....'
 [x] Done

W3$ python worker.py
 [*] Waiting for messages. To exit press CTRL+C
 [x] Received 'taks3...'
 [x] Done
 [x] Received 'taks6......'
 [x] Done

RabbitMQ supports message acknowledgments to make sure a message is never lost. An ack(nowledgement) is sent back from the consumer to tell RabbitMQ that a particular message had been received, processed and that RabbitMQ is free to delete it. The following lines of code is doing it in callback() function:

ch.basic_ack(delivery_tag = method.delivery_tag)

We compare the line with a line in send.py we used in previous chapter (Hello world : RabbitMQ) where we explicitly turned them off via the no_ack=True flag:

channel.basic_consume(callback,
                      queue='hello',
                      no_ack=True)

With acknowledgments turned on which is default setting, we can be sure that even if we kill a worker using CTRL+C while it was processing a message, nothing will be lost. Soon after the worker dies all unacknowledged messages will be redelivered.

In previous section, we can make sure that even if the consumer dies, the task isn't lost.
However, our tasks will still be lost if RabbitMQ server stops!

When RabbitMQ quits or crashes it won't remember the queues and messages unless we tell it not to. How?
We should mark both the queue and messages as durable:

For queue to be persistent:

channel.queue_declare(queue='task_queue', durable=True)

Then, the message as well:

channel.basic_publish(exchange='',
                      routing_key="task_queue",
                      body=message,
                      properties=pika.BasicProperties(
                         delivery_mode = 2, # make message persistent
                      ))

By using the basic.qos() method with the prefetch_count=1 setting, we can tell RabbitMQ not to give more than one message to a worker at a time. Or, in other words, don't dispatch a new message to a worker until it has processed and acknowledged the previous one. Instead, it will dispatch it to the next worker that is not still busy.

channel.basic_qos(prefetch_count=1)

1. Work Queues

RabbitMQ & Celery Tutorials