Celery + Redis 队列化 K8s 推理架构实战

dev.to · AI#Celery#Redis#Kubernetes

Editor brief · 编辑速览

AI 推理任务耗时长、默认配置全部不适配；文章给出 task_acks_late、worker_prefetch_multiplier=1、broker_transport_options 等关键配置，以及 Redis StatefulSet + Celery Deployment 的具体部署结构。

文章思维导图

Knowledge map

拖拽缩放

Full translation

完整中文译文

Celery 的默认配置是为毫秒级任务设计的。这些默认项在面对一个需要等待模型运行一分钟的任务时，每一项都是错的——而且它产生的失败并非报错，而是一个看似空闲、却握着自己根本没开始执行的四十个任务的 worker。

四个 Kubernetes 对象。Redis 作为 StatefulSet，带有 PersistentVolumeClaim 和一个 headless Service。Celery worker 的 Deployment。存储 broker URL 和提供商 API key 的 Secret。以及，如果你想通过 Celery 而非自己的数据表获取结果，一个 result backend——可以是同一个 Redis，尽管把两者指向同一个实例意味着 broker 宕机时结果也同时丢失。

生产者（无论它是什么）不需要出现在这个架构图里。它只需导入同一个 task 模块，调用 .delay() 或 .apply_async()，然后拿到一个 id。这篇文章讲的全是消费侧。

# tasks.py
from celery import Celery
import os

app = Celery("inference", broker=os.environ["CELERY_BROKER_URL"])

app.conf.update(
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    worker_prefetch_multiplier=1,
    task_soft_time_limit=180,
    task_time_limit=210,
    broker_transport_options={"visibility_timeout": 900},
)

@app.task(bind=True, max_retries=3)
def infer(self, job_id: str, prompt: str) -> None:
    if already_done(job_id):
        return
    write_result(job_id, call_model(prompt))

三个在这里错误的 Celery 默认值

Prefetch（预取）。 Celery 文档指出 worker_prefetch_multiplier 默认为 4，即每个 worker 进程一次预留四条消息。十个并发进程就是四十个任务被一个 pod 占有，其中大多数只是在内存里干坐着。对毫秒级任务来说这是吞吐量的优势。对一个九十秒的任务来说，这会导致队列在所有观察者眼中呈现已被清空的状态，从而破坏自动扩缩容——参见"根据队列长度扩缩容 Celery workers"。把它设为 1。

Early acknowledgement（提前确认）。 Celery 文档指出 task_acks_late 默认关闭，这意味着任务在交给 worker 时就已确认，而非在执行完成后。如果 pod 在调用过程中被驱逐，任务就直接消失了。开启它会把失败模式从"工作丢失"转变为"工作重复"——对于模型调用，这是只有在任务具有幂等性之后才值得做的权衡。所以先让任务幂等，然后开启它，并配合 task_reject_on_worker_lost 让被杀掉的 worker 重新入队，而不是静默丢弃。

No time limit（无时间限制）。 Celery 文档指出 task_time_limit 和 task_soft_time_limit 都默认无限制。因此一个对提供商挂起的 HTTPS 连接会无限占用一个 worker 槽位。软限制在你的任务内部抛出 SoftTimeLimitExceeded，以便你记录失败；硬限制则直接杀掉进程。把软限制设得低于你所用 provider 客户端自身的读取超时时间，这样你得到的是有意义的异常而非突然的中断。

Redis 作为 broker 给了你什么，没给你什么

Redis 并不是一个消息 broker；Kombu 在其之上实现了一个。必须知道的一个后果是 transport 的 visibility_timeout，Kombu 文档指出默认为 3600 秒。一条在该时间窗口内未被确认的任务会被重新投递给另一个 worker，这与 SQS 的行为完全一致。

一小时足够长以至于大多数人从未注意到它，又足够短以至于一个真正卡住的任务最终会重复执行。真正重要的数字不是模型调用本身耗时，而是从投递到确认的整段时间——在 late ack 模式下，这包含了 Celery 在任务内部施加的任何重试退避时间。如果你的任务以指数退避重试三次，要预算的是总时间，而不是一次尝试的时间。

3600 秒的默认值来自 Kombu Redis transport 参考文档，读取时间是 2026 年 8 月。这是库级默认而非服务端配额，所以可能随依赖升级而变化，而非随供应商公告。参见 Kombu: Redis transport。

Redis 也不会给你持久的确认语义和队列镜像。Kombu 所谓的确认不过是从一个 in-flight 结构中删除，这就是为什么模拟的 visibility timeout 之所以存在，以及为什么 Redis 故障切换可能以真正具有发布者确认的 broker 不会的方式丢失或重复工作。如果持久性比运维简单更重要，替代方案是一个真正的 broker；等效的搭建方案见"RabbitMQ for queued model inference on Kubernetes"。

