Task Routing in Celery — Complete Guide
In this tutorial, you will learn about Task Routing in Celery. We cover key concepts, practical examples, and best practices to help you master this topic.
Route Celery tasks to specific workers using queues, routing keys, custom routers, and topic-based routing for fine-grained control over task execution.
What You Learn
You will learn how to define task queues, configure routing keys, create custom routers, use RabbitMQ topic routing, and assign tasks to different worker pools based on task type or priority.
Why It Matters
Not all tasks are equal. CPU-intensive tasks should go to separate workers from I/O-bound tasks. High-priority tasks should skip the queue. Routing allows you to optimize resource allocation and meet performance requirements per task type.
Real-World Use
DodaTech routes file scanning tasks to CPU-optimized workers and database cleanup tasks to memory-optimized workers. Alert notifications go to a dedicated low-latency queue with always-on consumers.
Queue Definitions
flowchart LR
P[Producer] --> D[Default Queue]
P --> H[High Priority Queue]
P --> C[CPU Tasks Queue]
W1[Default Worker] --> D
W2[High Priority Worker] --> H
W3[CPU Worker] --> C
style H fill:#f90,color:#fff
from celery import Celery
app = Celery('routing', broker='redis://localhost:6379/0')
app.conf.update(
task_routes={
'tasks.high_priority': {'queue': 'high'},
'tasks.cpu_intensive': {'queue': 'cpu'},
'tasks.io_intensive': {'queue': 'io'},
},
task_queues={
'high': {'routing_key': 'high.#'},
'cpu': {'routing_key': 'cpu.#'},
'io': {'routing_key': 'io.#'},
'default': {'routing_key': 'default.#'},
},
task_default_queue='default',
task_default_routing_key='default.task',
)
@app.task
def high_priority(data):
return f"High priority: {data}"
@app.task
def cpu_intensive(data):
return f"CPU task: {data}"
@app.task
def io_intensive(data):
return f"I/O task: {data}"
Explicit Routing with apply_async
from celery import Celery
app = Celery('explicit', broker='redis://localhost:6379/0')
@app.task
def analysis_task(data):
return f"Analysis: {data}"
# Route to specific queue
result = analysis_task.apply_async(
args=("urgent analysis",),
queue='high',
routing_key='high.analysis'
)
print(f"Task routed to 'high' queue: {result.id}")
# Route with exchange (RabbitMQ only)
result2 = analysis_task.apply_async(
args=("batch analysis",),
exchange='analysis',
routing_key='batch.analysis'
)
print(f"Task with exchange routing: {result2.id}")
Custom Task Router
from celery import Celery
app = Celery('router', broker='redis://localhost:6379/0')
class TaskRouter:
"""Route tasks based on task name and arguments."""
def route_for_task(self, task, args=None, kwargs=None):
if task == 'tasks.process_image':
return {'queue': 'cpu', 'routing_key': 'cpu.image'}
elif task == 'tasks.send_email':
return {'queue': 'email', 'routing_key': 'email.send'}
elif task == 'tasks.generate_report':
if kwargs and kwargs.get('priority') == 'high':
return {'queue': 'high', 'routing_key': 'high.report'}
return {'queue': 'io', 'routing_key': 'io.report'}
return {'queue': 'default', 'routing_key': 'default.task'}
app.conf.update(
task_routes=[TaskRouter()],
task_queues={
'cpu': {'routing_key': 'cpu.#'},
'email': {'routing_key': 'email.#'},
'io': {'routing_key': 'io.#'},
'high': {'routing_key': 'high.#'},
'default': {'routing_key': 'default.#'},
}
)
@app.task
def process_image(path):
return f"Processing image: {path}"
@app.task
def send_email(to, subject):
return f"Sending email to {to}: {subject}"
@app.task
def generate_report(report_id, priority='normal'):
return f"Report {report_id} (priority: {priority})"
RabbitMQ Topic Routing
When using RabbitMQ broker, Celery leverages RabbitMQ's topic exchanges:
from celery import Celery
app = Celery('topic_router', broker='amqp://guest:guest@localhost:5672//')
app.conf.update(
task_queues={
'log.error': {'exchange': 'logs', 'routing_key': 'log.error'},
'log.warning': {'exchange': 'logs', 'routing_key': 'log.warning'},
'log.info': {'exchange': 'logs', 'routing_key': 'log.info'},
'audit': {'exchange': 'logs', 'routing_key': 'audit.#'},
},
task_default_exchange='logs',
task_default_exchange_type='topic',
)
@app.task
def log_error(message):
return f"ERROR: {message}"
@app.task
def log_warning(message):
return f"WARNING: {message}"
@app.task
def log_info(message):
return f"INFO: {message}"
Worker-Specific Queues
Start workers that consume from specific queues:
# Worker for CPU tasks only
celery -A tasks worker --queues=cpu --concurrency=4 --hostname=cpu-worker@%h
# Worker for high priority tasks only
celery -A tasks worker --queues=high --concurrency=2 --hostname=high-worker@%h
# Worker for multiple queues
celery -A tasks worker --queues=default,io,email --concurrency=8 --hostname=general-worker@%h
# Verify which queues each worker consumes
from tasks import app
inspect = app.control.inspect()
active_queues = inspect.active_queues()
for worker, queues in active_queues.items():
print(f"{worker}:")
for q in queues:
print(f" - {q['name']} ({q['routing_key']})")
Expected output:
celery@cpu-worker:
- cpu (cpu.#)
celery@high-worker:
- high (high.#)
celery@general-worker:
- default (default.task)
- io (io.#)
- email (email.send)
Routing Based on Task Arguments
from celery import Celery
app = Celery('arg_router', broker='redis://localhost:6379/0')
class ArgBasedRouter:
def route_for_task(self, task, args=None, kwargs=None):
if task == 'tasks.process_file':
file_type = kwargs.get('file_type', 'default')
routes = {
'video': {'queue': 'cpu', 'routing_key': 'cpu.video'},
'image': {'queue': 'cpu', 'routing_key': 'cpu.image'},
'text': {'queue': 'io', 'routing_key': 'io.text'},
}
return routes.get(file_type, {'queue': 'default'})
return None
app.conf.task_routes = [ArgBasedRouter()]
@app.task
def process_file(path, file_type='default'):
return f"Processing {file_type} file: {path}"
Common Mistakes
1. Not Defining Queues Explicitly
Without explicit queue definitions, Celery creates default queues. For routing to work, define all queues in task_queues configuration.
2. Starting Workers Without --queues
By default, workers consume only the default queue. If you route tasks to custom queues, start workers with --queues=queue_name.
3. Mismatched Routing Keys
If the routing key in apply_async does not match any queue binding, the task goes nowhere. Verify routing keys match queue definitions.
4. Using Routing Without a Message Broker that Supports It
Redis broker has limited routing support. For full topic-based routing with wildcards, use RabbitMQ as the broker.
5. Not Testing Routing
Always verify that tasks arrive at the correct queue. Use the management UI (RabbitMQ) or redis-cli (Redis) to inspect queue contents after routing.
Practice Questions
1. How do you define task queues in Celery?
Use task_queues config with queue names and routing keys. Example: task_queues={'high': {'routing_key': 'high.#'}}.
2. How does a custom router work?
Create a class with route_for_task(self, task, args, kwargs) that returns a dict with queue and routing_key. Assign it to task_routes.
3. What is the difference between queue and routing_key in Celery?
A queue is a named destination. The routing_key is used by the broker to match the task to the correct queue binding.
4. How do you start a worker for a specific queue?
Use celery -A app worker --queues=queue_name. Multiple queues: --queues=q1,q2,q3.
Challenge
Design a routing Strategy for a multi-service platform. Tasks: email notifications (fast, low resource), video transcoding (slow, CPU-heavy), database backups (nightly, I/O-heavy), and billing (time-sensitive, high priority). Define queues, routers, and worker configurations for each.
FAQ
Mini Project: Multi-Queue Task System
# routed_tasks.py
from celery import Celery
import time
app = Celery('routed_project', broker='redis://localhost:6379/0')
app.conf.update(
task_queues={
'critical': {'routing_key': 'critical.#'},
'default': {'routing_key': 'default.#'},
'batch': {'routing_key': 'batch.#'},
},
task_routes={
'routed_tasks.urgent_process': {'queue': 'critical'},
'routed_tasks.batch_process': {'queue': 'batch'},
},
task_default_queue='default',
)
@app.task
def urgent_process(data):
print(f"[CRITICAL] Processing urgent: {data}")
time.sleep(0.5)
return f"Urgent done: {data}"
@app.task
def batch_process(data):
print(f"[BATCH] Processing batch: {data}")
time.sleep(2)
return f"Batch done: {data}"
@app.task
def normal_process(data):
print(f"[DEFAULT] Processing normal: {data}")
time.sleep(1)
return f"Normal done: {data}"
# run_routing.py
from routed_tasks import urgent_process, batch_process, normal_process
import time
urgent_process.delay("payment_failed")
batch_process.delay("report_123")
normal_process.delay("user_signup")
print("All tasks submitted to their queues")
print("Start workers:")
print(" Worker 1: celery -A routed_tasks worker --queues=critical --concurrency=2")
print(" Worker 2: celery -A routed_tasks worker --queues=batch --concurrency=4")
print(" Worker 3: celery -A routed_tasks worker --queues=default --concurrency=8")
Expected output:
All tasks submitted to their queues
Start workers:
Worker 1: celery -A routed_tasks worker --queues=critical --concurrency=2
Worker 2: celery -A routed_tasks worker --queues=batch --concurrency=4
Worker 3: celery -A routed_tasks worker --queues=default --concurrency=8
What's Next
Now that you understand task routing, explore task queues for organizing and prioritizing task execution, then learn about task priority for controlling execution order within queues.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro