重试、幂等性与死信处理
使用指数退避、幂等键和死信路由增强任务的韧性，应对有问题的消息。
重试、幂等性与死信处理 是 CoddyKit 上的免费 FastAPI Backend Development Bootcamp 课时。 这是第 3 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 FastAPI Backend Development Bootcamp 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 FastAPI Backend Development Bootcamp 课程共包含 4 节课。
本课时的部分内容尚未翻译，以英文显示。
Why Tasks Need Resilience
In a FastAPI backend, you push slow work (sending emails, charging cards, calling third-party APIs) to a Celery worker so the HTTP request returns fast. But background work runs in a hostile world: networks blip, APIs rate-limit you, and workers crash mid-task.
A resilient task must survive three failure modes:
- Transient failures — retry with exponential backoff so you do not hammer a struggling service.
- Duplicate delivery — the same message may be processed twice, so tasks must be idempotent.
- Poisoned messages — a task that fails forever must be routed to a dead-letter queue instead of looping endlessly.
This lesson wires all three together.
At-Least-Once Delivery
Celery brokers (RabbitMQ, Redis) give you at-least-once delivery, not exactly-once. A message is acknowledged (ack) only after the task finishes. If a worker dies after doing work but before acking, the broker redelivers the message and the task runs again.
With acks_late=True the ack happens after execution — safer against crashes, but it guarantees that some tasks will run twice. That is the core reason idempotency is not optional.
from celery import Celery
app = Celery("jobs", broker="redis://localhost:6379/0")
# Recommended resilience defaults
app.conf.update(
task_acks_late=True, # ack only after the task body returns
task_reject_on_worker_lost=True, # requeue if the worker is killed
worker_prefetch_multiplier=1, # don't hoard messages on one worker
)Automatic Retries with autoretry_for
The simplest way to retry is to declare which exceptions are retryable. Celery catches them and re-schedules the task automatically.
autoretry_for— the exception classes that trigger a retry.max_retries— the cap before the task is marked failed.retry_backoff— turns on exponential backoff (delays double each attempt).
Only retry on transient errors (timeouts, 5xx, connection resets). Never blindly retry a ValueError from bad input — it will fail identically every time.
import requests
from celery import Celery
app = Celery("jobs", broker="redis://localhost:6379/0")
@app.task(
autoretry_for=(requests.exceptions.RequestException,),
max_retries=5,
retry_backoff=True, # 1s, 2s, 4s, 8s, ...
retry_backoff_max=600, # cap the delay at 10 minutes
retry_jitter=True, # randomize to avoid thundering herd
)
def call_payment_api(charge_id: str):
resp = requests.post("https://api.example.com/charge", json={"id": charge_id}, timeout=10)
resp.raise_for_status()
return resp.json()Exponential Backoff, Explained
Exponential backoff means the wait between attempts grows geometrically: the delay for attempt n is roughly base * 2 ** n, capped at a maximum. This gives a struggling downstream service time to recover instead of being retried into the ground.
Jitter adds randomness so that thousands of tasks that failed at the same instant do not all retry at the same instant (the “thundering herd”). Here is the math Celery applies, as a plain standalone program.
import random
def backoff_delay(attempt, base=1, cap=600, jitter=True):
delay = min(cap, base * (2 ** attempt))
if jitter:
delay = random.uniform(0, delay) # full jitter
return delay
for attempt in range(8):
raw = min(600, 1 * (2 ** attempt))
print(f"attempt {attempt}: raw={raw:>3}s jittered~={backoff_delay(attempt):6.1f}s")Manual Retry with self.retry
When you need custom logic — deciding the delay from a response header, or only retrying on certain status codes — bind the task and call self.retry() explicitly.
Use bind=True to get self, read self.request.retries to know the attempt number, and pass countdown for the delay. Raising the result of self.retry() stops the current execution cleanly.
import requests
from celery import Celery
app = Celery("jobs", broker="redis://localhost:6379/0")
@app.task(bind=True, max_retries=5)
def sync_inventory(self, sku: str):
resp = requests.get(f"https://api.example.com/stock/{sku}", timeout=5)
if resp.status_code == 429: # rate limited
wait = int(resp.headers.get("Retry-After", 2 ** self.request.retries))
raise self.retry(countdown=wait)
resp.raise_for_status()
return resp.json()["qty"]What Idempotency Really Means
An operation is idempotent if running it twice has the same effect as running it once. Because Celery delivers at-least-once, every task that mutates state (charging a card, creating a record, decrementing stock) must be idempotent or you will double-charge customers.
The standard tool is an idempotency key: a unique identifier for the business intent, not for the message. You record “I have already processed key X” in durable storage and short-circuit on the next delivery.
- Pass the key in from the API request (clients can supply it too).
- Store it in a table or Redis with a
UNIQUEconstraint. - The constraint — not application logic — is what makes it race-safe.
An Idempotency Guard
Here is the pattern in isolation: a guard that remembers completed keys and refuses to run the effect twice, even under concurrent calls. In real code the seen set becomes a Redis SET NX or a DB row with a unique key, but the logic is identical.
import threading
class IdempotencyGuard:
def __init__(self):
self._seen = set()
self._lock = threading.Lock()
def run_once(self, key, effect):
with self._lock: # the unique-constraint stand-in
if key in self._seen:
return "skipped (duplicate)"
self._seen.add(key)
return effect()
guard = IdempotencyGuard()
charges = []
def charge():
charges.append(99)
return "charged 99"
print(guard.run_once("order-123", charge))
print(guard.run_once("order-123", charge)) # duplicate delivery
print("total charges applied:", len(charges))Idempotency Inside a Celery Task
In practice you wrap the effect in a database transaction and let a UNIQUE constraint be the source of truth. Insert the idempotency key first; if the insert raises a unique-violation, a previous (or concurrent) delivery already handled it, so you return early.
This keeps the check and the side effect atomic — there is no window where one delivery sees “not done” while another is mid-charge.
from sqlalchemy.exc import IntegrityError
from celery import Celery
app = Celery("jobs", broker="redis://localhost:6379/0")
@app.task(bind=True, acks_late=True, max_retries=3, retry_backoff=True)
def charge_order(self, order_id: str, idem_key: str):
with db_session() as s:
try:
s.add(ProcessedKey(key=idem_key)) # UNIQUE column
s.flush() # raises on duplicate
except IntegrityError:
s.rollback()
return {"status": "already_processed", "order_id": order_id}
amount = payment_gateway.charge(order_id, idempotency_key=idem_key)
s.commit()
return {"status": "charged", "amount": amount}Poisoned Messages and the Dead-Letter Queue
Some messages can never succeed: malformed payloads, references to deleted rows, a bug that always throws. Retrying them forever wastes workers and floods your logs. These are poisoned messages.
A dead-letter queue (DLQ) is a separate queue where exhausted or rejected messages are parked for inspection, alerting, or manual replay. With RabbitMQ you declare a queue with a x-dead-letter-exchange argument; messages that are rejected (nack with requeue=False) or that exceed a TTL get routed there automatically by the broker.
from kombu import Exchange, Queue
dead_exchange = Exchange("dlx", type="direct")
task_queues = (
Queue(
"payments",
Exchange("payments"),
routing_key="payments",
queue_arguments={
"x-dead-letter-exchange": "dlx",
"x-dead-letter-routing-key": "payments.dead",
},
),
Queue("payments_dead", dead_exchange, routing_key="payments.dead"),
)Routing Exhausted Tasks to the DLQ
The broker dead-letters on reject, but Celery's retry machinery does not auto-reject when max_retries is hit — it just marks the task FAILED. To send exhausted tasks to a DLQ you catch MaxRetriesExceededError (or detect the final attempt) and explicitly forward the payload to your dead-letter task or queue.
The dead-letter handler should never reprocess — it records the failure, emits an alert, and stores the payload so an operator can replay it after fixing the root cause.
from celery import Celery
from celery.exceptions import MaxRetriesExceededError
app = Celery("jobs", broker="redis://localhost:6379/0")
@app.task(bind=True, max_retries=5, retry_backoff=True)
def process_event(self, payload: dict):
try:
do_work(payload)
except TransientError as exc:
try:
raise self.retry(exc=exc)
except MaxRetriesExceededError:
dead_letter.delay(payload, reason=str(exc)) # park it
except PermanentError as exc:
dead_letter.delay(payload, reason=str(exc)) # never retry
@app.task
def dead_letter(payload: dict, reason: str):
store_failed_message(payload, reason)
alert_oncall(reason)Putting It All Together
A production-grade resilient task combines every piece:
- acks_late so crashes redeliver instead of losing work.
- autoretry_for on transient errors only, with exponential backoff + jitter.
- An idempotency key guarded by a unique constraint so redelivery is harmless.
- A dead-letter path for permanent errors and exhausted retries.
The mental model: retry the transient, deduplicate the duplicate, dead-letter the doomed. Each mechanism covers a different failure mode — together they let a worker fail safely instead of silently corrupting data.
@app.task(
bind=True, acks_late=True,
autoretry_for=(TransientError,),
max_retries=5, retry_backoff=True, retry_jitter=True,
)
def handle_webhook(self, payload: dict, idem_key: str):
if already_processed(idem_key): # unique-constraint check
return "duplicate-ignored"
try:
result = apply_effect(payload, idem_key)
except PermanentError as exc:
dead_letter.delay(payload, reason=str(exc))
return "dead-lettered"
mark_processed(idem_key)
return resultQuick Check
Your Celery task charges a credit card and runs with acks_late=True. Because the broker delivers at-least-once, the same message is occasionally processed twice. What is the correct primary defense against double-charging?
Recap
You learned how to make Celery tasks survive real-world failure:
- Celery delivers at-least-once;
acks_late=Trueprotects against worker crashes but guarantees occasional duplicate runs. - Exponential backoff with jitter (via
retry_backoff/autoretry_foror manualself.retry) handles transient failures without overwhelming downstream services — retry only transient errors. - Idempotency keys backed by a unique constraint make duplicate deliveries harmless and keep the check atomic with the side effect.
- Dead-letter queues park poisoned messages and exhausted retries for alerting and manual replay, instead of looping forever.
Remember the rule: retry the transient, deduplicate the duplicate, dead-letter the doomed.
常见问题解答
「重试、幂等性与死信处理」课时是免费的吗？
是的 — 「重试、幂等性与死信处理」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 FastAPI Backend Development Bootcamp 课程的其余内容，请升级到 CoddyKit PRO。 FastAPI Backend Development Bootcamp 课程共包含 4 节课。
「重试、幂等性与死信处理」这节课中我会学到什么？
使用指数退避、幂等键和死信路由增强任务的韧性，应对有问题的消息。 你通过在浏览器中直接运行的动手代码来练习 FastAPI Backend Development Bootcamp，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 FastAPI Backend Development Bootcamp 需要有经验吗？
无需任何先前经验。CoddyKit 上的 FastAPI Backend Development Bootcamp 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 3 节课，共 4 节。
「重试、幂等性与死信处理」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 FastAPI Backend Development Bootcamp 课中编写并运行代码吗？
能。每节 FastAPI Backend Development Bootcamp 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。