TopolearnSeparating slow and fast work into queues keeps a backlog of heavy jobs from delaying time-critical tasks.
By default every task goes to one queue named celery, and every worker consumes it. That's fine until a burst of 10-minute video encodes blocks password-reset emails.
Routing sends tasks to named queues:
- Per call:
apply_async(queue="emails"). - Per task in config:
task_routes = {"proj.tasks.send_email": {"queue": "emails"}, "proj.video.*": {"queue": "video"}}; glob patterns work. - Per task in the decorator:
@app.task(queue="emails").
Then start workers for specific queues: celery -A proj worker -Q emails -c 8 and celery -A proj worker -Q video -c 2. A worker can consume several queues (-Q celery,emails). Queues are created automatically on first use (task_create_missing_queues, on by default).
Design queues by workload characteristics, not by feature: latency-sensitive vs batch, CPU-bound vs IO-bound, memory-hungry. Each kind then gets a worker pool sized and configured for it, and you can scale them independently.
Don't forget: a task routed to a queue nobody consumes simply waits there forever with no error. When you add a queue, add a worker for it.
Example
# proj/celery.py
from celery import Celery
app = Celery("proj", broker="redis://localhost:6379/0")
app.conf.task_routes = {
"proj.tasks.send_email": {"queue": "emails"},
"proj.video.*": {"queue": "video"}, # every task in proj.video
}
@app.task(name="proj.tasks.send_email")
def send_email(to): ...
@app.task(name="proj.tasks.cleanup")
def cleanup(): ... # default queue: "celery"
cleanup.apply_async(queue="maintenance") # per-call override
# $ celery -A proj worker -Q emails -c 8 -n emails@%h
# $ celery -A proj worker -Q video -c 2 -n video@%h
# $ celery -A proj worker -Q celery,maintenance -n default@%hRemember
- Default queue is 'celery'; route with task_routes, queue= or the decorator
- Start workers per queue with -Q
- Split queues by workload: latency, CPU vs IO, memory
- A queue without a consumer silently accumulates tasks
Watch out for
- Adding a route to a new queue but no worker consuming it.