Your consumer raises on one message. What happens next depends on the library, and the two most common Python choices behave in opposite ways: a plain pika consumer gets the same message back again and again, a Celery worker with default settings acknowledges it and moves on, so the failed message is gone. Neither ends up in a dead-letter queue unless you set that up. Here is what each one does, how to make failures land in a dead-letter queue, and how to send them back afterwards.
A typical BlockingConnection consumer acknowledges manually (pika's default since 1.0) and acks at the end of the callback. If the callback raises, the exception leaves start_consuming(), the process usually exits or reconnects, and the broker requeues every unacknowledged message. The poison message is delivered again, the consumer raises again, and the loop starts over.
On a classic queue that loop has no end. On a quorum queue in RabbitMQ 4 it ends after 20 deliveries, the default delivery limit, and the message is then dropped, or dead-lettered if the queue has a dead-letter exchange. Either way you did not decide it.
Catch the failure in the callback and reject the message without requeue:
import logging, pika
log = logging.getLogger("orders")
conn = pika.BlockingConnection(pika.URLParameters("amqp://guest:guest@localhost:5672/%2F"))
ch = conn.channel()
# the dead-letter route: an exchange and a queue that collects what failed
ch.exchange_declare("orders.dlx", "direct", durable=True)
ch.queue_declare("orders.dlq", durable=True)
ch.queue_bind("orders.dlq", "orders.dlx", "orders.dlq")
ch.queue_declare("orders.process", durable=True, arguments={
"x-dead-letter-exchange": "orders.dlx",
"x-dead-letter-routing-key": "orders.dlq",
})
def on_message(ch, method, props, body):
try:
handle(body)
except Exception:
log.exception("order failed, dead-lettering %s", props.message_id)
ch.basic_nack(method.delivery_tag, requeue=False)
return
ch.basic_ack(method.delivery_tag)
ch.basic_qos(prefetch_count=20)
ch.basic_consume("orders.process", on_message)
ch.start_consuming()requeue=False hands the message to the queue's dead-letter exchange; RabbitMQ adds an x-death header that names the queue, the reason (rejected), the time, and the original exchange and routing key. Queue arguments are fixed when the queue is created: if orders.process already exists without them, queue_declare fails with PRECONDITION_FAILED. For existing queues set a policy instead, which can change at any time:
rabbitmqctl set_policy orders-dlx '^orders\.process$' \
'{"dead-letter-exchange":"orders.dlx","dead-letter-routing-key":"orders.dlq"}' \
--apply-to queuesTwo more traps with pika:
auto_ack=True acknowledges on delivery. A failure afterwards loses the message, and nothing reaches a dead-letter queue.BlockingConnection's I/O loop, so heartbeats stop. After the heartbeat timeout the broker closes the connection and requeues the message: a slow message looks like a failing one. Run long work in a thread and ack through conn.add_callback_threadsafe.Not every failure is a bug: a timeout or a database failover passes. A retry with a delay that the broker holds, not your process, is a wait queue with a TTL whose dead-letter exchange points back at the work exchange. The consumer counts the rounds in x-death and gives up after a few:
MAX_ROUNDS = 5
def rounds(props, queue):
for death in (props.headers or {}).get("x-death", []):
if death["queue"] == queue and death["reason"] == "rejected":
return death["count"]
return 0
def on_message(ch, method, props, body):
try:
handle(body)
ch.basic_ack(method.delivery_tag)
except TemporaryError:
if rounds(props, "orders.process") >= MAX_ROUNDS:
park(ch, props, body) # publish to orders.dlq with confirms, then ack
ch.basic_ack(method.delivery_tag)
else:
ch.basic_nack(method.delivery_tag, requeue=False) # to orders.retry.wait, back in 30 sHere orders.process dead-letters into a wait queue orders.retry.wait (x-message-ttl: 30000, no consumers) whose own dead-letter exchange routes back to orders.process. RabbitMQ increments the count of the same x-death entry on every round, which is what makes the counter work. The x-death decoder shows such a loop in plain words if you paste one of these messages.
Celery acknowledges a task message when the worker receives it (task_acks_late = False is the default). If the task raises, Celery stores the failure in the result backend, logs it, and the message is already gone from RabbitMQ. If the worker is killed mid-task, the same: the message was acked. There is nothing to replay.
To keep failed tasks in RabbitMQ, ack late and reject on the final failure:
from celery import Celery
from celery.exceptions import Reject
app = Celery("orders", broker="amqp://guest:guest@localhost:5672//")
app.conf.task_acks_late = True # ack after the task returns
app.conf.task_reject_on_worker_lost = True # a killed worker requeues instead of acking
@app.task(bind=True, max_retries=3)
def process_order(self, order_id):
try:
handle(order_id)
except TemporaryError as exc:
if self.request.retries >= self.max_retries:
raise Reject(f"gave up after {self.request.retries} retries: {exc}", requeue=False)
raise self.retry(exc=exc, countdown=2 ** self.request.retries)
except Exception as exc:
raise Reject(str(exc), requeue=False)Reject(requeue=False) sends a basic.reject without requeue, so the message goes to the queue's dead-letter exchange. Give Celery's queue one with the policy above ('^celery$' for the default queue); declaring it through queue_arguments on an existing queue fails the same way as with pika.
Three Celery details matter here:
self.retry() is a new publish. The retried message is a copy with the retries header increased; the original is acknowledged. Retries therefore never show in x-death, only the final reject does.consumer_timeout (30 minutes by default), so long countdowns end in PRECONDITION_FAILED. Keep countdowns short, or use a broker-side wait queue as above.acks_late, Reject has no effect. The message was acked on arrival.The body is Celery's protocol: a JSON array of [args, kwargs, embed]. The headers carry task (the task name), id, retries, argsrepr and kwargsrepr, next to the broker's x-death. The x-death entry's exchange and routing key are where Celery published it: with default settings exchange celery and routing key celery, otherwise whatever your task routes say.
To run the task again, publish the message unchanged to that exchange and routing key: same body, same task and id headers, with x-death and the other death headers removed. A worker picks it up as if it were new. Keep the body exactly as it is; Celery checks nothing, but your task gets whatever arguments the body holds.
The loop is the same in both cases: read from the dead-letter queue, publish to the original route with publisher confirms, acknowledge the dead-lettered copy only after the confirm.
DEATH = ("x-death", "x-first-death-", "x-last-death-", "x-delivery-count", "x-acquired-count")
ch.confirm_delivery()
while True:
method, props, body = ch.basic_get("orders.dlq", auto_ack=False)
if method is None:
break
death = props.headers["x-death"][0]
props.headers = {k: v for k, v in props.headers.items() if not k.startswith(DEATH)}
ch.basic_publish(death["exchange"], death["routing-keys"][0], body, props, mandatory=True)
ch.basic_ack(method.delivery_tag)Replay a few first. If the consumer still fails on them, they come back with a fresh x-death, and the rest are still where they were. Replay one message first has the details.
basic_nack(requeue=False) on failure, and a dead-letter exchange on the queue.BlockingConnection thread; heartbeats must keep going.task_acks_late and task_reject_on_worker_lost on, Reject(requeue=False) on the final failure.consumer_timeout.x-death, death headers stripped, one message first.