Routing High-Priority Jobs in Celery

This is the Celery-native implementation of the scheduling model in Priority Queues & Job Fairness, under Queue Fundamentals & Architecture — separate queues, dedicated pools, and the broker caveats that quietly break naive setups.

Problem Statement

Your Celery app runs everything on the default celery queue. A nightly analytics rollup enqueues 80,000 tasks at 02:00, and the 02:05 password-reset email lands behind all of them. Setting priority=9 on the task changed nothing, because your broker is Redis and Redis-backed Celery ignores per-message priority in the way most people assume it works. You need urgent tasks served within seconds regardless of backlog, without over-provisioning the entire fleet.

Prerequisites

• Celery 5.3+ with either Redis or RabbitMQ. The correct approach differs between them, which is most of this guide.
• The ability to run more than one worker process with different -Q arguments — the dedicated pool is what actually guarantees latency.
• Kombu Queue definitions available to both producer and consumer, so declarations match.

Step 1 — Define the Queues Explicitly

Never rely on the implicit default queue. Declare each class, with its exchange and routing key, in one place.

# celeryconfig.py
from kombu import Exchange, Queue

task_default_queue = "normal"
default_exchange = Exchange("tasks", type="direct")

task_queues = (
    Queue("urgent", default_exchange, routing_key="urgent"),
    Queue("normal", default_exchange, routing_key="normal"),
    Queue("bulk",   default_exchange, routing_key="bulk"),
)

# Declarative routing: the caller never chooses a queue, the rule set does.
task_routes = {
    "app.auth.send_password_reset":   {"queue": "urgent"},
    "app.auth.send_2fa_code":         {"queue": "urgent"},
    "app.billing.*":                  {"queue": "normal"},
    "app.analytics.rollup_*":         {"queue": "bulk"},
    "app.exports.generate_report":    {"queue": "bulk"},
}

task_acks_late = True
worker_prefetch_multiplier = 1     # see step 4 — this is not optional here

Glob patterns in task_routes keep the table short, but they are matched in dictionary order with the first match winning, so put specific names above wildcards. A task that matches nothing lands on task_default_queue, which is why the default should be normal and never urgent.

Step 2 — Reserve Capacity with Dedicated Workers

Routing alone changes nothing if one worker consumes all three queues: it will happily spend every slot on bulk work. The latency guarantee comes from a pool that only serves the urgent queue.

# Dedicated urgent pool — small, always idle enough to answer immediately
celery -A app worker -Q urgent -c 4 -n urgent@%h --prefetch-multiplier=1

# General pool — normal first, but helps urgent under burst
celery -A app worker -Q normal,urgent -c 12 -n normal@%h --prefetch-multiplier=1

# Bulk pool — isolated so a backlog cannot touch the other classes
celery -A app worker -Q bulk -c 6 -n bulk@%h --prefetch-multiplier=4

Celery consumes a -Q list with strict left-to-right preference per fetch, so -Q normal,urgent does not mean "urgent first". The dedicated pool is what provides the ordering guarantee; the mixed pool provides burst absorption. Size the urgent pool from the urgent arrival rate times its p99 duration, with at least 2× headroom — usually a small number, because genuinely urgent work is rare.

Bulk gets a higher prefetch because its jobs are long and throughput matters more than latency; urgent gets 1 because a prefetched task is a task that cannot be stolen by a free worker.

Step 3 — Understand Broker-Native Priority

Celery's priority argument behaves differently per transport, and the difference is the source of most confusion.

For RabbitMQ, declare the priority range on the queue itself — an existing queue cannot be upgraded in place, it must be deleted and recreated:

task_queues = (
    Queue("urgent", Exchange("tasks"), routing_key="urgent",
          queue_arguments={"x-max-priority": 9}),
)
task_queue_max_priority = 9
task_default_priority = 5

For Redis, the emulation means priority only distinguishes work within the same queue name, using a handful of sub-keys, and it interacts poorly with visibility_timeout and reclaim. On Redis, ignore priority and use separate queues — the approach in step 1 — which is portable and observable regardless of transport.

Step 4 — Do Not Let Prefetch Undo It

A worker with prefetch_multiplier=4 and concurrency 12 reserves up to 48 messages before executing any of them. Those messages are unavailable to every other worker, so an urgent task arriving one millisecond later waits behind 48 already-claimed bulk jobs. This single setting has ruined more priority designs than any broker limitation.

worker_prefetch_multiplier = 1     # urgent/normal pools
task_acks_late = True              # required for prefetch=1 to be safe on restarts

