Work celery: how to consume from all the queues?

Question

I have

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

I see that a new queue is being created in the route returned by the custom router, which I assume is due to CELERY_CREATE_MISSING_QUEUES .

Now in the working node that I run, I do not pass the -Q argument, and it only consumes from the default queue, which seems to match the documentation -

By default, it will be consumed from all the queues defined in CELERY_QUEUES (which, if not specified by default for the queue called celery).

Is there a way to get my working node to consume from all queues, including those created dynamically?

Thanks,

Answer 1

The employer needs to report these automatically or dynamically created queues, so you need a way to get these queue names and store them, perhaps when you create them or get them, perhaps from rabbitmqctl list_queues , if you use RabbitMQ as a broker, and, for example, add a signal handler to add these dynamic queues to consumed workers.

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, you can also get workers to start consuming 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 .

Hope this helps, I will edit the question when I get more information about this.