把商品图生成接进业务系统，第一个绕不开的问题就是：任务提交之后，怎么知道它什么时候完成。

最直觉的做法是同步等。请求发出去，requests.post() 一直挂着，等上游返回图片地址再往下走。这个方案在测试环境通常能跑通，上线之后一定会出事：

• 出图耗时波动大，同一批任务有的 8 秒有的 90 秒，同步等会把连接池占满；
• 中间加一层网关，网关的超时时间往往比出图时间短，请求被切断，任务却已经在上游跑起来了；
• 用户点一次"生成"要等半分钟，页面体验很差，还没有进度。

正确的做法是异步：提交拿任务号，结果靠回调 + 兜底把状态补回来。

一、五个环节

一条完整的链路是这样走的：

示意图：提交、生成中、回调、兜底轮询、对账五个环节（本文绘制）

1.提交：带幂等键落库，返回任务号，不等结果。

2.生成中：任务进入执行态，记录发起时间和超时上限。

3.回调到达：上游推通知，验签后更新状态。

4.轮询兜底：超过预期时长主动查一次，补上丢掉的回调。

5.对账：每天比对本地任务表与上游记录，找出不一致的。

五步里，第 3 步和第 4 步是很多人只做一半的地方——只做回调会丢单，只做轮询会重复查，两套并存才稳。

二、回调：先验签，再条件更新

回调接口第一件事不是解析业务字段，是验签。不验签等于把状态更新权交给了任何人。

# Flask 示例：回调入口
import hmac, hashlib, json
from flask import Flask, request, jsonify

app = Flask(__name__)
SIGN_SECRET = b"your-webhook-secret"

def verify(raw_body: bytes, signature: str) -> bool:
expect = hmac.new(SIGN_SECRET, raw_body, hashlib.sha256).hexdigest()
return hmac.compare_digest(expect, signature or "")

@app.post("/callback/image-task")
def callback():
raw = request.get_data() # 必须用原始 body 验签
if not verify(raw, request.headers.get("X-Signature", "")):
return jsonify(code=401, msg="bad signature"), 401

data = json.loads(raw)
task_id, status = data["task_id"], data["status"]

# 条件更新：只允许单向流转，旧状态不覆盖新状态
updated = db.execute("""
UPDATE image_task
SET status = :status,
image_url = :url,
finished_at = NOW()
WHERE task_id = :tid
AND status IN ('queued', 'running')
""", {"status": status, "url": data.get("image_url"),
"tid": task_id})

if updated == 0:
# 要么任务不存在，要么已经是终态（重复通知或乱序）
log.warning("callback skipped task_id=%s status=%s",
task_id, status)

return jsonify(code=0)

三个细节值得单独说：

验签用原始 body。先把 JSON 解析再序列化回去验签，字段顺序和空格变了，签名必然对不上。这个问题在本地测不出来，上线才发现。

用条件更新，不要先查后写。WHERE status IN (...) 就是条件更新，一句话解决并发覆盖。先 SELECT 再 UPDATE 是典型的竞态写法，两个回调同时到达时会互相覆盖。

`updated == 0` 要记日志。这个分支包含了三种情况：任务不存在、重复通知、乱序。不记日志，出问题只能靠猜。

三、兜底轮询：按 P95 定阈值，加随机抖动

回调一定会有丢的，网络抖动、服务重启、灰度发布都会导致丢。兜底不是可选项。

# 每 2 分钟扫一次"应该完成但还没完成"的任务
STALE_SQL = """
SELECT task_id FROM image_task
WHERE status IN ('queued', 'running')
AND submitted_at < NOW() - INTERVAL '3 minutes'
AND poll_count < 5
LIMIT 200
"""

def poll_pending():
for row in db.query(STALE_SQL):
detail = api.get_task(row["task_id"]) # 上游查询接口
if detail["status"] == "running":
sql = ("UPDATE image_task SET poll_count = poll_count + 1"
" WHERE task_id = :t")
db.execute(sql, {"t": row["task_id"]})
continue
apply_status(row["task_id"], detail["status"],
detail.get("image_url"))

阈值怎么定：按上游耗时的 P95 定，不要按平均值。平均值是 20 秒的时候，P95 可能是 60 秒。按平均值定阈值，会导致大量还在正常生成的任务被误判。

轮询频率加随机抖动。固定间隔重试会把失败时间点对齐成新的一波尖峰，加 random.uniform(0, 30) 秒的抖动就能打散。

次数要有上限。poll_count < 5 就是上限，连续查不到结果就转人工，不要无限重试——无限重试在夜里会一直占着任务，第二天还没人知道。

四、乱序：三种处理

同一个任务的多条通知，可能前后颠倒。处理原则只有一条：状态单向流转，回退一律丢弃。

除了条件更新，还有两处要防：

通知体里必须带状态，不能只带 ID。只有 ID 的通知无法判断这是"开始生成"还是"生成完成"，收到之后还得反查一次，白白多一次请求。

落库时以任务号 + 状态作为去重依据。重复通知在入库这一层就被拦掉，不用靠应用层判断。

上线初期先只记录不处理。把收到的通知按时间顺序记下来，观察几天真实乱序比例，再决定要不要加版本号。大多数上游的乱序是低频事件，先看清楚再设计。

五、对账：一天一次，专治对不上

回调和轮询都做了，还是会有差异。差异来源很多：回调延迟跨天、轮询漏了某批、上游自己的状态变更没通知。

对账任务放在低峰期跑，做三件事：

1. 拉昨天的全部任务，与本地表按任务号比对；

2. 本地是"生成中"、上游是"完成"的，补状态并补下载；

3. 本地是"完成"、上游查不到的，标记异常，人工确认。

对账本身也要幂等。同一份数据跑两次，结果应该一致。

六、状态口径：对外只暴露三种

后端状态可以有很多：queued、running、retrying、succeeded、failed、blocked。但前端只该看到三种：处理中、可下载、失败。

中间态一旦外透，产品经理会问"retrying 和 running 有什么区别"，客服会照着状态给用户解释，最后所有人都被内部实现绑住。

失败要分类，因为处理方式不同：

四类混在一起统计，看上去合格率九成，扣掉拦截和超时，实际能用的可能只剩七成。分开记，才知道该加并发还是该改提示词。

七、接入方式的选择

上面这套逻辑，自己写大约两三百行，不算复杂，但要做好回调验签、条件更新、兜底、对账四件事，需要一点耐心。

如果不想维护这些，也可以直接用平台的接口层。Flux Art 的开放接口在 https://open-api.flux-art.cn/openapi/v1，走的是同一套提交-回调模式，签名与重试的细节在文档里写清了。它是多模型 AI 视觉创作与生产平台，接口后面挂着多个图像与视频模型，换模型不用换接入方式。

八、小结

异步出图的状态同步，核心就三句话：

• 提交拿号，不等结果；
• 回调为主，轮询兜底，对账收尾；
• 状态单向流转，旧的不覆盖新的。

把这三点守住，剩下的都是细节。细节里最容易翻车的是验签和条件更新，两处都在代码里，写完记得用并发压一遍。