AI生图异步任务设计与轮询退避并发控制实战

有一次给业务线集成 AI 生图能力我图省事直接把上游接口包成了同步调用传提示词进去等图片 URL 出来返回给前端。上线第三天超时率直接奔 40% 去了查了半天才发现根因根本不在我写的代码里而是上游生图平均耗时 20 多秒我们自己的网关超时阈值只有 10 秒。再往后我把这种拿异步当同步的做法彻底放弃了花了两周沉淀出一套通用方案覆盖异步任务管理、轮询退避、并发控制。这篇文章就是把这套东西完整拆开讲一遍里面有设计思路、有代码骨架、也有我踩过的边界问题。适合正在做 AI 应用、AI 工作流、或者想把生图/生视频能力接入业务的后端工程师参考。1. 为什么 AI 生图/视频天生不是同步接口1.1 一次生成动辄几十秒网关等不了你先算一笔账。生成一张 1024x1024 的图片即便是比较快的模型GPU 推理也要 3~10 秒如果走云端的排队调度加上模型冷加载平均响应普遍在 10~30 秒。视频更夸张一个 5 秒的短视频片段从推理、帧间一致性优化到编码出片经常要 30 秒到 5 分钟遇到复杂风格或超高清分辨率十分钟以上也不奇怪。但绝大多数 HTTP 服务的基础设施不是为这种延迟设计的。Nginx 默认 proxy_read_timeout 是 60 秒云厂商 SLB 的闲置超时通常 30~60 秒客户端 SDK 的默认超时普遍也是 10~30 秒。你用一个 10 秒超时的调用去等一个 30 秒的任务等待的结局大概率不是成功而是一个 timeout。这里容易犯的错误是把超时调大就能同步等待了。调大确实能撑住一部分场景但带来了三个新问题网关和连接池的连接被长耗时请求长期占用并发一上来其他普通接口也被拖死客户端断开后服务端往往还在傻等上游白白消耗算力超时语义变得极模糊到底是上游失败、网络抖动、还是任务还在排队你分辨不出来。1.2 服务端的 GPU 调度决定了它只能异步从上游 AI 服务提供方的视角看生图/生视频接口也不该做成同步。这类服务背后通常是 GPU 池 队列。每次请求进来系统把任务丢进队列由调度器安排到空闲 GPU 上跑如果 1000 个任务同时进来GPU 只有 10 张那么后 990 个任务就是纯排队。如果做成同步连接会被排队任务占着调度器反而没法做优先级、抢占和取消上游的限流、计费粒度也会变得很难设计。所以你会发现主流的生图/视频 API 都采用同一个模式先提交任务拿到一个 task_id然后靠轮询或者回调获取最终结果。这不是产品缺陷是这类服务的天然形态。你真正要整合的不是一个返回图片 URL 的同步函数而是一条提交任务 - 查看进度 - 获取结果的异步链路。1.3 强行同步化的封装层是事故高发区我见过很多团队包括当时的我自己会写一个这样的同步封装def generate_image_sync(prompt: str, timeout: int 60) - str: task_id ai_client.submit(prompt) # 上游是异步的 while True: status ai_client.query(task_id) if status in (success, failed): return ai_client.fetch_result(task_id) time.sleep(2)看起来简洁问题是这个函数一旦被当成普通 API 暴露出去每一个在途生成都会占用一个 worker 线程/协程在那里 sleep 轮询。假设你有 50 个请求同时生成就有 50 个线程被挂在这种等待循环里。线程池一满后面哪怕只是查个任务状态的轻量请求也进不来。更麻烦的是服务进程一重启这些内存里的等待循环全部消失任务状态没人接管用户拿到的就是任务神秘失踪。不是说同步封装绝对不能用而是它只能被用在非常受控的场景调用方明确知道底层是异步的、有独立连接池、有可靠超时和状态恢复机制。大多数情况下没必要自己造这个轮子直接做异步任务模型更干净。2. 异步任务的最小可靠模型状态机与存储选型2.1 先定义清楚任务状态机再写代码异步任务的核心不是队列而是状态。状态设计得不够细后面排查问题会非常痛苦。我常用的五状态模型状态含义谁写入下一步pending已接收入队等待 worker 处理API 层worker 领取后转 runningrunningworker 正在调用上游 AI 服务worker成功转 success失败转 failedsuccess上游返回结果结果可读取worker结束按 TTL 清理failed上游失败或重试耗尽worker / 调度器结束记录 errortimeout超过 deadline 未完成巡检任务 / worker结束可触发补偿这个状态机最大的价值是可观测任何一个任务卡在哪个环节看状态就能定位。后面所有逻辑——轮询退避、并发控制、异常处理——本质上都是围绕这五状态转移做的。2.2 存储选型Redis 够用DB Redis 更稳管理任务状态常见方案有三种我简单对比一下方案优势劣势适用阶段纯 Redis读写快、支持 TTL 自动过期、API 简单持久化弱宕机可能丢任务复杂查询弱原型、内部小工具DB 任务表 Redis 队列任务台账可靠、可 SQL 统计查询Redis 只做待处理队列需要写两套读写逻辑正式产品化专用任务队列框架Celery/BullMQ/Temporal功能齐全有重试、调度、分布式支持引入额外运维依赖调试黑盒大型团队/复杂编排我现在的默认组合是DB 记录任务详情 Redis 存状态和队列。DB 表里的字段包括 task_id、请求参数、状态、错误信息、创建时间、最后更新时间、deadline、幂等键Redis 里只放热数据例如状态字符串、待处理队列、信号量计数。这样既保证了任务不会因为 Redis 重启而永久丢失又利用了 Redis 的低延迟处理高频轮询。2.3 API 契约submit 同步返回poll 慢慢来无论上游怎么变化我建议你的系统对外暴露的接口保持稳定POST /v1/tasks # 提交任务返回 task_id GET /v1/tasks/{task_id} # 查询状态、结果、错误信息 DELETE /v1/tasks/{task_id} # 取消任务可选但强烈建议有前端拿到 task_id 后先展示生成中再利用退避策略轮询详情接口。接口层不要做任何阻塞等待。这套契约跟前端配合也非常自然用户点一下生成按钮立刻进入 loading轮询有结果后再切换成结果图/视频。3. 轮询退避的精髓既要及时又别把自己打挂3.1 固定频率轮询是所有人第一天就会踩的坑最简单粗暴的轮询就是每 2 秒查一次查不到就继续。它的问题有两个一是空转率极高二是请求共振。先看量假设 200 个用户同时在生成视频每个任务平均 3 分钟每 2 秒轮询一次那么单个任务要查询 90 次200 个任务就是 18000 次请求。如果这些请求还要穿透到 DB 或者上游 AI 接口压力就更大。再看共振所有客户端都用每 2 秒这个固定节奏第一秒查询请求全部同时到达第二秒系统闲着第三秒又瞬间打满。这种锯齿形的流量尖峰对基础设施很不友好。3.2 指数退避 全抖动最实用的算法组合业界最常用的是指数退避但在指数基础上一定要加随机抖动。我这里给出一个可复用的 Python 实现import asyncio import random async def wait_with_backoff(attempt: int, base: float 1.0, cap: float 15.0) - None: Full Jitter 指数退避。 attempt 从 0 开始base 为初始等待上限cap 为最大等待上限。 exp min(cap, base * (2 ** attempt)) delay random.uniform(0, exp) await asyncio.sleep(delay)为什么用全抖动而不是固定算出来的指数值因为如果 100 个客户端在 attempt3 时都算出同一个等待值并同时醒来退避就失去意义了——大家依旧会在同一时刻打上来。把等待值随机化可以让请求在时间轴上散开服务端的峰值压力平滑很多。实际项目中我也会用 Fibonacci 退避作为替代它比指数退避增长更温和适合那些慢但不能太慢的任务。对大多数生图/视频场景指数退避 抖动已经足够。3.3 轮询退避要分层设计前端和后端不是一回事有些团队只设计了后端的轮询却忘了控制前端结果前端页面用 1 秒固定频率轰炸详情接口。我建议至少分两层做统一规划。轮询方建议初始间隔建议最大间隔关键注意点前端浏览器轮询1~2 秒10~15 秒页面隐藏时暂停用 visibilitychange 兜底后端 worker 轮询上游2~3 秒15~30 秒必须带 attempt 上限和 deadline回调/webhook事件驱动-回调是主动推送轮询只做兜底另外要记住退避不是让你无限重试。一个任务连续失败 3 次大概率不是抖动问题而是上游在报错或者服务有问题。此时要停止轮询并把任务标记为 failed写入 error 信息。否则你会发现退避反而变成了另一种风暴源。4. 并发控制把 AI 请求压到安全水位4.1 并发失控的真实代价不控制并发最先出事的是上游 AI 服务。大多数生图/视频平台对单账号有两类限制每分钟请求数RPM和在途并发数。超过之后会返回 429 或 503如果频繁触发甚至可能被上游暂时封禁。下游自己的服务也会遭殃数据库查询变慢、连接池耗尽、最终把任务队列本身拖垮。举个例子。运营同事一次性提交了 30 个视频生成任务每个任务在上游要处理 5 分钟。如果你的 worker 不设并发上限30 个任务会同时发起调用上游瞬间被打满后面所有用户的任务全部排队体感就是系统卡死了。这不是上游不够强而是你没有在上游和业务之间做好水坝。4.2 信号量限流最直接的并发闸门在 asyncio 生态里最简单的并发控制就是信号量MAX_CONCURRENT 5 # 根据上游配额和自身处理能力调整 semaphore asyncio.Semaphore(MAX_CONCURRENT) async def dispatch_to_upstream(task_id: str) - None: async with semaphore: # 只有拿到信号量才会真正发起上游调用 await call_upstream(task_id)信号量的语义非常简单超过 MAX_CONCURRENT 的 worker 就等在 semaphore 外面等里面某个任务完成信号量释放才放一个进去。它解决的是同一时刻有多少任务在飞的问题这也是 AI 任务最关键的指标。如果项目是多机部署单机信号量就不够了需要分布式版本。可以用 Redis 的原子计数做一个简单的在途数量控制任务开始时 INCR结束或失败时 DECR如果当前计数已经达到上限就直接返回繁忙或让 worker 等待。要注意给计数加过期时间防止 worker 崩溃后计数泄漏导致闸门永远关闭。4.3 双层配额全局上限 用户级隔离只做全局并发控制还是会有单个用户把所有额度吃满的情况。建议再加一层用户维度限制。限额类型示例值作用全局并发在途任务数5~20保护上游配额和下游自身处理能力单用户并发在途任务数1~3防止某个用户刷爆队列单用户提交频率每 10 秒最多 1 次控制恶意/重复请求实现时用户级限制可以在入队前检查该用户当前处于 pending/running 状态的任务数是否已达到上限。这个方法需要一条查询在 Redis 里可以给每个用户维护一个在飞任务集合用 SADD/SREM 维护。全局并发则用信号量那一套边界清晰。5. 一套可落地的生产骨架FastAPI Redis 示例前面讲的是思路这一节给一个能直接跑的基础骨架。我会演示一个小型服务包含提交、查询、worker 消费、并发控制和退避轮询。技术栈选 FastAPI asyncio Redis因为这是比较轻量又能覆盖大多数场景的组合。5.1 模块划分整体分四块API 层POST /v1/tasks 提交GET /v1/tasks/{id} 查询队列Redis List待处理任务暂时存在这里worker从队列 BLpop 取出任务控制并发调用上游结果存储Redis 里以状态键和结果键保存任务最终数据。这里会把同步封装问题直接绕开——API 层只入队绝不等待真正的漫长等待发生在 worker 内部维护的轮询循环里。5.2 基础代码提交、查询、worker 消费import asyncio import json import uuid from fastapi import FastAPI, HTTPException import redis.asyncio as aioredis app FastAPI() redis_client aioredis.from_url(redis://localhost:6379/0) QUEUE_KEY ai_task_queue MAX_CONCURRENT 5 semaphore asyncio.Semaphore(MAX_CONCURRENT) def status_key(task_id: str) - str: return ftask:{task_id}:status def result_key(task_id: str) - str: return ftask:{task_id}:result app.post(/v1/tasks) async def create_task(payload: dict) - dict: task_id uuid.uuid4().hex # 上游请求参数先原样存起来worker 取任务时再解析 await redis_client.set(ftask:{task_id}:payload, json.dumps(payload)) await redis_client.set(status_key(task_id), pending) # 入队能用 lpush/rpop 还是 brpop取决于 worker 端 await redis_client.lpush(QUEUE_KEY, task_id) return {task_id: task_id} app.get(/v1/tasks/{task_id}) async def get_task(task_id: str) - dict: status await redis_client.get(status_key(task_id)) if not status: raise HTTPException(status_code404, detailtask not found) result await redis_client.get(result_key(task_id)) result_data json.loads(result) if result else None return {status: status.decode(), result: result_data} async def worker_loop() - None: while True: item await redis_client.brpop(QUEUE_KEY, timeout5) if not item: continue task_id item[1].decode() payload_raw await redis_client.get(ftask:{task_id}:payload) payload json.loads(payload_raw) if payload_raw else {} async with semaphore: await redis_client.set(status_key(task_id), running) try: result await call_upstream(payload) await redis_client.set(status_key(task_id), success) await redis_client.set(result_key(task_id), json.dumps(result)) except Exception as exc: await redis_client.set(status_key(task_id), failed) await redis_client.set(ftask:{task_id}:error, str(exc))这段代码里需要注意两个细节。第一brpop 的 timeout 设为 5 秒是为了让 worker 可以定期醒来做清理逻辑如果设置成负数会一直阻塞反而不好退出。第二信号量包住了整段上游调用确保全局并发不会超过 5 个在飞任务。5.3 上游也是异步的worker 里的退避轮询上面代码里call_upstream(payload)是 mock。真实场景中上游 AI 服务本身往往也是异步的所以 worker 内部要做一次提交 轮询 退避async def call_upstream(payload: dict) - dict: upstream_task_id await ai_service.submit(payload) # mock: 上游建立任务 attempt 0 max_attempts 12 deadline asyncio.get_event_loop().time() 15 * 60 # 15 分钟硬截止 while True: if asyncio.get_event_loop().time() deadline: raise TimeoutError(upstream task deadline exceeded) state await ai_service.query(upstream_task_id) if state success: return await ai_service.fetch_result(upstream_task_id) if state in (failed, canceled): raise RuntimeError(fupstream task failed: {state}) attempt 1 if attempt max_attempts: raise RuntimeError(retry attempts exhausted) await wait_with_backoff(attempt, base2.0, cap30.0)这里有一个学习成本最低但也最容易忽略的原则退避循环必须同时设置 attempt 上限和 deadline。否则上游任务万一卡住worker 会无限等下去任务永远停在 running信号量也一直不释放。加 deadline 的好处是无论什么异常最多 15 分钟就能把资源释放出来至少能给用户一个明确的 timeout 结果。5.4 兜底webhook 回调优先轮询退避兜底很多成熟的上游 AI 服务支持配置 webhook 回调地址任务完成时会主动 POST 结果过来。我强烈建议有条件就用回调把轮询退避留作兜底。原因很简单回调是事件驱动几乎零空转轮询无论怎么优化本质上都是猜测结果什么时候好。引入回调后要注意幂等。回调 HTTP 请求可能因为网络原因被发送多次所以回调处理函数要先把 task_id 拿出来检查任务是否已经是终态如果是就直接返回不做重复写结果。轮询和回调同时存在时也要保证谁先到谁生效这同样靠状态机的终态检查来兜底。6. 上线前最容易漏掉的边界超时、幂等与清理6.1 超时不是轮询自然带出来的要显式定义很多系统上线后才发现一个 bug任务状态一直是 running但没有任何代码在管它。虽然每个任务都在退避循环里可一旦 worker 进程异常退出队列里或 running 状态的任务就变成了孤儿。建议加一个巡检逻辑定期扫描所有处于 running 状态且最后更新时间超过阈值比如 10 分钟的任务把它们标记为 timeout释放信号量并通知用户。实现上可以用另一个异步任务每 30 秒执行一次也可以直接在 brpop timeout 醒来的 5 秒里顺手扫一遍。这个巡检是异步任务系统的安全网。6.2 幂等键防止重复提交重复扣费AI 生图/视频一般按次计费重复提交会带来真金白银的损耗。前端按钮防抖只能挡住一部分服务端必须做幂等。简单做法是让客户端传一个Idempotency-Key请求头或参数服务端以它为唯一键如果已经存在就返回已有 task_id不再创建新任务。在 Redis 里可以这样实现# NX 保证键不存在时才写入EX 设置过期时间防堆积 ok await redis_client.set(fidem:{idem_key}, task_id, nxTrue, ex86400) if not ok: old_task_id await redis_client.get(fidem:{idem_key}) return {task_id: old_task_id.decode(), reused: True}这个逻辑成本极低强烈建议放进任何对外 AI 能力接口。6.3 结果和描述信息要定期清理生图/视频任务的产物多数是对象存储 URL 或二进制文件任务完成后马上让用户去取才是有效的。Redis 里最终结果键和 error 键建议设置 TTL比如 7 天DB 任务表则定期归档超过 30 天的记录。否则结果越堆越多存储成本会变成一笔让人肉疼的账单。6.4 可以长期跟踪的监控指标最后分享几个我每次上线异步任务系统都会盯的指标任务状态分布pending/running/success/failed/timeout 的数量任何一个状态突然猛涨都是信号平均任务完成耗时和 P95 耗时用于判断用户等待体感每次轮询的空转命中率查询到非终态的比例空转过高就说明轮询太频繁或退避间隔太短上游 429/5xx 错误率这个指标往往是并发控制失效的第一信号。我个人在跑这类系统时最大的体会是异步任务、轮询退避、并发控制这三件事看起来是三个独立技术点实际上是一条链路上的三兄弟——并发控制不当会导致上游限流限流后会引发重试风暴重试风暴又会让任务排队时间变长排队变长又迫使你不断加长轮询等待。把它们放在一起设计整体水位才能稳。整套方案我已经在多个业务场景里复用从内部工具到对外 API 都适用核心思路不变变的只是参数和存储细节。希望能帮你少踩几个我踩过的坑。