怎样在Python中用Celery配合RabbitMQ实现任务重试与死信队列-Python教程-PHP中文网

celery 的 retry 机制默认不走 rabbitmq 死信队列，因其 retry() 在 worker 内部重发新消息，不经过 amqp 协议层，故无法触发 ttl+dlx；需手动配置带 x-dead-letter-exchange 的队列并用 send_task() 发送带 expiration 的消息。

怎样在python中用celery配合rabbitmq实现任务重试与死信队列

为什么 Celery 的 retry 机制默认不走 RabbitMQ 死信队列

Celery 自带的 retry() 是在客户端（worker 进程内）重入任务队列，不经过 Broker 的 AMQP 协议层重投，因此不会触发 RabbitMQ 的 TTL + DLX 死信路由。你看到任务重试了，但 RabbitMQ 根本不知道——它只看到一个新发布的消息。真要让 RabbitMQ 参与重试生命周期（比如超时后自动进死信队列），必须绕过 retry()，改用原生 AMQP 控制：手动发布带 expiration 的消息，并配置队列的 x-dead-letter-exchange 参数。

如何声明支持死信的 Celery 队列（RabbitMQ 侧）

不能只靠 CELERY_TASK_ROUTES 或 task_routes；必须显式声明队列并绑定 DLX 属性。Celery 本身不自动设置 x-dead-letter-exchange，得在 app.conf.task_queues 或启动时用 Queue 对象手动定义：

from kombu import Queue, Exchange
<p>app.conf.task_queues = {
'default': {
'exchange': 'default',
'routing_key': 'default',
},
'retry_queue': Queue(
'retry_queue',
exchange=Exchange('retry_exchange'),
routing_key='retry',</p><h1>关键：让过期消息被转发到 dlx</h1><pre class="brush:python;toolbar:false;">    queue_arguments={
        'x-dead-letter-exchange': 'dlx',
        'x-dead-letter-routing-key': 'dead.letter',
        'x-message-ttl': 60000,  # 60秒后过期
    }
),

}

Sakura python draw

使用Python的turtle和random库，递归绘制分形樱花树，并动画模拟花瓣自然飘落效果。

下载

必须确保 dlx 这个 exchange 已存在（可提前用 rabbitmqctl 创建，或由 Celery worker 自动声明）
x-message-ttl 设太短（如 1000ms）会导致正常慢任务也被误判为失败
如果用 task_routes 指向该队列，记得 route key 要匹配 routing_key

怎样让失败任务自动发往带 TTL 的重试队列（而非调用 retry()）

核心是放弃 self.retry()，改用 app.send_task() 手动发消息，并带上 AMQP headers 和 expiration：

@app.task(bind=True, autoretry_for=(Exception,), retry_kwargs={'max_retries': 0})
def risky_task(self):
    try:
        # 实际逻辑
        raise ValueError("boom")
    except Exception as exc:
        # 不调用 self.retry()
        self.app.send_task(
            'risky_task',
            args=self.args,
            kwargs=self.kwargs,
            queue='retry_queue',
            countdown=5,  # 这里只是建议延迟，真正生效靠 x-message-ttl
            headers={'x-death': []},  # 避免被当成已死信循环投递
        )
        raise Ignore()  # 显式终止当前任务，不进 result backend

autoretry_for 必须设 max_retries=0，否则 Celery 会先走自己的 retry 流程，和你的手动逻辑冲突
countdown 在 RabbitMQ 场景下只是“尽力而为”，真正控制重试间隔的是队列级 x-message-ttl + 消费者 requeue 行为
Ignore() 很关键：防止任务状态变成 FAILURE，干扰监控；同时避免重复记录 result

死信队列里的任务怎么查、怎么人工干预

RabbitMQ 不会自动把死信写进 Celery 的 result backend，它们就静静躺在 dead.letter 绑定的队列里。你需要额外步骤处理：

用 rabbitmqadmin 查看死信队列长度：rabbitmqadmin list queues name messages | grep dead.letter
用 celery inspect scheduled 看不到这些任务——它们已脱离 Celery 调度上下文
人工重放：写个脚本从死信队列消费，解析 body（是 JSON 序列化的 Celery message），再调用 app.send_task() 重新入正常队列
注意死信消息的 redelivered 字段为 True，可用于区分首次投递和死信回流

真正的难点不在配置，而在于死信之后的可观测性断层：Celery 的 Flower、Prometheus exporter 都看不到这些消息。你得在 RabbitMQ 层加监控告警，或者在消费者端对 x-death header 做日志标记。