Creating queues dynamically with Celery

Question

I am writing a mailing-list manager using Django, Celery, and RabbitMQ. When a message comes in, a task is executed for each recipient. All tasks go to a single queue, and one or more workers consume tasks from the queue, constructing the email messages and sending them.

The single queue causes a fairness problem: if a message comes in to a large mailing list, a large number of tasks are added to the queue, and other messages cannot get through to other, smaller mailing lists until all the messages to the large list have been sent. How can I find a way around this?

Conceptually, a solution would be to create a queue for each mailing list and have the worker(s) consume tasks from the various queue round robin. Is this possible in Celery, given that I need to be able to create new mailing lists dynamically? I have not seen functionality for creating queues dynamically or for making the worker(s) consume from new queues.

Answer 1

As shown in the image below consider a system which makes use of a topic exchange instead of a direct exchange.

Unlike a direct exchange, Topic exchange allows us to route different messages to different queues. this is done by setting routing_key for each message and binding certain queues to accept only messages with a particular routing keys.

A system can be set up that high priority go to a dedicated queue and consumer, similarly normal or low priority messages are handled by one or more queues.

Celery supports load balancing across multiple workers(number of workers are hardware dependent). Celery config setting like BROKER_POOL_LIMIT, PREFETCH_LIMIT etc help better load balance and reduce congestion.