Keep prefetch_multiplier=1 on every queue where latency matters and raise it only on bulk pools where throughput dominates. The full trade-off — throughput versus fairness versus memory — is covered in tuning prefetch and consumer concurrency.

Step 5 — Watch Each Queue Separately

Aggregate queue depth hides exactly the failure you built this to prevent.

# Depth and head-of-queue age per class — urgent must stay near zero
celery_queue_length{queue="urgent"}
max by (queue) (celery_queue_oldest_task_age_seconds)

- alert: UrgentQueueBacklog
  expr: celery_queue_length{queue="urgent"} > 50
  for: 2m
  labels: { severity: critical }
  annotations:
    summary: "Urgent queue backing up — dedicated pool is saturated or stuck"

If the urgent queue ever has a sustained backlog, the dedicated pool is undersized or blocked on a slow dependency; both are worth paging for, because that queue exists precisely so it is never deep.

Step 6 — Size the Urgent Pool from Arrival Data

The dedicated pool only guarantees latency if it is large enough to absorb the urgent arrival rate without ever queueing, and small enough that its idle capacity is not wasteful. Little's Law gives the floor directly: the concurrency you need is arrival rate multiplied by service time.

def urgent_pool_size(arrivals_per_sec: float, p99_seconds: float,
                     headroom: float = 2.0) -> int:
    """Little's Law with headroom: L = λ × W, doubled so a burst still finds a free slot."""
    return max(2, math.ceil(arrivals_per_sec * p99_seconds * headroom))

urgent_pool_size(0.8, 1.5)    # 3  — 0.8 password resets/sec, 1.5s p99
urgent_pool_size(12, 0.4)     # 10 — a busier cache-invalidation stream

Two properties matter more than the exact number. First, the urgent pool should sit mostly idle — utilisation above roughly 50% means queueing has already begun and your p99 will degrade sharply at the next burst. Second, its cost is bounded and small: three or four processes reserved permanently is a rounding error against a fleet of twenty, and it is what converts "usually fast" into a latency guarantee you can put in an SLO. Re-derive the number whenever the urgent route list changes, because adding one chatty task to that queue is the usual way a working design quietly stops working.

Verification

1. Routing lands where you expect. celery -A app inspect active_queues on each worker must show exactly the queues you intended, and send_password_reset.apply_async() must appear on urgent with no explicit queue= argument.
2. Urgent beats a backlog. Enqueue 50,000 bulk tasks, wait for the bulk queue to build, then enqueue one urgent task and measure end-to-end latency. It must complete in seconds.
3. Prefetch is not hoarding. With the bulk pool running, rabbitmqctl list_queues name messages_ready messages_unacknowledged (or redis-cli LLEN) should show unacknowledged counts near concurrency, not multiples of it.
4. A restart does not lose work. Restart the urgent pool mid-task; with acks_late, the task must be redelivered and complete.

Gotchas & Edge Cases

task_routes is producer-side. The routing decision happens where apply_async is called, so a producer with stale config publishes to the old queue. Deploy configuration to producers and consumers together, and prefer computing the route from a shared module over duplicating the table.

Queue arguments are immutable. Adding x-max-priority to an existing RabbitMQ queue raises PRECONDITION_FAILED. Create a new queue name, move consumers, then drain and delete the old one.

A worker with no -Q consumes only the default queue. After renaming task_default_queue, workers started without -Q silently consume nothing. Always pass -Q explicitly in production.

Autoscalers must be per-pool. A single HPA across a deployment that serves all three queues scales on aggregate depth, so a bulk backlog adds urgent-pool replicas that have nothing to do while the bulk queue stays behind. Give each pool its own deployment and its own scaling metric, as covered in horizontal worker scaling.

Priority does not preempt. A worker already executing a 20-minute bulk task will not drop it for an urgent one. Only reserved capacity fixes that, which is why the dedicated pool matters more than any priority setting.

celery_queue_length needs an exporter that understands your broker. For Redis, queue length is LLEN of the queue key; for RabbitMQ it is the management API. Getting this wrong produces a dashboard that shows zero during an incident — see instrumenting Celery with a Prometheus exporter.

Related

• Priority Queues & Job Fairness — the scheduling theory behind these settings.
• Implementing priority queues with Redis sorted sets — per-message ordering when you control the queue implementation.
• Preventing tenant starvation with weighted queues — fairness across customers, which routing alone does not provide.
• Celery Architecture & Configuration — brokers, pools, and the wider configuration surface.