Celery 异步任务全生命周期调用链

把一个普通函数用 @app.task 标记后，Celery 会把它“登记成任务”：给它分配一个唯一的任务名（后续靠任务名找到并执行它）。同时也会记录“发消息的默认规则”：当你调用 delay/apply_async 时，要把任务名、参数、task_id、倒计时/ETA、以及需要的元信息打包成一条可传输的字符串（常见是 JSON）。另外还会记录“发到哪里”：默认走哪个 queue、使用什么 routing_key（或在代码里显式指定），这样 Broker 才知道把消息放进哪个队列，Worker 才会从对应队列里取到并执行。最后再记录任务在执行生命周期里没有成功完成即任务失败时怎么处理（重试次数/间隔、超时等）。

调用 delay/apply_async：把“执行意图”封装成异步指令——任务名、Args/Kwargs、task_id、routing_key、ETA/countdown 等元数据；再通过 Kombu 将消息非阻塞投递到 Broker。调用会立刻返回 AsyncResult（追踪句柄，持有 task_id），之后可用它查询状态（PENDING/STARTED/SUCCESS/FAILURE）或取结果；同时 apply_async 还能在调用时覆盖默认路由（指定 queue/priority 等），把同一任务按场景分发到不同 Worker 集群。

生产者把消息推给 Broker：RabbitMQ 遵循 AMQP，把消息先投递到 Exchange（默认 celery），再由 Exchange 按 routing_key + binding 规则决定进入哪个 Queue；Redis 没有 Exchange 层，通常把“队列名”直接映射为一个 List Key，消息会直接写入该 Key 对应的列表。

核心动作：队列（Queue）主动绑定（Binding）到交换机（Exchange），并在绑定时写下自己的过滤规则（Routing Key）。

• 工作流：
  • Producer 发消息到 Exchange，并给消息贴一个 routing_key
  • Exchange 遍历自己所有 Binding（哪些 Queue 绑定了我、各自写了什么规则）
  • 对比“消息的 key”和“绑定的 key”，匹配成功就把消息投递到对应 Queue
• 不同 Exchange 的绑定匹配方式
  • Direct：严格匹配。例：Queue A 绑定 key=error，只有消息 key=error 才进入
  • Topic：通配匹配。例：绑定 stock.#，可匹配 stock.usd/stock.cn
  • Fanout：忽略 key。只要绑定了，就给所有绑定队列各发一份拷贝（广播）

Redis 用 List 模拟队列：Producer 执行 LPUSH 把消息塞到列表左端；Worker 执行 BRPOP 阻塞式从右端弹出消息（队列为空时连接挂起，不消耗 CPU；有新消息时立刻被“弹醒”）。因为没有 Exchange，路由通常在客户端侧就决定好写入哪个 List Key。

• 存储架构（内存 vs 分层）
  • Redis：消息就是 List 里的元素，主要驻留在 RAM，读写极快，但堆积大了会吃满内存，极端会 内存溢出OOM。
  • RabbitMQ：同时管理 内存 + 磁盘。消息可写入磁盘 segment 文件并维护索引；堆积很大时会触发 paging，把部分消息从内存换到磁盘，用“变慢”换“服务不挂”。
• 持久化策略
  • Redis
    • RDB：定时快照（像“拍照”）。间隔内断电可能丢最近一段时间的新任务。
    • AOF：追加写命令（像“记账”）。高频 LPUSH 会让日志迅速膨胀，磁盘 IO 压力变大，极端会拖慢 Redis。
  • RabbitMQ
    • delivery_mode=2（persistent）：消息到达 Exchange/Queue 后，RabbitMQ 不只放内存(默认（非持久化）)，而是会把该消息写入磁盘（segment 文件）；必要时调用 fsync 强制把 OS 缓存刷到物理磁盘。只有在“已经安全落盘”之后，RabbitMQ 才会向生产者确认收到——因此断电重启后还能从磁盘把未完成消息恢复回来。代价是吞吐下降（磁盘 IOPS 限制）。在 Celery 中通常可通过 app.conf.task_default_delivery_mode = 'persistent' 开启。
    • watermark + paging：内存逼近水位线时强制刷盘，只在内存保留必要索引，避免内存中积压的消息太多导致内存溢出 OOM。
• 总结建议
  • Redis：像高速缓存区——适合高频轻量、偶发丢几条可接受的任务（验证码/通知等）。
  • RabbitMQ：像专业仓库——适合波动大、关键任务“绝对不能丢”的场景（订单/计费等，建议开启持久化）。

分发：Broker 把队列里的消息“交给”哪些 Worker、以什么节奏交付。背压：当 Worker 忙/慢时，限制消息从 Broker（中间件）进入 Worker（执行进程）内存的速度，避免内存被撑爆或延迟失控。

• RabbitMQ：Push（主动推）+ 配额控制（QoS/Prefetch）
  • 机制：Worker 建连后声明 prefetch_count（预取数量），RabbitMQ 会按配额把消息推过去。
  • 在途消息：推送给 Worker 但尚未 ACK 的消息属于 Unacked（在途）。
  • 例子：prefetch=4 时，RabbitMQ 最多同时让该 Worker 持有 4 条未确认消息；只要 Worker ACK 了一条，才再补推下一条。
  • 背压效果：受控涌入。重任务（AI 推理/生图）常把 prefetch 设小（甚至 1）；轻任务（发邮件）可设大提高吞吐。
