Celery worker重启后任务丢失或重复执行？

Question

Celery worker 重启时，若未正确配置任务持久化与消费者确认机制，极易引发两类典型问题：**任务丢失**与**重复执行**。当 broker（如 RabbitMQ 或 Redis）未启用消息持久化（`delivery_mode=2`），且 worker 异常退出前未完成 ACK，未被消费的消息可能在重启后直接丢弃；反之，若 worker 在处理完任务但尚未发送 ACK 时崩溃（如进程被 `kill -9`、OOM Killer 终止或网络中断），broker 会因超时重发该任务，导致下游重复执行（尤其在无幂等设计的支付、库存扣减等场景中后果严重）。此外，使用 `acks_late=True` 但未配合 `reject_on_worker_lost=True`（Celery ≥5.0）或 `task_reject_on_worker_lost=True`（旧版），亦会加剧重复风险。根本症结在于消息语义保障（At-Least-Once vs Exactly-Once）与应用层幂等性的错配，而非 Celery 本身缺陷。

Answer 1

```html

一、现象层：典型故障表征与线上告警信号

• Worker进程异常退出后，监控平台（如Prometheus + Grafana）显示celery_worker_tasks_received_total突降但celery_worker_tasks_failed_total无显著上升——暗示任务静默丢失；
• RabbitMQ Management UI 中队列Messages Ready为0，但Messages Unacknowledged持续非零且缓慢增长，重启Worker后该值归零——暴露ACK未及时发出；
• 支付系统日志中出现同一order_id在5分钟内触发两次pay_process任务，数据库中生成两条重复扣款记录；
• Redis作为Broker时，LRANGE celery queue 0 -1返回空，但业务方坚称已调用apply_async()——因delivery_mode=1（非持久化）导致消息随broker重启蒸发。

二、机制层：Celery消息生命周期与语义保障断点分析

下图展示Celery任务从发布到执行的完整链路及关键确认点：

graph LR
A[Producer: apply_async] -->|1. delivery_mode=2?| B[RabbitMQ/Redis Broker]
B -->|2. 消息入队+持久化落盘| C[Worker Fetch]
C -->|3. prefetch_count限制| D[Task in Memory]
D -->|4. acks_late=True? → ACK时机延后至task.return_value| E[Task Execution]
E -->|5. 进程崩溃于return前 → broker重发| F[Duplicate Delivery]
F -->|6. 无幂等校验 → DB双写| G[业务一致性破坏]

三、配置层：核心参数对照表与危险组合警示

配置项	安全值（推荐）	高危值（禁用）	影响范围
broker_transport_options={'delivery_mode': 2}	✅ 必须启用	❌ delivery_mode=1	Broker级消息持久化
task_acks_late = True	✅ 配合reject_on_worker_lost=True	❌ 单独启用	ACK时机与崩溃窗口错配
worker_prefetch_multiplier = 1	✅ 防止单Worker积压过多unack消息	❌ >1（尤其在长耗时任务场景）	OOM时批量丢失ACK机会

四、架构层：At-Least-Once语义与应用幂等性的协同设计

必须打破“Broker保证不丢消息 → 应用无需幂等”的认知误区。真实链路存在三重不确定性：

1. 网络层：TCP连接中断导致ACK包未抵达Broker（即使Worker已执行成功）；
2. 运行时层：Python GIL阻塞+信号处理延迟，使atexit钩子无法在kill -9下触发；
3. 存储层：DB事务提交成功但ACK未发，或ACK发出但网络抖动丢失。

因此，幂等性不是备选方案，而是强制契约。推荐采用「状态机+唯一业务键」模式：

def process_payment(order_id: str, amount: Decimal):
    # 幂等关键：先查再判，而非先判再查
    status = PaymentStatus.objects.filter(
        order_id=order_id,
        status__in=['pending', 'processing']
    ).select_for_update().first()
    
    if not status:
        # 已完成或不存在 → 直接返回
        return {'result': 'idempotent_skip'}
    
    # 执行核心逻辑（扣库存、调支付网关等）
    status.status = 'success'
    status.save()
    return {'result': 'processed'}

五、运维层：可验证的健康检查清单

• 【Broker检查】RabbitMQ执行sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged durable，确认关键队列durable=true；
• 【Worker检查】启动时日志必须包含acknowledging messages at task completion（acks_late=True）或acknowledging messages at task receipt；
• 【压测验证】模拟kill -9后，观察celery inspect active_queues是否残留unack任务，且5分钟后是否被重新分发；
• 【审计追踪】在任务入口统一注入task_id与retry_count，通过ELK聚合分析重复触发率（目标<0.001%）。

```