Celery 任务退避抖动重试

示例说明

消息队列任务的失败多为下游瞬态，重试策略决定恢复速度与下游压力的平衡。指数退避让间隔随次数翻倍并封顶，抖动错开多任务的同步重试波峰。acks_late 加单条预取的组合意味着：任务执行完才确认消息，worker 崩溃时任务重新投递，不丢不重由幂等性兜底。

技术要点

• 退避延迟按重试次数指数增长并封顶
• 抖动打散多任务同步重试
• acks_late 保证 worker 崩溃不丢任务

代码解读

book_shipment 声明最大五次重试与自动重试异常类型，任务内随机抛出承运商不可用异常；except 分支手工计算含抖动的倒计时后调用 self.retry。autoretry_for 与手工 retry 二选一即可，这里双轨展示两种写法。stats 任务用 inspect 汇聚集群活跃与排队任务数。

使用建议

任务必须幂等，重试与重投递都会造成重复执行。抖动比例按任务并发规模调。inspect 调用有性能开销，监控别打太频繁。

代码示例

import random

from celery import Celery

app = Celery('shipping', broker='redis://localhost:6379/1')
app.conf.update(
    task_acks_late=True,
    worker_prefetch_multiplier=1,
    task_reject_on_worker_lost=True,
)


class CarrierUnavailable(Exception):
    pass


@app.task(
    bind=True,
    max_retries=5,
    autoretry_for=(CarrierUnavailable,),
    retry_backoff=2,
    retry_backoff_max=600,
    retry_jitter=True,
)
def book_shipment(self, order_id: str, carrier: str):
    try:
        if random.random() < 0.5:
            raise CarrierUnavailable(f'{carrier} api warming up')
        return {'order': order_id, 'carrier': carrier, 'tracking': 'TRK-88'}
    except CarrierUnavailable as exc:
        countdown = min(2 ** self.request.retries * 2, 600)
        countdown += random.uniform(0, countdown * 0.3)
        raise self.retry(exc=exc, countdown=countdown)


@app.task
def shipment_stats():
    inspector = app.control.inspect()
    active = inspector.active() or {}
    reserved = inspector.reserved() or {}
    return {
        'active_tasks': sum(len(w) for w in active.values()),
        'reserved_tasks': sum(len(w) for w in reserved.values()),
    }