Celery Task Routing and Reliability

How you split the queues decides throughput, and when you send the acknowledgement decides whether you lose work or run it twice.

When Everything Goes Into One Queue

Celery is the distributed task queue most widely used in Python: a producer puts a message on a broker and a worker pulls it off and runs it. The default configuration puts every task on a single queue. When sending an email and encoding a video stand in the same line, a video encode that takes tens of seconds blocks the short mail job behind it.

The starting point is to split queues by the nature of the work. Put short, frequent tasks and long, heavy tasks on different queues, and attach separate workers to each. Then a backlog on one side does not spread its delay to the other.

How an Exchange Routes Messages to Queues

Queue separation is implemented as routing rules. Celery follows the Advanced Message Queuing Protocol (AMQP) model, where a message does not go straight to a queue but passes through an exchange. The exchange looks at the routing key and the binding rules to decide which queue receives it.

Exchanges come in four kinds by routing behavior, and the kind you use determines the rule for picking a queue.

In a topic exchange, feed.# is a binding pattern that accepts every key beginning with feed.. When publishing you use a concrete key such as feed.sync. Queue definitions and routing rules are configured like this.

from celery import Celery
from kombu import Exchange, Queue
 
app = Celery('myapp')
 
default_exchange = Exchange('tasks', type='direct')
feed_exchange = Exchange('feeds', type='topic')
 
app.conf.task_queues = (
    Queue('default', exchange=default_exchange, routing_key='celery'),
    Queue('priority', exchange=default_exchange, routing_key='priority',
          queue_arguments={'x-max-priority': 10}),
    Queue('feed_tasks', exchange=feed_exchange, routing_key='feed.#',
          durable=True),
)
 
app.conf.task_routes = {
    'myapp.tasks.high_priority': {'queue': 'priority', 'routing_key': 'priority'},
    'feeds.*': {'queue': 'feed_tasks', 'routing_key': 'feed.sync'},
}

task_routes maps task name patterns to queues and supports both wildcards and function-based routers. You can also name the queue directly in code with apply_async(queue=...), but keeping the rules in configuration means the call site does not need to know queue names.

Ack Timing: Loss on One Side, Duplication on the Other

Routing is a throughput problem; acknowledgement (ack) is a reliability problem. A worker sends an ack to the broker to say it has processed the message, and the broker removes the message from the queue only once that ack arrives. When the ack is sent splits into two options.

Early ack, the default, sends the ack as soon as the message is received. If the worker dies mid-execution, that task is gone. Late ack sends the ack after execution completes, so on a crash the broker redelivers the same message to another worker.

app.conf.task_acks_late = True
app.conf.task_reject_on_worker_lost = True
app.conf.worker_prefetch_multiplier = 1

task_acks_late=True prevents loss, and task_reject_on_worker_lost=True requeues the message even when the worker terminates abnormally. The price is that duplicate execution has to be accepted.

Absorbing Duplicate Execution With Idempotency

Turning on late ack makes the delivery guarantee at-least-once, which means the same message can run twice. Duplication is normal in a distributed queue, and it has more than one cause.

Unlike RabbitMQ, a Redis broker does not have AMQP acknowledgements as such, so it emulates them with a visibility timeout. Once a worker takes a message, the broker republishes it if no ack arrives within the timeout (3,600 seconds by default). Long tasks whose processing time exceeds the timeout therefore carry a high duplication risk.

app.conf.broker_transport_options = {
    'visibility_timeout': 3600,  # Redis only, in seconds
}

Since duplication cannot be eliminated, design for idempotency so that running the same work several times produces the same result. The most reliable approach is a database unique constraint. Store the task_id as the idempotency key, and when a second execution hits IntegrityError, return the existing result.

from django.db import IntegrityError, transaction
 
@app.task(bind=True, acks_late=True)
def process_payment(self, order_id):
    try:
        with transaction.atomic():
            payment = Payment.objects.create(
                order_id=order_id,
                idempotency_key=self.request.id,  # use task_id as the key
                amount=calculate_amount(order_id),
            )
            charge_credit_card(payment)
            return {'status': 'success', 'payment_id': payment.id}
    except IntegrityError:
        existing = Payment.objects.get(idempotency_key=self.request.id)
        return {'status': 'duplicate', 'payment_id': existing.id}

Retries rest on the same premise. autoretry_for catches the exception and retry_backoff=True grows the retry interval exponentially: 1 second, 2 seconds, 4 seconds. Attaching retries to a task that is not idempotent piles up partial executions and produces more duplication, not less.

@app.task(bind=True, autoretry_for=(Exception,), retry_backoff=True, max_retries=5)
def fetch_url(self, url: str) -> str:
    import requests
    r = requests.get(url, timeout=10)
    r.raise_for_status()
    return r.text[:2000]

Prefetch and Serialization

A worker fetches several messages ahead of time. worker_prefetch_multiplier sets how many, and the default of 4 is tuned for short tasks. For long tasks, lower it to 1 so that work reserved by one worker does not starve the others.

Serialization is directly a security matter. The default, JSON, carries only data and is safe. Pickle carries Python objects as-is and creates a risk of arbitrary code execution, so block it with accept_content=['json'] and do not run workers as root. Prefetch and serialization govern throughput and safety respectively, but they belong in the same decision as routing, ack, and idempotency.

Summary

Queue separation decides throughput, and ack timing decides reliability. Preventing loss with late ack means accepting duplicate execution in exchange, and idempotency plus exponential-backoff retries are what absorb that duplication. With Redis, the visibility timeout becomes another source of duplication on long tasks, so tune it together with expected task duration. Routing, ack, idempotency, prefetch, and serialization are not independent settings but one reliability bundle to be decided together.