Celery Task Routing with task_routes

Routing is the mechanism behind almost every operational lever in Celery — isolation, priority, per-workload scaling — and it belongs with the rest of the configuration surface in Celery Architecture & Configuration, part of Backend Frameworks & Worker Scaling.

Problem Statement

Every task in your application runs on the default celery queue. You cannot scale image processing independently of email, you cannot give a slow report its own worker pool, and a backlog in one workload delays all the others. Individual teams have started passing queue="whatever" at each call site, so the mapping now lives in ninety places and nobody can answer which tasks land where.

Prerequisites

  • Celery 5.3+, with configuration loaded identically by producers and consumers.
  • Kombu Queue definitions available on both sides, so declarations match.
  • A broker whose queue declarations you can change — note that RabbitMQ queue arguments are immutable after creation.
  • Worker processes you can start with distinct -Q arguments.

Step 1 — Declare Queues Before Routing to Them

Routing to an undeclared queue is a silent failure on most brokers: the message goes to an exchange with no binding and disappears.

# celeryconfig.py
from kombu import Exchange, Queue

task_default_queue = "default"
task_default_exchange = "tasks"
task_default_routing_key = "default"

tasks_exchange = Exchange("tasks", type="direct")

task_queues = (
    Queue("default",  tasks_exchange, routing_key="default"),
    Queue("media",    tasks_exchange, routing_key="media"),
    Queue("reports",  tasks_exchange, routing_key="reports"),
    Queue("email",    tasks_exchange, routing_key="email"),
)

A direct exchange with one routing key per queue is the right default. Reach for topic only when you genuinely need pattern-based fan-out, because its wildcard matching makes it much harder to answer "where does this task go?" by reading the config.

Step 2 — Write the Routing Table

task_routes maps task names to destinations. Patterns are matched in insertion order, and the first match wins, so specific entries must precede wildcards.

task_routes = {
    # Specific first — a wildcard above these would swallow them.
    "app.media.generate_thumbnail": {"queue": "media", "routing_key": "media"},
    "app.reports.monthly_summary":  {"queue": "reports", "routing_key": "reports"},

    # Then patterns.
    "app.media.*":   {"queue": "media",   "routing_key": "media"},
    "app.reports.*": {"queue": "reports", "routing_key": "reports"},
    "app.email.*":   {"queue": "email",   "routing_key": "email"},
}
Which routing decision wins A precedence ladder for Celery routing. An explicit queue argument passed to apply_async wins over everything. Below it, a matching entry in task_routes decides. Below that, a queue set on the task decorator applies. If nothing matches, the message goes to task_default_queue. The diagram warns that call-site overrides make the routing table unreliable. Routing precedence, highest first 1 · apply_async(queue="…") at the call site wins over everything — which is why it should be rare 2 · task_routes match (first entry wins) the single source of truth you want to be using 3 · queue= on the @task decorator useful for a task that always belongs somewhere specific 4 · task_default_queue never make this your urgent queue

Step 3 — Use a Callable Router for Dynamic Decisions

When the destination depends on the arguments rather than the task name — routing by tenant tier, by payload size, by region — a callable router is the supported mechanism.

def route_task(name, args, kwargs, options, task=None, **kw):
    """Return a routing dict, or None to fall through to the next router."""
    if name.startswith("app.reports."):
        # Large exports get their own queue so they cannot block small ones.
        if kwargs.get("row_estimate", 0) > 100_000:
            return {"queue": "reports_large", "routing_key": "reports_large"}
        return {"queue": "reports", "routing_key": "reports"}

    if kwargs.get("tenant_tier") == "enterprise":
        return {"queue": "priority", "routing_key": "priority"}

    return None        # fall through to the dict-based table

task_routes = (route_task, {          # routers are tried in order
    "app.media.*": {"queue": "media"},
    "app.email.*": {"queue": "email"},
})

Returning None rather than a default is what makes routers composable: a callable handles the cases it knows about and lets the static table cover the rest.

Step 4 — Start Workers That Consume Those Queues

celery -A app worker -Q media   -c 4  -n media@%h  --prefetch-multiplier=1
celery -A app worker -Q reports -c 2  -n reports@%h --prefetch-multiplier=1
celery -A app worker -Q email,default -c 8 -n email@%h

A queue with no consumer accumulates silently — no error, no warning, just a growing depth nobody is watching. Make queue-to-worker coverage an explicit check rather than an assumption:

def unconsumed_queues(app) -> set[str]:
    """Declared queues that no live worker is consuming. Run in CI or at boot."""
    declared = {q.name for q in app.conf.task_queues}
    active = app.control.inspect().active_queues() or {}
    consumed = {q["name"] for queues in active.values() for q in queues}
    return declared - consumed

Step 5 — Verify Routing Without Publishing

Celery can resolve a route without sending anything, which makes routing testable in CI.

def test_routing_table():
    cases = {
        "app.media.generate_thumbnail": "media",
        "app.media.transcode_video":    "media",
        "app.reports.monthly_summary":  "reports",
        "app.email.send_receipt":       "email",
        "app.unknown.task":             "default",
    }
    for name, expected in cases.items():
        route = app.amqp.router.route({}, name, args=(), kwargs={})
        assert route["queue"].name == expected, f"{name} → {route['queue'].name}"