对这个工作负载的实际影响是，你不能把队列当作"什么必须发生"的记录。在接收时把自己的 pending 行写入你自己的数据库，让队列只是一个工作信号而非事实来源。这样丢失的消息就是一条永远不会离开 PENDING 状态的行，一个 sweeper 可以重新入队它，而不是用户按了一个按钮然后什么都没发生的静默蒸发。

Worker Deployment，以及如何正确关闭它

默认的 Kubernetes grace period 是 30 秒。收到 SIGTERM 的 Celery worker 会执行热关闭：停止接收新任务，等待运行中的任务完成。一个九十秒的模型调用加上三十秒的 grace period，意味着每次部署时 SIGKILL 都会落在调用中途。把 terminationGracePeriodSeconds 设得高于你的硬任务时间限制。

apiVersion: apps/v1
kind: Deployment
metadata:
  name: celery-inference-worker
spec:
  replicas: 2
  selector:
    matchLabels: { app: celery-inference-worker }
  template:
    metadata:
      labels: { app: celery-inference-worker }
    spec:
      terminationGracePeriodSeconds: 240
      containers:
        - name: worker
          image: registry.example.com/inference-worker:1.4.0
          args: ["celery", "-A", "tasks", "worker",
                 "--loglevel=INFO", "--concurrency=4", "--prefetch-multiplier=1"]
          envFrom:
            - secretRef: { name: inference-secrets }
          resources:
            requests: { cpu: "200m", memory: "512Mi" }
            limits:   { memory: "1Gi" }

注意没有 CPU limit 而有一个低的 CPU request。一个在等待 socket 的 worker 几乎不消耗 CPU，而这个工作负载上的 CPU limit 只会带来不必要的节流。内存才是值得设限的资源，因为大响应在内存中累积是这个 pod 真实会死亡的方式。

当 pod 被 OOM 杀死时

以上全部假设的是 SIGTERM。内存不足导致的杀死是 SIGKILL，以上全不适用：没有热关闭，没有 grace period，没有确认或重新入队的机会。worker 在一条指令和下一条指令之间消失，握着其 prefetch 所占有的那么多未确认投递。

接下来发生什么完全取决于前文的两个设置。如果 task_acks_late 处于关闭状态——即 Celery 默认值——那些任务在移交时就已经确认了，所以它们直接消失了，唯一的痕迹是一条永远不会变成 done 的 pending 行。如果开启了 late acks，它们就是未确认的，Kombu 的 Redis transport 会在其 visibility_timeout 之后将它们返回。除非你改过，默认是 3600 秒，所以对 stock 配置下 OOM 被杀死的 worker 的恢复时间的诚实描述是"最多一小时后"。在一个面向用户的 pipeline 上，这和丢失请求无法区分。

这就是把 visibility_timeout 设为一个衍生自你工作实际的值而非留空的理由：长到足以覆盖最坏情况任务及其重试，足够短以至于被杀的 pod 只是一段人能够容忍的延迟。在你的硬时间限制之后几分钟的量级，而不是一小时。

降低被杀频率是另一半。一个模型响应先作为字符串保存，然后解析，然后再序列化，是同一份 payload 的三份副本，而并发为 8 的 worker 正在同时做八次。两项缓解措施很廉价：让并发数与内存限制成正比而非与 CPU request 成正比；设置 --max-tasks-per-child 以便定期回收 worker 进程，这能约束任何泄漏物的累积。观察那些 last state 显示 OOMKilled 且退出码为 137 的 pod——一个 Deployment 可能在外观上完全健康，却悄悄地每隔几分钟就杀死并重启一个 worker，外部唯一的症状是一小部分任务需要一小时。

总结

把 Redis 以 StatefulSet + PVC + headless Service 的方式部署。把得到的 CELERY_BROKER_URL 和你的 provider key 放进一个 Secret。

用上述四个配置覆盖编写 tasks.py，并让 task 体以检查任务是否已完成作为开头。

用 worker 命令作为 args 构建镜像，而非 baked 进 entrypoint 脚本，这样修改并发数不需要重新构建。

用 terminationGracePeriodSeconds 高于 task_time_limit 的方式 Apply Deployment。

入队一个任务，然后在调用中途 kubectl delete pod 那个 worker，确认任务被重新投递而非丢失——这是验证 late acks 确实开启了的测试。

在 worker 运行期间观察队列深度。当 prefetch multiplier 为 1 时它应该跟踪真实的积压，这才是对其做基于队列深度的自动扩缩容的前提。

延伸阅读

Scaling Celery Workers by Queue Length on Kubernetes

RabbitMQ for Queued Model Inference on Kubernetes

Kubernetes for Model Serving