我的 AI 应用第一版，所有任务都这样处理：用户点"生成"，前端发一个 HTTP 请求，后端在这个请求里同步跑完整个生成流程，然后返回结果。
上线第三天就开始出事。上午十点，一个用户提交了 200 页文档做批量生成，这个请求跑了 12 分钟还没结束。这 12 分钟里，整个 Web 服务一个新请求都接不住——所有 worker 都在处理这一个长任务。用户等不及刷新又提交了一次。又一个 12 分钟。
症状比原因更值得记
那天的故障现象是：CPU 没打满，但所有接口都超时；内存监控正常；队列里什么都没有——因为压根没有队列。
我一开始怀疑是数据库锁，后来发现就是最朴素的問題：处理线程被占住了。8 个 worker 线程，8 个长任务，就没有线程去接新请求。
这类故障不像崩溃那样有明显的起点。你会先看到"某个接口变慢"，慢慢变成"全挂"，中间夹杂着奇怪的 502，因为负载均衡器把超时连接重置了。
队列不是"更好的架构"，是"必须的东西"
我换成了队列 + 独立 worker：
Web 服务 → 投递任务到队列 → 立刻返回 task_id
↓
Worker 消费
↓
写结果到存储 → Web 服务轮询/推送第一个好处不是解耦，是Web 进程再也不会被长任务拖垮。哪怕 worker 全忙，Web 只负责收单。
第二个好处是重试变得可能：原来处理到一半失败，用户得手动重试；现在 worker 崩了，任务自动回队列。
第三个好处是能限流：同一用户最多 5 个任务在跑，这是队列的语义。
选哪种队列
我用过三种，各有各的位置：
| 方案 | 适合 | 坑 |
|---|---|---|
| 数据库表当队列 | 任务量小（每天几百个） | 抢占和锁要自己处理 |
| Redis List/Stream | 中等量，要快 | 任务丢了就没了，除非开持久化 |
| RabbitMQ / SQS 类 | 任务量大，要可靠投递 | 运维成本高一点 |
如果任务量不大，我建议别引入新组件，直接用数据库表：
CREATE TABLE tasks (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT NOT NULL,
kind TEXT NOT NULL,
payload JSONB NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
attempts INT NOT NULL DEFAULT 0,
max_attempts INT NOT NULL DEFAULT 3,
available_at TIMESTAMPTZ NOT NULL DEFAULT now(),
locked_at TIMESTAMPTZ,
worker_id TEXT,
error TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX idx_tasks_claim ON tasks (status, available_at)
WHERE status = 'pending';worker 抢占任务用一条 UPDATE 就够，不需要额外的分布式锁：
UPDATE tasks
SET status = 'running',
locked_at = now(),
worker_id = %s,
attempts = attempts + 1
WHERE id = (
SELECT id FROM tasks
WHERE status = 'pending' AND available_at <= now()
ORDER BY available_at
LIMIT 1
FOR UPDATE SKIP LOCKED
)
RETURNING *;FOR UPDATE SKIP LOCKED 是关键。多个 worker 同时执行这条语句时，各自拿到不同的行，不会互相阻塞。我早期用 SELECT 然后 UPDATE，结果多个 worker 抢到同一个任务，跑了三遍。
长任务必须自己处理的四件事
队列只负责"让任务被跑起来"，语义还得自己写。这四件事漏一个都出事。
超时
GPU 推理卡住是常事（显存竞争、CUDA 死锁）。没有超时，一个卡住的任务会把 worker 占到天荒地老。 我给每个任务设了三层超时：单次执行的墙钟超时（超过就杀掉进程，任务回队列）、单次模型调用的超时（HTTP 客户端层设）、以及整个任务的 deadline（超了就标记失败，不再重试）。
TASK_TIMEOUT_SEC = 300 # 单次执行上限
DEADLINE_SEC = 1800 # 任务总时长上限取消
用户点"停止"，任务得真的停。实现上有个坑：取消请求会发到 Web 服务，而任务在 worker 里。
我的做法是 Web 服务写一个 cancelled_at 字段，worker 在每个可中断的阶段（切分文档、每生成一张图）检查一次。
worker 不能直接被 kill——进程里可能有正在写的文件、正在开的连接，硬杀会留半成品。
幂等
这是最重要的一条，也是我最后才补上的。
我遇到过的情况：用户提交批量生成，worker 处理到第 30 张图时因下游超时失败。任务重试，从第 1 张重跑，前 29 张的结果被覆盖，产生了 30 次计费调用。
修复方式是让每一步都有幂等键：
def process_batch(task_id):
items = get_items(task_id)
for idx, item in enumerate(items):
if check_cancelled(task_id):
return
# 幂等键由 task_id + 序号 + 内容哈希决定
key = f"{task_id}:{idx}:{hash_item(item)}"
result = cache.get(key)
if result is None:
result = do_generate(item)
cache.set(key, result, ttl=86400)
save_step_result(task_id, idx, result)save_step_result 用 (task_id, idx) 做唯一键，重复写入不会产生两份。
更彻底的做法是把任务拆成一堆子任务，每个独立可重试。
重试的分类
不是所有失败都该重试。
def should_retry(exc, attempts):
if attempts >= 3:
return False
if isinstance(exc, ValidationError):
return False # 参数错了，重试一百次也没用
if isinstance(exc, RateLimitError):
return True # 该退避后重试
if isinstance(exc, TimeoutError):
return True
return False指数退避 + 抖动必须一起用。纯指数退避在集体故障恢复那刻会惊群——所有任务同时重试，把刚恢复的下游再打死。
delay = min(base * (2 ** attempts), max_delay) * random.uniform(0.8, 1.2)定时任务：多副本是最容易翻车的地方
定时任务比队列任务更容易出问题，因为它"看起来总能跑"。
我第一次部署时在应用里加了个定时器，每小时跑一次数据同步。开发环境单副本一切正常，上了生产三副本，第二天早上收到三份一样的邮件。
原因是每个副本都跑了自己的定时器。这不是 bug，是设计如此——应用副本不该假设自己是唯一的。解决办法是加一把分布式锁：
def try_acquire_lock(name: str, ttl: int = 3600) -> bool:
key = f"cron_lock:{name}"
return redis.set(key, instance_id, nx=True, ex=ttl)nx=True 保证只有第一个能拿到。ex 保证进程挂了锁能自动释放，不然一次崩溃就让定时任务永久停摆。
另一个更省事的做法是把定时任务跟 Web 服务分开部署，单独进程跑定时器。我后来两个都做了——定时器独立 + 锁。
队列深度是个必须监控的指标
我给自己定的一条线是队列深度超过 500 就告警。
为什么是 500？因为它跟处理速度有关：worker 平均每分钟处理 20 个任务时，500 个任务就是 25 分钟积压，体验崩了。
所以告警规则不该只写"队列 > 500"，而应该写成"按当前处理速率，预计消化时间超过 5 分钟"。
def estimated_drain_minutes(queue_depth, completed_last_10min):
rate = completed_last_10min / 10
if rate <= 0:
return float('inf')
return queue_depth / rate最后几点
worker 的资源限制要写死。 跑 GPU 推理的 worker 和跑数据库查询的 worker 内存画像完全不同，混在一个池子里，一个内存吃多了会把别人挤死。
worker 要能被安全重启。 部署新版本时先让 worker 处理完手上的任务再退出，然后起新的。直接重启的结果是正在跑的任务全丢。
Web 服务只返回 task_id，不返回结果，结果通过轮询或 WebSocket 推。这是从"同步请求"到"异步任务"最核心的改变。 具体用什么队列取决于任务量和可靠性要求；换成哪种队列，上面这些语义都得自己写。