文生视频异步任务网关治理实战：状态机、幂等与可靠回调

文生视频这波热度有多高不用我多说，但真正把这类业务从demo推到线上稳定跑的人，大概都绕不开一件事：异步任务怎么治理。视频生成不是普通接口调用，一个prompt丢进去，显卡要算几十秒甚至几分钟，整个交互模型天然就是异步的。文生视频和发邮件、生成图片这种异步任务还有一个本质区别：单任务耗时长、成本高、失败容忍度低。排队中的请求能不能及时处理，GPU资源够不够，任务跑挂了用户会不会原地爆炸，这些问题几乎全压在后端链路上，而首当其冲的就是任务网关层。

这篇文章就聊我做文生视频业务时，在异步任务网关层治理上踩过的坑和最终的解决方案。里面涉及的选型思路、状态机设计、代码片段、排查手段，都是线上验证过的。适合正在做或准备做AIGC视频类服务的后端同学参考，尤其是任务链路比较复杂、需要自己管理任务状态和回调通知的团队。

1. 为什么文生视频业务必须认真治理异步任务网关层

1.1 从一次线上事故说起：回调丢失带来的连环崩溃

先讲一个我们真实遇到的事故。上线初期，我们的架构非常简单：客户端调用生成接口，后端创建任务，丢给worker生成，worker完成后再把视频URL回调给客户端。问题就出在这个"回调"上。某天晚上高峰，回调服务因为一次发版导致的连接误关闭，大量回调消息直接丢掉，一批任务其实已经生成成功了，但客户端永远收不到通知，用户在前端看到的是"生成中"状态一直在转圈。比较尴尬的是，任务生成的视频其实都躺在对象存储里，客户就是拿不到。

这个事故的直接原因是回调不可靠，但往深了挖，根子在网关层设计。我们没有为回调设计持久化、重试和兜底查询，整个链路是"尽力而为"，而不是"保证送达"。这也是我后来把异步任务网关层单独拎出来治理的原因：文生视频这种高成本、长耗时的任务，网关层不是简单的转发代理，它其实是整个任务生命周期的状态中枢。

1.2 异步任务在文生视频架构中的位置

从整体架构来看，一条文生视频的完整链路大致是：

客户端 -> 网关层（任务提交/查询/回调通知） -> 队列 -> worker（GPU推理） -> 对象存储 -> 状态持久化

网关层在这里承担了三类工作。第一，承接所有任务请求，做参数校验、鉴权、限流；第二，维护任务状态流转，让用户随时能问到"我的视频生成到哪一步了"；第三，负责结果通知，通过轮询或回调把最终结果送达客户端。

说白了，网关层就是整个异步任务的"交通枢纽"。队列保证worker不会被请求打爆，状态表让任务有迹可循，回调让用户不用一直傻等。这三个环节里，任何一个设计得不够严谨，后面一定会出问题。

1.3 网关层治理要解决的四个核心问题

我做了这段时间之后，把异步任务网关层的核心问题收敛成四个：

• 可追踪：每个任务必须能查到全生命周期的状态，包括排队中、生成中、成功、失败、超时、取消，还得看到每个状态的耗时分布，否则出问题完全没法定位。
• 可限流：文生视频的GPU资源是稀缺且贵的，网关层必须从入口就控制并发和速率，否则后端会把显卡打爆，其他人全部排队。
• 可送达：生成结果无论通过回调还是轮询，必须可靠到达客户端手里，不允许"视频生成了但用户不知道"这种埋雷场景。
• 可自愈：任务处理过程中难免有网络抖动、worker异常、服务重启，网关层要能自动重试、超时置位、补偿修复，而不是靠人肉半夜爬起来改数据库。

这四个问题，其实就是我后面所有设计的出发点。先把这个想清楚，再去选型，就不会被各种框架带偏。

2. 任务状态机与数据模型设计

2.1 状态机设计：不要只有"成功"和"失败"

