7.1 任务队列：run 丢给 worker 异步执行
第 6 章给 ResearchCopilot 配好了缓存与限流，服务能扛住更多请求了，但有一个更根本的问题一直没动：run 还是在 HTTP 请求里同步跑完的。用户 POST 一个 run，连接就挂在那里等——一次多跳检索加模型生成，几秒到几十秒不等。请求超时、网关断连、部署重启，任何一个环节抖动，用户就白等一场，而那个 run 的状态也随进程一起蒸发。
本节把 ResearchCopilot 推进到
researchcopilot/queue.py：POST /runs只写库、入队、立刻返回 202，真正的执行搬到独立 worker 进程；用 Redis 的BLMOVE做一条可靠队列，真连本机 Redis 与 Postgres 跑出一次完整生命周期。
7.1.1 为什么要把 run 从请求里搬走
同步执行的问题不是「慢」，而是慢的代价由连接承担。一次 run 要几秒到几十秒，这段时间里：
- 客户端得一直挂着连接，移动网络一切换就断；
- 反向代理 / 网关有超时（常见 30–60s），run 一超时，网关先返回 504，用户看到失败，可 worker 其实还在跑；
- 服务要扩容时，一个实例同时只能跑有限个同步 run，水平扩展被连接数绑死；
- 进程一重启，正在跑的所有 run 全部丢失，没有任何痕迹。
把执行搬进队列，这四条同时消失：请求变成一次「投递」，几十毫秒返回；执行在 worker 侧独立进行，进程崩了任务还在队列里。
7.1.2 裸队列为什么会丢任务
最省事的队列写法是 LPUSH 生产、BRPOP 消费：worker 从列表右端取一条，取到就执行。它在 demo 里跑得通，却有一个致命缺陷——任务在「取走」和「执行完」之间没有任何保护：
LPUSH queue run-1 # API 入队
BRPOP queue # worker 取走 run-1（此刻 run-1 已不在队列里）
↓ worker 崩了 / 被 kill / 机器断电
run-1 永久丢失 # 队列里没有它，内存里也没有它
BRPOP 是「取出即删除」，任务一旦被取走就从队列消失。worker 若在取走之后、完成之前崩溃，这条任务既不在队列、也没有结果，静默丢失。研究助理这种场景里，丢一个 run 意味着用户等了一个永远不来的报告。
| 取任务方式 | 语义 | worker 崩溃时 |
|---|---|---|
BRPOP queue | 取出即删除 | 任务丢失 |
RPOPLPUSH src dst | 取出并搬到备份列表 | 任务留在 dst，可重投 |
BLMOVE src dst LEFT RIGHT | 同上，阻塞版 | 任务留在 dst，可重投 |
正解是可靠队列：取任务时不是「删除」，而是「原子搬到另一个列表」。Redis 的 LMOVE（阻塞版 BLMOVE）就是这个动作——从 queue:runs 头取一条，同时推到 queue:runs:processing 尾。任务在 processing 里等 worker 确认；worker 跑完再从 processing 删掉，才算真正消化。崩溃时任务静静躺在 processing 里，由恢复程序重新投递。
7.1.3 环境：起 Redis 与 Postgres
本机两个服务映射在非默认端口（5432 与 6379 已被占用），等价容器命令：
docker run -d --name ab-pg -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=research -p 15435:5432 postgres:17-alpine
docker run -d --name ab-redis -p 16380:6379 redis:7-alpine
连接串分别是 postgresql://postgres:postgres@localhost:15435/research（实测 PostgreSQL 17.11，驱动 psycopg 3.3.6）与 redis://localhost:16380/0（服务端实测 Redis 7.4.11，客户端 redis 8.1.0）。连不上先 docker ps | grep -E 'ab-pg|ab-redis'。本节复用第 5 章的 sessions / runs / messages 三张表，run 的输入输出存在 messages.payload（jsonb），runs 只存状态与时间戳。
7.1.4 接口改造：写库 → 入队 → 立刻 202
第 3 章的 POST /sessions/{id}/runs 原本直接调 execute_run()；现在改成「先写库、再入队、马上返回」。顺序不能反——必须先把 run 落成 queued 再入队，否则 worker 可能先拿到任务、却发现库里还没有这条 run：
def enqueue(r, conn, run_id, session_id, workspace_id, inp) -> None:
# 1) 先写库：session + run(queued) + user 消息，一个事务
with conn.cursor() as cur:
cur.execute(
"INSERT INTO sessions (id, workspace_id, title, created_at, updated_at) "
"VALUES (%s, %s, %s, %s, %s) ON CONFLICT (id) DO NOTHING",
(session_id, workspace_id, "研究：向量数据库选型", now(), now()),
)
cur.execute(
"INSERT INTO runs (id, workspace_id, session_id, status, created_at) "
"VALUES (%s, %s, %s, 'queued', %s)",
(run_id, workspace_id, session_id, now()),
)
cur.execute(
"INSERT INTO messages (workspace_id, session_id, role, payload, created_at) "
"VALUES (%s, %s, 'user', %s, %s)",
(workspace_id, session_id,
json.dumps({"content": inp}, ensure_ascii=False), now()),
)
conn.commit()
# 2) 再入队：顺序不能反，否则 worker 可能读到不存在的 run
r.lpush(QUEUE, json.dumps({"run_id": run_id, "input": inp}))
接口因此从「跑完才返回」变成「受理即返回」——这正是第 3 章定下的 202 Accepted 语义：创建与执行是两件事，202 诚实地说「我收下了，但还没跑完」，返回体里带上 run_id 供客户端轮询。
7.1.5 worker：原子认领 + 执行 + 写回
worker 是一个独立进程，循环从队列里取任务。它有三步，每步都对应一个数据库动作：
def main(limit: int) -> None:
r = connect_redis()
conn = connect_pg()
log("worker 启动，等待任务（BLMOVE 阻塞）")
for _ in range(limit):
# 可靠取任务：原子地从 QUEUE 头移到 PROCESSING 尾，崩溃也不丢
raw = r.blmove(QUEUE, PROCESSING, 5, "LEFT", "RIGHT")
if raw is None:
log("空闲超时，退出")
return
task = json.loads(raw)
rid = task["run_id"]
log(f"取到 run={rid}：{QUEUE} -> {PROCESSING}（可靠队列）")
# 原子认领：只有 queued 才能变 running，attempts 自增
with conn.cursor() as cur:
cur.execute(
"UPDATE runs SET status='running', started_at=%s, "
"attempts=attempts+1 "
"WHERE id=%s AND status='queued' RETURNING attempts",
(now(), rid),
)
row = cur.fetchone()
conn.commit()
if row is None:
log(f" run={rid} 非 queued，跳过（已被其它 worker 领走）")
r.lrem(PROCESSING, 1, raw)
continue
log(f" 状态迁移 queued -> running（原子认领，attempts={row[0]}）")
answer = run_minagent(task["input"]) # 真实项目调卷一的 minagent.run
with conn.cursor() as cur:
cur.execute(
"UPDATE runs SET status='succeeded', finished_at=%s WHERE id=%s",
(now(), rid),
)
cur.execute(
"INSERT INTO messages (workspace_id, session_id, role, payload, "
"created_at) SELECT workspace_id, session_id, 'assistant', %s, %s "
"FROM runs WHERE id=%s",
(json.dumps({"content": answer}, ensure_ascii=False), now(), rid),
)
conn.commit()
# 确认完成：从 PROCESSING 移除，任务才算真正消化
r.lrem(PROCESSING, 1, raw)
r.lpush(DONE, raw)
两个「原子」值得说透。取任务是原子的：BLMOVE 一条命令完成「弹出 + 推入」，中间没有可乘之机，不会出现「弹出了但没推进去」的窗口。认领也是原子的：UPDATE ... WHERE id=%s AND status='queued' 依赖数据库的行级锁——如果两个 worker 同时拿到同一个 run（重投时可能发生），只有一个的 WHERE status='queued' 能命中，另一个 RETURNING 拿到空、直接跳过。幂等从这里就开始建立：同一个 run 无论被投递几次，状态机只允许它从 queued 迁移一次。
还有一处刻意的顺序：先更新状态、再写结果消息，两步在同一个事务里提交。这样要么「run 变 succeeded 且 assistant 消息落库」一起发生，要么都不发生；不会出现「状态说成功了、但消息还没写」的中间态。相比之下，把状态更新放在消息写入之后、又不包事务，就会给客户端制造一个「查状态是 running、查消息是空」的窗口——状态和它的产物必须同生共死。
7.1.6 真跑：一次完整的 run 生命周期
先起 worker（后台），再从 driver 侧入队三个 run，最后核对终态：
/tmp/agentbook/venv/bin/python worker.py 3 > worker.log 2>&1 &
/tmp/agentbook/venv/bin/python driver.py
wait; cat worker.log
driver 侧输出（Python 3.14.6）：
[2026-10-10T19:05:52+00:00] 入队 run=6fb23a51-82fa-4d69-ae2e-427af8491a54 写库(queued) -> LPUSH -> 返回 202
[2026-10-10T19:05:52+00:00] 入队 run=8c3b8f4f-aed3-40d3-bb3a-f103a5ec2523 写库(queued) -> LPUSH -> 返回 202
[2026-10-10T19:05:52+00:00] 入队 run=0644afb4-2024-4d7c-819e-276e24e919a3 写库(queued) -> LPUSH -> 返回 202
[2026-10-10T19:05:52+00:00] API 立刻返回：3 个 run 已受理（202），不等执行结果
worker 侧生命周期日志：
[2026-10-10T19:05:51+00:00] worker 启动，等待任务（BLMOVE 阻塞）
[2026-10-10T19:05:52+00:00] 取到 run=6fb23a51-82fa-4d69-ae2e-427af8491a54：queue:runs -> queue:runs:processing（可靠队列）
[2026-10-10T19:05:52+00:00] 状态迁移 queued -> running（原子认领，attempts=1）
[2026-10-10T19:05:52+00:00] 执行完成（56 ms）running -> succeeded
[2026-10-10T19:05:52+00:00] output: 围绕「对比 pgvector 与 Qdrant」，命中 3 条本地语料，结论见引用。
[2026-10-10T19:05:52+00:00] 取到 run=0644afb4-2024-4d7c-819e-276e24e919a3：queue:runs -> queue:runs:processing（可靠队列）
[2026-10-10T19:05:52+00:00] 状态迁移 queued -> running（原子认领，attempts=1）
[2026-10-10T19:05:52+00:00] 执行完成（60 ms）running -> succeeded
[2026-10-10T19:05:52+00:00] output: 围绕「RAG 的召回率怎么评」，命中 3 条本地语料，结论见引用。
[2026-10-10T19:05:52+00:00] 取到 run=8c3b8f4f-aed3-40d3-bb3a-f103a5ec2523：queue:runs -> queue:runs:processing（可靠队列）
[2026-10-10T19:05:52+00:00] 状态迁移 queued -> running（原子认领，attempts=1）
[2026-10-10T19:05:52+00:00] 执行完成（60 ms）running -> succeeded
[2026-10-10T19:05:52+00:00] output: 围绕「GIL 对并发的影响」，命中 3 条本地语料，结论见引用。
[2026-10-10T19:05:52+00:00] 处理完指定数量，退出
注意三点。其一，driver 的四行在 19:05:52 内全部打完——API 只做了写库和入队，不等执行，几十毫秒返回。其二，worker 每个 run 都走完 取到 → queued→running → succeeded 三段，状态是显式迁移的，不是「跑完直接写结果」。其三，attempts=1 说明每个 run 只被认领了一次，没有重复执行。最后核对终态（限定本 workspace）：
runs: [('succeeded', 3, 3)] # 3 个 run 全部 succeeded，attempts 合计 3
messages: [('assistant', 3), ('user', 3)]
queue:runs = 0 | processing = 0 | done = 3
queue:runs 和 processing 都归零、done 是 3——任务既执行了、也被确认了。如果 worker 在某个 run 执行途中崩溃，processing 里会留下那一条，而 queue:runs 已空——这正是可靠队列留下的「未完成任务」痕迹，恢复程序据此重投即可。
顺带一提，DONE 这个列表在生产里通常不会真的留着（任务执行完就可以丢了），本节保留它只是为了在演示里能直观看到「有几个任务走到了终点」。真实系统更关心的是 processing 里有没有堆积——那是 worker 跟不上或正在崩溃的直接信号。
7.1.7 同步执行 vs 异步队列
是不是所有 run 都该异步？不一定。两者各有适用面：
| 维度 | 同步执行 | 异步队列 |
|---|---|---|
| 返回时机 | 跑完才返回 | 受理即返回 202 |
| 客户端复杂度 | 直接拿结果 | 需轮询 / SSE 订阅 |
| 长任务（>30s） | 网关超时风险高 | 天然支持 |
| 进程崩溃 | 任务丢失 | 任务留在 processing，可重投 |
| 水平扩展 | 被连接数绑死 | 加 worker 即可 |
| 结果时效 | 强一致（拿到的就是最终） | 最终一致（要等状态翻转） |
| 调试 | 简单，栈一路到底 | 跨进程，要日志/追踪串联 |
| 适用 | 短任务、交互式问答 | 长任务、后台 digest、批处理 |
判断标准其实很朴素：任务时长是否超过你能接受的连接时长。交互式的一问一答（几百毫秒）同步更简单；而 ResearchCopilot 的多跳检索报告、周期 digest 都是分钟级，必须异步。工程上常见的是两条路都留：POST /runs 默认异步返回 202，客户端再决定是轮询还是订阅第 3.2 节的 SSE 事件流。
还有一条容易忽略的代价：异步把「一次调用」拆成了「两个状态」。同步模式下，「请求成功」等于「任务成功」；异步模式下，请求成功只代表「入队成功」，任务本身可能还在 queued、可能 running、也可能最终 failed。客户端必须学会读 run.status，后端必须保证每个 run 都有确定的终态——没有终态的队列比同步调用更糟，因为用户连「失败了」都不知道。这也是下一节要补死信的原因：一个 run 绝不能永远卡在 running。
7.1.8 真实环境会怎样
真实系统里，这套骨架的形状一行不改，但四处会变重：
- 队列不该只用 Redis 列表。Redis 列表适合「量不大、要低延迟」的场景；任务量大或要持久保障时，用 RabbitMQ / Kafka / 云 SQS，它们内建了重投、死信、可见性超时。本节用 Redis 是因为它把「可靠队列」的机制暴露得最清楚。
processing要有超时回收。worker 崩溃后任务留在processing，得有个 janitor 扫描「躺太久」的任务重新投递——这需要给每条任务记一个领取时间，第 8 章讲超时与泄漏时会补上。- worker 要能优雅退出。收到 SIGTERM 时应停止取新任务、把手头这个跑完再退，否则又制造一条留在
processing的任务。 - 队列要有可观测性。
queue:runs长度、processing长度、任务等待时长、执行时长都该进第 13 章的看板——队列积压是容量问题最早的信号。
本节验收清单：
-
POST /runs只写库并入队，立即返回 202 与run_id - worker 用
BLMOVE把任务从queue:runs原子搬到processing - run 状态迁移是原子的：
WHERE status='queued'认领，重复投递不会重复执行 - 执行完从
processing确认移除，任务不再残留 - 崩溃时未完成的任务留在
processing，可被重新投递
到这里，run 已经能安全地「投出去、跑起来、写回来」了。但现实里 worker 会遇到上游抖动——模型 503、网络超时，一次失败就判死太脆弱。下一节 7.2 重试、死信与幂等 会给这条队列加上「失败重试、重试耗尽进死信、重试必须幂等」的三件套。想回看 run 状态机与 202 的来历，翻 3.1 REST 设计：session / run / message 资源建模 。
阅读导航：上一节：6.3 限流与配额：按租户 token 预算的令牌桶 · 下一节：7.2 重试、死信与幂等 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。