This test is worth more than it looks: routing bugs are otherwise discovered when a queue quietly stops receiving work, which can take days to notice.

Step 6 — Watch Every Queue Separately

Aggregate depth hides a single stalled queue The upper chart shows total queue depth across all queues, which looks stable. The lower charts break it down by queue and reveal that the reports queue is growing steadily while media and email are draining, a pattern completely invisible in the aggregate. Aggregate depth is not a queue health metric total depth looks fine media email reports growing — no consumer since the last deploy 
# Depth and oldest age per queue — never aggregate across queues
celery_queue_length{queue=~"media|reports|email|default"}
max by (queue) (celery_queue_oldest_task_age_seconds)

Step 7 — Keep Call Sites Out of the Routing Decision

The reason to centralise routing is that a queue name at a call site is a deployment decision embedded in application code. When operations later want to split the media queue in two, they must change ninety files instead of one table — and every file they miss keeps publishing to a queue that may no longer have consumers.

Enforce it mechanically. A lint rule that rejects queue= in apply_async calls outside an approved module costs an afternoon and prevents years of drift:

# tests/test_no_inline_queue_args.py — cheap guard, run in CI
import ast, pathlib

def test_no_inline_queue_kwarg():
    offenders = []
    for path in pathlib.Path("app").rglob("*.py"):
        tree = ast.parse(path.read_text())
        for node in ast.walk(tree):
            if (isinstance(node, ast.Call)
                    and getattr(node.func, "attr", "") in {"apply_async", "delay"}
                    and any(k.arg == "queue" for k in node.keywords)):
                offenders.append(f"{path}:{node.lineno}")
    assert not offenders, f"route via task_routes, not at the call site: {offenders}"

The exception worth allowing is a deliberate override in a small, reviewed module — a replay tool that must target a specific queue, for instance. Keeping those in one place preserves the property that matters: someone can read a single file and know where every task goes.

Step 8 — Split Queues by What Actually Differs

A routing table is only as useful as the queue design behind it, and the most common mistake is splitting by team or by feature rather than by operational characteristics. Queues exist so that work with different requirements can be scaled, tuned, and observed separately; if two workloads share every characteristic, splitting them adds monitoring surface and buys nothing.

Split when at least one of these differs materially:

Dimension Why it justifies a queue
Duration A 20-minute job on the same queue as a 200ms job blocks it; different queues get different prefetch and grace periods
Latency requirement An interactive queue needs reserved capacity; a batch queue does not
Downstream dependency One saturated API should not stall workloads that do not use it
Failure profile A flaky integration deserves its own retry policy and dead-letter queue
Scaling signal Work that scales with user traffic versus work that scales with data volume

Duration is the one that pays off most often and is noticed least. A single slow task class dragged into a fast queue changes the correct prefetch, the correct grace period, and the meaning of that queue's depth — which is why "we added one report task to the email queue" so frequently precedes an incident. When in doubt, split on duration first and merge later if the operational overhead proves unjustified; merging two queues is a much easier migration than splitting a busy one.

When two workloads deserve separate queues Four cards naming the dimensions on which a split is justified. Duration, because prefetch and drain windows differ. Latency requirement, because interactive work needs reserved capacity. Downstream dependency, because a saturated quota should not stall unrelated work. Failure profile, because a flaky integration needs its own retry policy and alerts. Split on characteristics, not on team or feature Duration one prefetch and one grace period cannot suit a 200ms job and a 20-minute job Latency requirement interactive work needs reserved capacity; batch work needs none Downstream dependency a rate-limited API should bound one queue, not every workload sharing the pool Failure profile a flaky integration wants its own retry policy, dead-letter queue and alert thresholds

Verification

  1. The table resolves as intended. The CI routing test above must pass for every task name in the codebase, including ones that should land on the default.
  2. Every queue has a consumer. unconsumed_queues() must return empty against a running fleet.
  3. No task publishes to an undeclared queue. Enable broker-side logging of unroutable messages; on RabbitMQ, set an alternate exchange and alert on anything arriving there.
  4. A new task lands correctly. Add a task matching an existing pattern and confirm, with inspect active_queues and a real publish, that it arrives where the table says.

Gotchas & Edge Cases

task_routes is producer-side. The routing decision happens where apply_async runs, so a producer with stale configuration publishes to the old queue after a rename. Deploy producers and consumers together during routing changes.

Dictionary order is match order. In Python 3.7+ dicts preserve insertion order, and Celery relies on it. A wildcard placed above a specific rule silently swallows it.

Renaming a queue orphans in-flight messages. Messages already on the old queue need a consumer until it drains. Keep the old worker running through one full drain before removing it.

RabbitMQ queue arguments cannot be changed. Adding x-max-priority or a dead-letter exchange to an existing queue raises PRECONDITION_FAILED; create a new queue and migrate.

task_default_queue catches everything unmatched. If it points at a queue with a small pool, a typo in a task name quietly routes real work into a starved queue. Point the default at your general-purpose pool and alert on unexpected traffic there.

Related