我见过不少团队做异步任务时，状态表里就三个值：pending、success、fail。短任务这么搞勉强能行，但文生视频绝对不行。一个视频任务可能要跑几十秒到几分钟，用户需要知道到底是在排队还是已经在生成，否则体验非常差。

我们最终敲定的状态机是这样的：

• CREATED：提交成功，任务已经落库，还没有入队
• QUEUED：任务已进入消息队列，等待worker消费
• PROCESSING：worker已拉取任务，正在生成视频
• SUCCEEDED：视频生成成功，结果地址已回写
• FAILED：生成失败，记录错误码和原因
• TIMEOUT：任务超时（排队超时或处理超时），由调度器扫描置位
• CANCELED：用户主动取消

每个状态还包括辅助字段，比如重试次数、最后更新时间等。状态流转规则我用一个小表列出来，方便大家对照：

这里踩过的一个坑是：一开始我们没区分CREATED和QUEUED，认为提交即入队。后来发现，消息发到队列也可能失败（比如队列抖动），如果状态表里只有"提交成功"，用户看到的状态就会误导人。所以建议宁可多两个状态，也不要为了简单丢失中间状态。

2.2 任务ID和幂等键：一个单独的全局ID还不够

任务ID本身很简单，我们用雪花算法生成64位ID，数据库主键用bigint，对外返回字符串。真正容易踩坑的是幂等键。

文生视频这种高成本业务，一定要支持幂等提交。因为客户端的网络环境非常不稳定，一个请求超时后，客户端会自动重试，如果后端没有幂等处理，同一个prompt可能被提交进去两次，等于GPU白白算了两遍，成本直接翻倍。更麻烦的是，用户下一次刷新页面还能看到两次记录，体验也很差。

我们的做法是：客户端在提交请求时，根据"用户ID + prompt内容 + 参数hash"生成一个idempotency_key（也允许客户端自己传一个UUID），服务端在创建任务前先按这个key查数据库，如果已存在就直接返回已有任务的task_id和状态，不会重复创建。底层就靠数据库唯一索引兜底，避免并发场景下的竞态。

-- 任务表中加唯一索引 UNIQUE KEY uk_idempotency (user_id, idempotency_key)

注意，这个唯一索引不能跨用户，否则不同用户提交了相同内容的prompt会被误判成重复。我们最开始就犯了这个错，索引只建在了idempotency_key上，结果两个用户生成"一只猫在跑步"，第二个用户的请求直接被挡住了。

2.3 状态表设计：除了状态，还要留足排查字段

任务状态表是整个网关层的数据核心。我直接给出我们线上使用的核心字段，不一定全适用，但可以当参考：

