在分布式任务队列中，任务执行失败是家常便饭。很多开发者以为，给Celery任务加上 autoretry_for 参数，就能自动实现指数退避重试。这里有个常见的误解：autoretry_for 只负责“抛出重试”这个动作，至于“等多久再试”，它可不管。要实现真正的阶梯式延迟重试，你必须手动计算 countdown，或者启用另一个关键参数：retry_backoff。
autoretry_for + retry_backoff：开箱即用的阶梯重试方案
对于网络请求超时、数据库连接中断这类临时性错误，这个组合是最省心的选择。它把复杂的指数退避逻辑封装在框架层，你不需要在任务函数里手动调用 self.retry()。
具体怎么用？几个参数是关键：
retry_backoff=2：这表示第一次重试等待2秒，第二次等待4秒，第三次等待8秒……延迟时间按2 ** n的指数增长。retry_jitter=True：这个参数默认是开启的，它会为每次重试延迟加上一个随机偏移量。好处是避免大量失败任务在同一时刻集体重试，对下游服务造成“惊群效应”。max_retries=5：这是重试次数的硬性上限。超过这个次数，任务就会彻底失败，并抛出MaxRetriesExceededError异常。- 需要特别注意：
autoretry_for只会捕获你指定的异常类型。像ValueError这类通常代表程序逻辑错误的异常，是不会触发自动重试的。
来看一个实际的代码示例：
@shared_task(
bind=True,
autoretry_for=(requests.RequestException, redis.ConnectionError),
retry_backoff=2,
retry_jitter=True,
max_retries=4)
def fetch_data(self, url):
return requests.get(url, timeout=8).json()
手动self.retry()：实现精细化退避控制
当你的重试策略需要更复杂的逻辑时，比如根据不同的错误类型设置不同的延迟，或者动态调整最大等待时间，autoretry_for 就显得力不从心了。这时，就必须回到手动模式，使用 bind=True 来获取任务上下文。
手动控制有几个技术要点：
- 必须设置
bind=True，这样才能通过self.request.retries获取当前是第几次重试，这是计算指数延迟的基础。 countdown参数的单位是秒，而且必须是一个数字，不能直接传入datetime.timedelta对象。- 切记不要用全局变量或者外部缓存来记录重试次数。Celery的Worker是无状态的，每次任务执行都是全新的上下文。
- 强烈建议为延迟时间设置一个上限（比如
max_delay）。否则，按照指数增长，第10次重试可能要等1024秒（超过17分钟），这在实际业务中往往是不可接受的。
下面是一个兼顾安全性和灵活性的写法：
@app.task(bind=True)
def send_email(self, user_id):
try:
send_mail_to_user(user_id)
except SMTPServerDisconnected as exc:
base = 2 ** self.request.retries
countdown = min(base, 60) # 设置上限，最多等待60秒
raise self.retry(exc=exc, countdown=countdown)
任务确认机制：决定重试能否生效的关键
很多情况下，任务明明配置了重试，却只执行了一次就默默失败了。问题往往不出在重试逻辑本身，而是底层的任务确认（ACK）机制。
task_acks_late=True：这个设置意味着，任务只有在执行完成后，Celery才会向消息袋里（如RabbitMQ）发送确认信号。如果Worker进程在执行中途崩溃，任务会被袋里重新投递给其他Worker。如果设为False（默认值），任务一被Worker取走就会发送ACK，一旦Worker崩溃，任务就永久丢失了。task_reject_on_worker_lost=True：当Worker进程被强制终止或意外崩溃时，这个设置会让任务被“拒收”并重新放回队列。不过，这个特性需要消息袋里的支持（RabbitMQ可以，但Redis不支持）。- 如果你使用Redis作为Broker，
task_reject_on_worker_lost是无效的。这时需要依赖Redis的visibility_timeout和任务重入队逻辑来作为兜底方案。
策略配置化：别把参数写死在代码里
线上环境想调整重试间隔，难道每次都要修改代码、重新部署？显然不划算。更优雅的做法是把重试策略参数化，从配置文件或环境变量中读取。
例如，在Django项目中，可以在 settings.py 中统一管理：
CUSTOM_RETRY_POLICY = {
'max_retries': int(os.getenv('CELERY_MAX_RETRIES', '3')),
'base_delay': float(os.getenv('CELERY_BASE_DELAY', '1')),
'max_delay': int(os.getenv('CELERY_MAX_DELAY', '120')),
}
在任务函数中，直接读取这些配置，实现策略与业务逻辑的解耦：
@app.task(bind=True)
def call_third_api(self, payload):
cfg = getattr(settings, 'CUSTOM_RETRY_POLICY', {})
try:
requests.post('https://api.example.com', json=payload, timeout=5)
except Exception as exc:
delay = min(cfg['base_delay'] * (2 ** self.request.retries), cfg['max_delay'])
raise self.retry(exc=exc, countdown=delay)
最后，必须清醒地认识到：重试机制是一种补救措施，而不是设计缺陷的遮羞布。如果任务本身存在数据库事务未正确回滚、缺乏幂等性设计，或者下游服务根本就是不可用状态，那么盲目重试只会让问题雪上加霜。正确的思路是，先确保单次任务执行足够健壮，然后再来讨论如何优雅地重试。
