celery task routing with CELERY_ROUTES not working with task subclasses

Question

I am using celery 3.0.20 and I have difficulties getting celery tasks to work with different queues.

My celery workers are launched with the queue configuration

-Q:w1 default -Q:w2 longrunning

The Django settings contains the following celery config:

CELERY_QUEUES = (
    Queue('default', Exchange('default'), routing_key='default'),
    Queue('longrunning', Exchange('longrunning'), routing_key='longrunning'),
)
CELERY_DEFAULT_QUEUE = 'default'
CELERY_DEFAULT_EXCHANGE_TYPE = 'direct'
CELERY_DEFAULT_ROUTING_KEY = 'default'

CELERY_ROUTES = ({'mytasks.task_a.run': {
                    'queue': 'longrunning',
                    'routing_key': 'longrunning'
             }},

So far, so good. All tasks go by default in the default queue, and task_a goes into the long-running queue.

The task_a module is implemented as follows:

from celery import task

@task
def run():
    # do some task work

Now, I have the problem that a different task is implemented as a class inheriting from celery.Task:

from celery import Task

class AnotherTask(Task):

    def run(self, *args, **kwargs):
        # do some task work

When the AnotherTask class now resides in task_b module, I cannot make this task to be executed in the longrunning queue: I tried adding it to CELERY_ROUTES in different varieties, none of them worked:

{'mytasks.task_b': {
    'queue': 'longrunning',
     'routing_key': 'longrunning'
}}

-

{'mytasks.task_b.AnotherTask': {
    'queue': 'longrunning',
     'routing_key': 'longrunning'
}}

-

{'mytasks.task_b.AnotherTask.run': {
    'queue': 'longrunning',
     'routing_key': 'longrunning'
}}

I also tried to switch to default exchange type 'topic', but that didn't work either.

Any hints how to get the task in class AnotherTask to be executed in the longrunning queue?

Michael Michael · Accepted Answer · 2014-05-28T13:07:58

OK, I found the problem.

It was due to the fact that all other tasks, ending up in the longrunning queue, where started regularly with celerybeat. The new task was started from the web environment. And the celerybeat settings imported the settings containing the CELERY_ROUTES dict, the web environment settings did not. Ugh!

celery task routing with CELERY_ROUTES not working with task subclasses

1 Answers