CREATE TABLE generation_task ( task_id BIGINT PRIMARY KEY, user_id VARCHAR(64) NOT NULL, idempotency_key VARCHAR(128) NOT NULL, prompt TEXT NOT NULL, status VARCHAR(20) NOT NULL DEFAULT 'CREATED', priority TINYINT NOT NULL DEFAULT 5, queue_name VARCHAR(32) NOT NULL DEFAULT 'default', worker_id VARCHAR(64) DEFAULT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, queued_at DATETIME DEFAULT NULL, started_at DATETIME DEFAULT NULL, finished_at DATETIME DEFAULT NULL, result_url VARCHAR(512) DEFAULT NULL, error_code VARCHAR(32) DEFAULT NULL, error_msg VARCHAR(512) DEFAULT NULL, retry_count INT NOT NULL DEFAULT 0, callback_url VARCHAR(512) DEFAULT NULL, callback_status VARCHAR(20) NOT NULL DEFAULT 'PENDING', callback_retry INT NOT NULL DEFAULT 0, callback_payload JSON DEFAULT NULL, UNIQUE KEY uk_idempotency (user_id, idempotency_key), KEY idx_status_updated (status, updated_at), KEY idx_callback_status (callback_status, callback_retry) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

这里有三个索引值得说一下。idx_status_updated是为了让调度器快速扫描超时任务，比如WHERE status IN ('QUEUED','PROCESSING') AND updated_at < NOW() - INTERVAL 5 MINUTE；idx_callback_status是给回调重试调度器用的，专门捞那些回调还没送达的任务。另外，queued_at、started_at这些时间戳字段不要偷懒不写，后面做耗时分析、性能排查全靠它们。

3. 提交链路与队列选型的实际考量

3.1 提交接口的处理流程：先写库，再发消息

任务提交接口是整个链路的第一道关卡。我们最终实现的处理顺序是：

1. 参数校验：检查prompt是否为空、长度是否超限、图片是否合法、用户是否还有配额；
2. 幂等检查：根据idempotency_key查库，命中就直接返回已有任务；
3. 创建任务：插入一条status=CREATED的记录，拿到task_id；
4. 发送队列消息：把task_id和必要参数推送到MQ；
5. 更新状态：如果消息发送成功，把status更新为QUEUED；如果发送失败，把status置为FAILED。

这里最核心的一个设计原则是：先落库，再发消息，消息发送成功后才更新状态。为什么要这样？大家想一个问题：如果先发消息再落库，worker那边已经把任务拉起来开始算了，数据库里却找不到这个任务，状态从何谈起？反过来，先落库再发消息，即使消息发送失败，任务至少是CREATED状态，调度器可以捞出来重新处理。

关于消息发送成功但数据库更新失败的情况，我的建议是不要在这个环节做太复杂的事务。我们曾经尝试把"插入任务"和"更新状态"放在一个数据库事务里，并且把发送MQ也做成事务消息。后来发现，事务消息在并发量上来之后，要么是发送超时导致整个事务回滚，要么是消息发出去了但事务迟迟没提交，消费者拿不到数据。很折磨。最终我们简化成了上面这种"本地事务更新数据库 + 消息发出后异步确认"的模式，配合一个补偿任务：每隔几分钟扫描一次CREATED状态超过N分钟的任务，重新补发消息。

3.2 队列选型：Redis List还是MQ？

文生视频业务对队列的核心要求是：容量要大、支持延迟重试、能支撑一定的优先级。我们先后对比了几种方案：

• Redis List（BRPOPLPUSH）：简单，延迟低，但不够灵活，重试、死信、优先级都不好做，而且大任务堆积时Redis内存会告急。
• RabbitMQ：功能完整，支持延迟队列、死信队列、优先级队列，运维也相对成熟。
• Kafka：吞吐量极高，但更适合事件流场景，用来做任务队列有点大材小用，而且Kafka的按key消费和重试语义不如RabbitMQ直观。

我们最终选了RabbitMQ。一个很重要的原因是，文生视频的任务量其实没有想象中那么大，几十条queue的吞吐完全够用，RabbitMQ的重试和TTL机制用起来很舒服。具体上我们建了三个队列：

• task.submit.queue：普通任务提交，worker从这里消费；
• task.delay.queue：延迟重试队列，比如worker处理失败后，把消息丢到这里，3分钟后重新入队；
• task.dlx.queue：死信队列，处理失败超过N次的消息进死信，便于人工捞出来分析。

优先级这块，RabbitMQ的x-max-priority参数实测有效，但别把优先级档位设置太多，因为每个优先级其实是独立的内部队列，太多优先级反而增加调度开销。我们用3档：低（1）、普通（5）、高（10），高优先级一般给付费用户或手动重新生成的任务。

3.3 worker消费的坑：先改状态，还是先处理任务？

worker消费消息后的顺序竟然也是个翻车点。我们最早是拿到消息就直接开始GPU推理，推理完再更新任务状态。看起来没毛病，但问题出在：如果worker在推理过程中突然宕机，消息已经从队列里拿走了，数据库里任务还停在QUEUED；等worker重启，这条消息早就丢了，任务就永远卡在QUEUED状态，没人去管。

这里推荐的做法是：worker拿到消息后，第一件事就是把数据库状态从QUEUED更新为PROCESSING，并且记录worker_id，然后再去调GPU推理。这样即使worker中途宕机，至少状态是PROCESSING，调度器扫描超时任务时就能发现它并重新投递。另外，状态更新和消息确认的顺序也要注意。我们总结的经验是：先更新数据库状态为PROCESSING，再给MQ确认ack。因为一旦ack，这条消息就彻底没了，如果状态没更新成功，任务就处于"队列里找不到、数据库还是QUEUED"的尴尬阶段，只能靠补偿任务捞。先更新状态再ack，虽然理论上有"更新成功但ack失败导致消息重复消费"的风险，但重复消费最多是重复生成一次视频，我们可以用数据库里的task_id做去重判断，比消息丢失更好恢复。

这段经验可以总结成一句话：异步链路中，宁可重复消费，也不要消息丢失。重复能靠幂等去重，丢失就只能靠扫描补偿，而补偿是有时间窗口的，用户根本等不了。

4. 回调通知与状态查询：两条腿走路

4.1 回调通知为什么不能做成"尽力而为"

文章开头说的那次事故，就是回调丢消息。从那以后，我们对回调做了彻底重构。核心思路借鉴了事务发件箱（Transactional Outbox）模式：业务操作和通知事件在同一个本地事务里搞，通知事件先落库，再通过异步调度器把事件推送给客户端。

具体实现是这样的：

1. worker完成生成后，把result_url写回任务表，同时把callback_status置为PENDING，表示"有一个回调待发送"；
2. 回调调度器每隔几秒扫描callback_status='PENDING'的任务；
3. 发现待发送任务后，向客户端的callback_url发一条HTTP POST请求，请求体包含task_id、status、result_url、error_info等；
4. 如果客户端返回2xx，把callback_status更新为SUCCEEDED；否则记录错误，callback_retry加1，等待下一轮扫描重试；
5. 重试达到上限（比如5次）后，把callback_status置为FAILED，此时只能靠客户端轮询兜底。

这个方案的好处是，回调消息不再依赖MQ的可靠性，而是躺在自己的数据库表里，只要数据库不丢，回调就一定能被重试到。和纯MQ方案相比，虽然实时性稍差（扫描间隔几秒），但对文生视频这种"用户等几十秒都不在乎"的场景完全够用。

# 伪代码：回调调度器核心逻辑 def process_pending_callbacks(): tasks = db.query( "SELECT * FROM generation_task " "WHERE callback_status = 'PENDING' AND callback_retry < 5 " "ORDER BY finished_at ASC LIMIT 100" ) for task in tasks: try: resp = http.post(task["callback_url"], json={ "task_id": task["task_id"], "status": task["status"], "result_url": task["result_url"], "error_code": task["error_code"], }, timeout=10) if resp.status_code == 200: db.execute( "UPDATE generation_task SET callback_status='SUCCEEDED' WHERE task_id=?", task["task_id"] ) else: db.execute( "UPDATE generation_task SET callback_retry=callback_retry+1 WHERE task_id=?", task["task_id"] ) except Exception: db.execute( "UPDATE generation_task SET callback_retry=callback_retry+1 WHERE task_id=?", task["task_id"] )

注意回调请求的超时不能设太长，10秒足够了。如果客户端回调接口响应慢，说明对方服务也有问题，重试几次就会自动放弃，不用死磕。

4.2 幂等回调：客户端收到两次回调也不怕

回调重试带来的副作用是：客户端可能会收到重复回调。比如第一次回调其实已经抵达了，只是服务端因为网络超时报错，于是重试了第二次。如果客户端没有幂等处理，看到两条回调就可能会创建两个下载任务、收两次费用。

所以回调的请求体里一定要带上task_id，并且建议客户端的回调处理接口按task_id做幂等。我们还在响应头里加了Idempotency-Key，跟请求体里的task_id保持一致。另外，我们约定回调是"最终状态通知"，只有SUCCEEDED或FAILED这两种终态才会触达回调，中间状态一律不通知，尽量减少客户端的处理复杂度。

4.3 查询接口：轮询的时机和频率

尽管做了可靠的回调通知，轮询查询接口依然是必须存在的兜底。因为回调可能因客户端服务故障而最终失败（callback_status=FAILED），此时用户只能通过轮询拿到结果。

状态查询接口做得非常简单：GET /api/v1/tasks/{task_id}，返回任务当前状态和时间戳。但轮询频率怎么定，其实有讲究。我们走过一段弯路：前端每500ms轮询一次，导致查询接口的QPS是提交接口的几十倍，给数据库造成了不小的压力。后来我们调整策略，让前端做自适应轮询：任务刚提交时每2秒查一次；超过30秒没结果时改成每5秒查一次；超过2分钟后改成每10秒查一次。

从网关层角度，我们还在查询接口上做了short-circuit优化：如果任务已经是终态（SUCCEEDED或FAILED），查询结果直接走本地缓存，不再查数据库，因为终态数据几乎不会再变化。这个优化把查询接口的数据库压力降了大概70%。

5. 超时、限流与任务取消的治理细节

5.1 超时设计：排队超时和处理超时分开

文生视频任务超时是很容易被忽略但又很致命的问题。我们定义了两种超时：

• 排队超时：任务进入QUEUED后，如果在N分钟内没有被任何worker消费，视为排队超时。我们的排队超时设为3分钟，因为高峰期确实会有排队，这个值给得比较宽。
• 处理超时：任务进入PROCESSING后，如果在M分钟内没有变成终态，视为处理超时。视频生成的耗时跟prompt长度、分辨率、帧数都有关系，我们最初设为3分钟，后来发现高清视频偶尔要4分钟，所以调到5分钟。

超时检测靠一个定时调度器完成，每分钟扫描一次：

SELECT task_id FROM generation_task WHERE status = 'QUEUED' AND queued_at IS NOT NULL AND updated_at < NOW() - INTERVAL 3 MINUTE; SELECT task_id FROM generation_task WHERE status = 'PROCESSING' AND started_at IS NOT NULL AND updated_at < NOW() - INTERVAL 5 MINUTE;

扫描到超时任务后，不能直接置成FAILED就完事。我们的处理是：排队超时的任务先尝试重新投递到队列，重试2次后还是超时，才置为FAILED，错误码写成QUEUE_TIMEOUT。处理超时的任务，先检查worker节点是否还活着（通过worker心跳表），如果worker活着，可能是生成慢，再多给一点宽限时间；如果worker已经失联，立即把任务重新入队。

5.2 限流：从入口控制并发

文生视频的GPU资源是硬约束，Active状态的任务数量一旦超过GPU卡数，剩下所有任务都会积压。所以网关层的限流要同时cover几个维度：

• 用户维度限流：每个用户同一时刻最多N个未完成任务。我们设的是3，超出直接返回TOO_MANY_PENDING_TASKS，让用户等已有任务完成再提交。
• 全局并发限流：全系统同时处于PROCESSING状态的任务数不能超过GPU卡数乘以一个系数（比如1.5，留一点buffer）。超出的任务卡在QUEUED阶段，不进入worker。这样即使有人恶意刷接口，GPU也不会被打爆。
• 提交速率限流：每用户每分钟最多提交M次，防脚本刷。

限流发生在提交接口的最前面，先限流再走幂等和落库，否则恶意请求会把数据库打垮。我们用的就是一个简单的Redis计数器，INCR加过期时间，代码几行就搞定，效果很稳定。

5.3 用户取消：终态之前怎么处理

任务取消也是个容易出问题的点。用户点了取消，如果任务还在排队中（QUEUED），好办，直接更新状态为CANCELED，再从队列里把消息捞出来丢掉即可。但如果在PROCESSING状态，GPU已经在算视频了，这时候取消有两个选择：

• 硬取消：直接告诉worker任务取消了，让它停止推理，释放GPU。省资源，但有可能生成一半的视频文件残留，需要清理。
• 软取消：让worker继续算完，但结果不再通知用户，只在后台保留。浪费资源，但逻辑简单。

我们最终的策略是：默认软取消，用户可以主动选"立即停止"来走硬取消。因为从产品角度，用户取消后往往还会回来再生成一次，软取消让worker算完，其实下一次的生成结果可以直接复用。不过这只是我们的产品取舍，不一定适合所有场景，大家根据自己的成本模型来定。

6. 常见问题与排查技巧实录

6.1 高频故障速查表

我把这段时间遇到的高频问题整理成了一个排查速查表，遇到同类问题可以直接对号入座：

6.2 排查状态不一致的三个技巧

异步任务最怕的是"数据库状态"和"实际业务状态"对不上。排查这种问题，我总结出三个比较实用的技巧。

第一，看时间戳。排到状态卡住的任务时，先看queued_at、started_at、finished_at，这几个时间戳一对比，很快能判断卡在哪个环节。比如started_at不为空但状态还是PROCESSING且超过5分钟，那基本就是worker处理中异常退出。

第二，看worker_id。PROCESSING状态一定要记录worker_id，排查时拿着worker_id去查这台机器的日志和心跳，能确认这台机器是否还活着。没有worker_id的话，这个问题会变得非常难查，只能盲猜。

第三，回放补偿。不要一次性把状态不对的任务全部重置，稳妥的做法是先查出来，按任务ID小批量处理，比如一次50条，处理完跑一遍状态统计，确认没问题再处理下一批。大范围直接update是一次高风险操作，很容易把还正常跑着的任务也带偏。

6.3 数据补偿脚本的教学示例

最后给一个非常实用的补偿脚本思路。当调度器因为各种原因漏处理了一批卡住的任务时，你需要一个人工或半自动的补偿流程。我们用的就是一个Python脚本配合SQL查询，先把候选任务捞出来，再逐个决策：

#!/usr/bin/env python3 # 补偿脚本：扫描超时未完成的任务并重置 import mysql.connector conn = mysql.connector.connect(...) def find_stuck_tasks(limit=100): cur = conn.cursor(dictionary=True) cur.execute(""" SELECT task_id, status, created_at, started_at, worker_id FROM generation_task WHERE status IN ('QUEUED', 'PROCESSING') AND updated_at < NOW() - INTERVAL 10 MINUTE ORDER BY created_at ASC LIMIT %s """, (limit,)) return cur.fetchall() def requeue_task(task_id): # 重新投递到MQ，这条消息需要包含完整task信息 mq.publish("task.submit.queue", {"task_id": task_id}) for t in find_stuck_tasks(50): if t["status"] == "PROCESSING": # 先确认worker是否存活，如果存活且生成中，跳过 if worker_alive(t["worker_id"]): continue requeue_task(t["task_id"]) print(f"requeued task {t['task_id']}")

注意，补偿脚本一定要有--dry-run参数，先跑一遍只打印不执行，确认无误后再真正执行。我刚开始写补偿脚本的时候，也在生产上吃过亏，一激动直接跑了个update，把正常任务的状态都改乱了，后来老老实实先dry-run。

7. 写在最后：一点实战体会

回看这一路，文生视频业务的异步任务网关层，最难的不是技术选型，而是把"任务全生命周期"这个思维贯穿到每一个细节里。状态机、幂等、回调、超时、限流，单独拎出来哪一个都是老话题，但真正合在一起面对真实业务的时候，总会有不少意料之外的暗坑。最大的体会就一句话：异步任务网关层的核心不是"快"，而是"可靠"，是让每一个任务都有一条清晰可见、可追踪、可补偿的生命周期路径。

如果你正在做类似的AIGC视频业务，我的建议是前期宁可多写几张表和几个调度器，也不要在可靠性和可观测性上省钱。那些看起来麻烦的机制，比如outbox回调、超时扫描、补偿任务，几乎都是等线上出过事故之后才补上的，早期一旦缺失，后期补的成本会高很多。最后再分享一个小技巧：每个关键状态流转的地方都打一条结构化日志，带上task_id和状态，排查问题的时候，一条链路从头拉到尾，比任何监控面板都好用。