Why do CELERY_ROUTES have both a "queue" and a "routing_key"?

Question

My understanding of AMQP is that messages only have the following components:

1. The message body
2. The routing key
3. The exchange

Queues are attached to exchanges. Messages can't have any knowledge of queues. They just post to an exchange, and then based on the exchange type and routing key, the messages are routed to one or more queues.

In Celery, the recommended way of routing tasks is through the CELERY_ROUTES setting. From the docs, CELERY_ROUTES is...

A list of routers, or a single router used to route tasks to queues. http://celery.readthedocs.org/en/latest/configuration.html#message-routing

And it includes an example...

To route a task to the feed_tasks queue, you can add an entry in the CELERY_ROUTES setting:

CELERY_ROUTES = {
    'feeds.tasks.import_feed': {
        'queue': 'feed_tasks',
        'routing_key': 'feed.import',
    },
}

But wait a minute -- According to AMQP, messages only come with a routing key! What the heck is the "queue" doing there?

Furthermore, there's this notion of a default queue. If you invoke a task which isn't caught by CELERY_ROUTES, it falls back to CELERY_DEFAULT_QUEUE. But again -- in AMQP, messages don't know about queues. Shouldn't that be the default routing key instead?

Solution 1

Is true that on Celery there is a bit of confusion when you go to Queues, one thing you must keep in mind is that queue parameter refers to a Celery Kombu Queue Object and not directly to a AMQP queue, you can understand this by reading this extract from the docs. Of course the fact that celery creates the queue and exchange with the same name is the origin of confusion of the usage of queue parameter. Always in the docs you can read this paragraph:

If you have another queue but on another exchange you want to add, just specify a custom exchange and exchange type:

CELERY_QUEUES = (
    Queue('feed_tasks',    routing_key='feed.#'),
    Queue('regular_tasks', routing_key='task.#'),
    Queue('image_tasks',   exchange=Exchange('mediatasks', type='direct'),
                       routing_key='image.compress'),
)

So in this way you could bind 2 different queue on the same exchange. After to route the task using only the exchange and the key you could use Routers class

class MyRouter(object):

    def route_for_task(self, task, args=None, kwargs=None):
        if task == 'myapp.tasks.compress_video':
            return {'exchange': 'video',
                    'exchange_type': 'topic',
                    'routing_key': 'video.compress'}
        return None

More here http://celery.readthedocs.org/en/latest/userguide/routing.html#routers

Solution 2

The point of having the queue declared in there, is for celery to create those queues and set up the configuration with RabbitMQ.

For a lower level AMQP client, you need to first declare the queue, then the exchange, and then finally bind the exchange to the queue. Later when posting messages, you just post to the exchange.

It seems like celery use this structure to do it automatically.

Comments

[Jocelyn delalande]

Not sure to understand : calling Queue() instantiate... a binding and not a queue ?

[Mauro Rocco]

In some way you can say that the Queue object in kombu corresponds to a AMQP queue bind to a an exchange with a certain routing key.