Celery worker: How to consume from all queues?

Question

I have

• set CELERY_CREATE_MISSING_QUEUES = True
• NOT defined CELERY_QUEUES
• defined CELERY_DEFAULT_QUEUE = 'default' (of type direct)
• a custom router class that creates routes on the fly as shown in this ticket (https://github.com/celery/celery/issues/150).

I see that the new queue in the route returned by the custom router gets created which I assume is because of CELERY_CREATE_MISSING_QUEUES.

Now in the worker node that I run, I don't pass the -Q argument and it consumes only from the 'default' queue which seems to be in line with the documentation -

By default it will consume from all queues defined in the CELERY_QUEUES setting (which if not specified defaults to the queue named celery).

Is there any way to get my worker node to consume from ALL queues including the ones that are created dynamically?

Thanks,

Answer 1

The worker needs to be told about these automatically or dynamically created queues, so you need a way to get these queue names and store them maybe when you create them or get them maybe from rabbitmqctl list_queues if you're using RabbitMQ as the broker, and for example add a signal handler to add these dynamic queues to the workers to consume from.

For example using celeryd_after_setup signal:

from celery.signals import celeryd_after_setup

@celeryd_after_setup.connect
def add_dynamic_queue(sender, instance, **kwargs):
    # get the dynamic queue, maybe stored somewhere
    queue = 'dynamic_queue'
    instance.app.amqp.queues.select_add(queue)

If you always have new dynamic queues created, you can also command the workers to start consuming from these queues at runtime using:

#command all workers to consume from the 'dynamic_queue' queue
app.control.add_consumer('dynamic_queue', reply=True)

# command specific workers
app.control.add_consumer('dynamic_queue', reply=True, destination=[w1@example])

See Adding Consumers.

I hope this helps, I'll edit the question when I get more info about this.