• Redis：Pull（按需拉）+ 自适应（BRPOP）
  • 机制：Worker 用 BRPOP 阻塞式拉取：有任务立即弹出；没任务就挂起等待，不忙轮询。
  • 背压效果：天然“按需取货”。Worker 处理慢 → 发起下一次拉取更晚；Worker 挂了 → 不再拉取，Broker 侧无需做复杂配额控制。

目标：防止 Worker 中途挂掉导致“任务丢失”。两种 Broker 的实现思路不同：

• RabbitMQ：内置 ACK
  • Unacked：消息投递给 Worker 后不会立刻删除，而是标记为 Unacked（待确认），对其他 Worker 不可见。
  • ACK 才删除：Worker 完成任务后发送 ACK，RabbitMQ 才真正从队列/磁盘中移除该消息。
  • 断连即重投：若 Worker 崩溃导致 TCP 连接断开且未 ACK，RabbitMQ 会把消息状态从 Unacked 立刻改回 Ready，重新排队并投递给其他健康 Worker（Redelivery）。
• Redis：备份队列 + 超时捞回
  • 原子位移：因为直接弹出会“消失”，Celery 往往用 RPOPLPUSH（或新版 BLMOVE）把消息从主队列原子移动到“处理中/备份队列”（常见 key 形如 unacked:*）。
  • 完成后删除备份：Worker 真正执行完成后，再把备份队列里的那条记录删掉，相当于“确认”。
  • visibility_timeout：Redis 不能像 RabbitMQ 一样实时感知断连，Celery 依赖超时（例如 1h）判断“处理太久=可能挂了”，再把备份队列里的消息捞回主队列实现重投（恢复速度取决于超时配置）。
• 核心差异：RabbitMQ 是“实时感应（断连触发）+ 立即重投”；Redis 是“延迟感应（超时触发）+ 备份队列捞回”。

这是任务的“高光时刻”：消息进入 Worker 后，会被还原为可执行的调用，并按以下阶段推进：

• 1) 反序列化
  • 校验 content-type（如 application/json），把消息体解包为 args/kwargs
  • 根据任务名/任务ID定位本地函数引用；若代码版本不一致/任务未注册，可能报 KeyError
• 2) 执行业务逻辑
  • 按并发模型运行：子进程 / 线程 / 协程
  • 超时控制：soft time limit 先抛 SoftTimeLimitExceeded 给清理机会；hard time limit 可能强制终止
  • 异常与重试：捕获异常准备 traceback；代码里调用 self.retry() 会按策略（如指数退避）重新投递带 ETA（预计到达时间） 的任务
• 3) 结果落盘
  • 可选：如果你需要查询结果，Worker 会把返回值/异常信息写到 Result Backend，并更新状态为 SUCCESS/FAILURE
  • 如果业务不需要返回值，建议开启 task_ignore_result=True（或任务级 ignore_result=True），避免结果在 Backend 大量堆积
• 4) ACK
  • Worker 向 Broker 确认“这条消息我已处理完”，Broker 才会彻底删除/释放这条消息
  • acks_late=True：执行完成后再 ACK（更可靠，Worker 中途挂了可触发重投）；相对地“收到即 ACK”吞吐更高但更易丢

当你调用 AsyncResult.get() 时，本质是在“查任务结果 + 等它就绪”，以及在结果写入后“按策略回收”。

• 客户端查询：get() 的两种模式
  • 轮询：按 interval（默认约 0.5s）循环向 Backend 发起读取（例如 Redis 的 GET celery-task-meta-<task_id> 或查询 DB 记录）。
  • 控制参数：timeout 防止无限等待；propagate 决定任务失败时是否把异常在客户端重新抛出。
  • 事件唤醒：以 Redis 为 Backend 时，可通过发布/订阅减少“频繁敲门”。客户端订阅与 task_id 相关的通知通道；Worker 写完结果后发布消息，客户端被瞬间唤醒，显著降低无效请求与 CPU 压力。
• 结果清理：TTL 与自动回收
  • Redis：写结果时顺带设置 EXPIRE，由 result_expires 控制保留时间（默认通常是 1 天）。时间到后 Key 自动物理删除。
  • 数据库：没有行级 TTL。结果会一直存在，需要周期性清理：常见由 celery.backend_cleanup（通常配合 celery beat）定时执行删除过期行。
• 核心配置与避坑
  • task_ignore_result：最关键优化。默认建议全局开启；只在确实需要返回值的任务上用 @app.task(ignore_result=False) 显式打开。
  • result_expires：结果保留时长（秒）。并发高时建议设短一些（例如 3600=1h）以降低堆积风险。
  • 经典坑：不启用 task_ignore_result=True，又从不调用 .get()/.forget()，结果会在 Backend 持续堆积直到过期，最终导致缓存/数据库被撑爆。