《AI Agent 工程实战》7.3 定时与延迟任务：周期语料更新与日报

7.3 定时与延迟任务：周期语料更新与日报

7.1 和 7.2 处理的是「用户提问触发的 run」——它们由 API 入队，worker 消费。但 ResearchCopilot 还有一类任务不由用户触发：语料索引得定期重建（否则新文档检索不到）、每个 run 完成后要延迟几分钟生成一份摘要、每天要跑一次日报。这些任务没有 HTTP 请求来「投递」它们，得由调度层自己发起。

本节把 ResearchCopilot 推进到 researchcopilot/scheduler.py：用 Redis 的 SET NX PX 加 token 校验做分布式锁，让多个 worker 副本抢同一个周期任务时只有一个执行；再用有序集合做延迟任务，真跑两个并发 worker 争抢同一把锁与一次延迟触发。

7.3.1 两类「自己动」的任务

调度任务分两种，触发方式和失败影响都不同：

两者的共同点是都由时间驱动，而不是由请求驱动。这带来一个新问题：多个 worker 副本会各自「看到」同一个时间点。如果每个副本都在「整点」醒来执行「重建索引」，索引就会被重建 N 次——纯属浪费，重则并发写坏索引。

区分这两类任务，还有一个容易忽略的维度：触发是「时间到」还是「事件后」。周期任务是「每过一个固定间隔就再来一次」，它不关心上一次的结果；延迟任务是「某个事件发生后，等一段时间再执行一次」，它带着事件的上下文（比如「哪个 run 完成了」）。这个区别决定了实现方式：周期任务靠「循环 + 抢锁」，延迟任务靠「把到期时间写进有序集合」。

7.3.2 多 worker 抢同一个定时任务的问题

假设 ResearchCopilot 部署了 3 个 worker 副本，每个副本里都跑一个「每 5 分钟重建索引」的循环。会发生什么？

• 3 个副本同时到点，3 个都开始重建索引——三份完全相同的重活同时开跑；
• 如果重建是「先清空再写入」，A 清空后 B 还在读，读到一半的空索引，检索全部落空；
• 每个副本都算一遍，CPU 与 IO 三倍浪费。

根因是：定时任务需要「全局唯一执行」，而副本之间不共享内存。单机时代用一个进程内锁就够了；多副本时代，这个锁必须放在所有副本都能访问的外部系统里——Redis 正好合适。

7.3.3 分布式锁：SET NX PX + token 校验

一把合格的 Redis 分布式锁要满足三点：互斥（同一时刻只有一个持有者）、防死锁（持有者崩了锁能自动释放）、防误删（不能删掉别人的锁）。三点分别对应三个机制：

import redis

# 只有锁的持有者（token 相同）才能删；否则会删掉别人的锁
RELEASE_LUA = """
if redis.call('get', KEYS[1]) == ARGV[1] then
  return redis.call('del', KEYS[1])
else
  return 0
end
"""

# 续期同理：token 不匹配就不动，避免把已易主的锁延长
RENEW_LUA = """
if redis.call('get', KEYS[1]) == ARGV[1] then
  return redis.call('pexpire', KEYS[1], ARGV[2])
else
  return 0
end
"""


def acquire(r: redis.Redis, name: str, ttl_ms: int, token: str) -> bool:
    """抢锁：SET key token NX PX ttl，原子且带过期，防死锁。"""
    return bool(r.set(f"lock:{name}", token, nx=True, px=ttl_ms))


def release(r: redis.Redis, name: str, token: str) -> bool:
    """释放：先比对 token 再删，绝不 DEL 别人的锁。"""
    return bool(r.eval(RELEASE_LUA, 1, f"lock:{name}", token))


def renew(r: redis.Redis, name: str, token: str, ttl_ms: int) -> bool:
    """续期：长任务在持有期内续租，避免任务没跑完锁先过期。"""
    return bool(r.eval(RENEW_LUA, 1, f"lock:{name}", token, ttl_ms))

SET key token NX PX ttl 是整把锁的核心。NX 保证只有 key 不存在时才能设置成功——这就是互斥，多个副本同时来，只有一个 SET 成功。PX ttl 给 key 设过期时间——这就是防死锁，持有者即使崩溃、来不及释放，锁也会在 ttl 后自动消失，其他副本能重新抢。SET NX PX 必须是一条原子命令：若拆成「先 SETNX 再 EXPIRE」两条，两条之间崩了就会留下一个永不过期的锁，把所有人挡在门外。

token 是每个持有者生成的唯一值（本节用 uuid4().hex），它是「防误删」的关键。设想这样的时序：A 抢到锁、执行时间超过了 ttl、锁自动过期；B 抢到锁；这时 A 执行完了，如果 A 直接 DEL lock:key，就把 B 的锁删了。解法是释放时先比对 token——只有锁里的值还是我的 token，才允许删。比对 + 删除必须在 Redis 端原子完成，所以用 Lua 脚本（GET 后 DEL 两条命令之间有窗口）。

7.3.4 周期任务：拿不到锁就跳过

有了锁，周期任务的写法就清晰了：每到一个周期，先抢锁，抢到才执行，抢不到说明别的副本正在跑，本轮跳过：

def main(name: str, start: float) -> None:
    r = connect_redis()
    token = uuid.uuid4().hex          # 每个副本一个唯一 token
    for i in range(1, 4):             # 三个周期
        target = start + 0.6 * i
        time.sleep(max(0.0, target - time.time()))
        if acquire(r, LOCK, TTL_MS, token):
            log(name, f"第 {i} 轮：拿到锁，执行 digest（扫描 128 个文档）")
            time.sleep(HOLD)          # 持锁执行
            release(r, LOCK, token)
            log(name, f"第 {i} 轮：完成并释放锁")
        else:
            log(name, f"第 {i} 轮：未拿到锁（已被占用），跳过本轮")

注意「拿不到就跳过」是刻意的：周期任务通常是幂等的重活，少跑一轮无非是数据晚更新一个周期，比「排队等着跑」简单得多。如果业务要求「每轮都必须跑」，那就不该用周期循环 + 抢锁，而该用一个「把任务投进队列」的调度器——调度器负责不重不漏地投递，worker 负责执行。

7.3.5 真跑一：两个 worker 争抢同一把锁

起两个 worker 副本 A 与 B，让它们对齐同一个时钟起点，各跑三个周期：

START=$(python -c "import time;print(time.time()+0.8)")
python sched_worker.py A "$START" > A.log 2>&1 &
python sched_worker.py B "$START" > B.log 2>&1 &
wait; cat A.log B.log

worker-A 的真实输出：

[2026-10-10T19:07:49+00:00] worker-A 第 1 轮：拿到锁，执行 digest（扫描 128 个文档）
[2026-10-10T19:07:49+00:00] worker-A 第 1 轮：完成并释放锁
[2026-10-10T19:07:49+00:00] worker-A 第 2 轮：未拿到锁（已被占用），跳过本轮
[2026-10-10T19:07:50+00:00] worker-A 第 3 轮：拿到锁，执行 digest（扫描 128 个文档）
[2026-10-10T19:07:50+00:00] worker-A 第 3 轮：完成并释放锁

worker-B 的真实输出：

[2026-10-10T19:07:49+00:00] worker-B 第 1 轮：未拿到锁（已被占用），跳过本轮
[2026-10-10T19:07:49+00:00] worker-B 第 2 轮：拿到锁，执行 digest（扫描 128 个文档）
[2026-10-10T19:07:50+00:00] worker-B 第 2 轮：完成并释放锁
[2026-10-10T19:07:50+00:00] worker-B 第 3 轮：未拿到锁（已被占用），跳过本轮

把两份日志对齐看，结论一目了然：每一轮里，两个副本恰好只有一个拿到锁——第 1 轮 A 拿、B 跳过；第 2 轮 B 拿、A 跳过；第 3 轮 A 拿、B 跳过。三个周期、三次执行，没有一轮重复跑。这正是分布式锁要保证的：不管部署几个副本，周期任务全局只执行一次。而且注意锁是「谁先到谁拿」，不是固定某个副本——副本数变了，行为不变。

7.3.6 释放与续期：别 DEL 别人的锁

上一节讲了 token 校验的道理，这里把它真跑一遍，看「用错 token 会怎样」：

A 抢锁: True
B 用错 token 释放: False -> 锁仍在: token-A
A 用对 token 释放: True -> 锁已删: None
续期前 ttl: 297 ms
token 匹配续期: True -> 续期后 ttl: 1997 ms
token 不匹配续期: False

三行输出各说明一件事。B 用错 token 释放返回 False，锁 token-A 纹丝不动——如果这里是裸 DEL，B 就会把 A 的锁删掉，然后第三个副本趁虚而入，互斥当场失效。A 用对 token 释放返回 True，锁消失——合法释放。续期时 token 不匹配返回 False——防止把已经易主给别人的锁错误延长。

续期解决的是另一个现实问题：任务执行时间可能超过锁的 ttl。如果 ttl 设 300ms、任务却要跑 500ms，任务跑到 200ms 时锁就过期了，别的副本会抢进来、同一份活被两个副本同时干。对策是在持有期内定期续租（比如每 ttl/3 续一次），让锁的存活时间跟着任务走。当然，续期本身也要比对 token——否则一个已经失去锁的副本会把新持有者的锁续下去。

7.3.7 延迟任务：ZSET 当定时器

周期任务靠「循环 + 抢锁」，延迟任务（「run 完成后 5 分钟生成摘要」）则靠有序集合。Redis 的 ZSET 按 score 排序，把「到期时间戳」当 score，它天然就是一个定时器：到点的任务 score 最小，排在最前。

def schedule(r, job_id: str, run_id: str, delay_s: float) -> None:
    """到点时间当 score 写进有序集合：ZADD 按 score 排序，天然是个定时器。"""
    due = time.time() + delay_s
    r.zadd(ZSET, {json.dumps({"job_id": job_id, "run_id": run_id}): due})
    print(f"[{now()}] 排入延迟任务 {job_id}，{delay_s:.0f}s 后到期（run={run_id}）")


def poll_once(r) -> bool:
    """取一个到点的任务：ZRANGEBYSCORE 找 + ZREM 原子认领（删成功才算抢到）。"""
    due = r.zrangebyscore(ZSET, 0, time.time(), start=0, num=1)
    if not due:
        return False
    raw = due[0]
    if r.zrem(ZSET, raw) == 0:      # 被别的 worker 抢走了
        return False
    job = json.loads(raw)
    print(f"[{now()}] 到期触发 {job['job_id']}：为 run={job['run_id']} 生成摘要")
    return True

ZADD 时 score 是 time.time() + delay_s，即「到点时刻」。轮询时用 ZRANGEBYSCORE ZSET 0 now LIMIT 0 1 找出「到期时间 ≤ 现在」的第一个任务。关键在于认领必须用 ZREM：ZRANGEBYSCORE 只是「看」，多个 worker 会同时看到同一个任务；ZREM 是「删」，谁删成功（返回 1）谁才真正认领到，删失败（返回 0）说明被抢走了。「看 + 删」分离、以删成功为准，这和 7.1 里 BLMOVE 的「原子搬走」是同一个思路——认领动作必须是原子的。

真跑一次（延迟 1 秒，模拟「5 分钟后生成摘要」）：

[2026-10-10T19:07:51+00:00] 排入延迟任务 digest-1，1s 后到期（run=run-abc）
[2026-10-10T19:07:51+00:00] 调度器空转等待到点……
[2026-10-10T19:07:52+00:00] 到期触发 digest-1：为 run=run-abc 生成摘要
[2026-10-10T19:07:52+00:00] 触发数=1，队列剩余=0

任务在 19:07:51 排入、约定 1 秒后到期，调度器在 19:07:52 检测到它到点、触发、并从集合中移除。触发数=1 说明只触发了一次（ZREM 保证），队列剩余=0 说明任务被正确消费。生产里把 delay_s 换成 300，就是「run 完成后 5 分钟生成摘要」。

ZSET 方案还有一个附带的好处：它天然能查「还有多少任务待触发」（ZCARD）以及「最早到期的任务是什么时候」（ZRANGE WITHSCORES）。这两条信息可以直接进看板，让运维一眼看出延迟队列有没有积压——而这正是「定时任务悄悄堆积、直到某天才被发现」这类事故的早期信号。

7.3.8 定时任务清单表

把 ResearchCopilot 的调度任务摊开，逐条定它的周期、失败影响与是否需要锁：

规律是：「有全局副作用」的任务必须加锁，「幂等且可重复」的任务可以不加。清理过期记忆虽然重复跑无害，但多副本同时删同一批数据会争锁、浪费，所以加锁省事；而指标聚合每个副本写自己那部分数据，本来就不冲突，加锁反而成了瓶颈。

还有一条纪律：定时任务本身要幂等。即使加了锁，也不能假设「一定只执行一次」——锁过期、副本重启、重放都可能让任务多跑。所以「重建索引」应该设计成「无论跑几次，结果都一样」（比如全量重建、原子替换），而不是「追加式」的（比如往索引里 INSERT 增量）。锁是优化，幂等才是底线：锁失效时，幂等保证系统仍然正确。

7.3.9 真实环境会怎样

本节的锁是「单 Redis 实例」的，真实系统还有几处要升级：

• 单实例锁不是强一致。Redis 主从异步复制下，主节点写入锁、还没同步就宕机，从节点升主后锁可能丢失，出现两个持有者。要求强一致时用 Redlock 或基于共识的锁（etcd / ZooKeeper）。对「重建索引」这类任务，单实例锁足够；对「扣款」这类强一致需求，单实例锁不够。
• 时钟漂移。延迟任务依赖各副本的墙钟，机器时钟不同步会让「到点」判断偏掉。生产里应让所有节点跑 NTP 同步，或者用 Redis 服务端时间（TIME 命令）而非本地时间。
• 调度器单点。如果用「一个中央调度器负责投递」，那个调度器本身就成了单点，需要高可用（主备选举）。另一种做法是去中心化——每个副本都尝试抢锁执行（本节的做法），把「唯一性」交给锁而不是交给某个进程。
• 长时间任务要续期。任务执行时间不可控时（比如重建一个很大的索引），必须在持有期内定期 renew，否则任务没跑完锁先过期。本节演示的 renew 就是这个用途。
• 调度要有全局观。当定时任务多起来（本节表格里就有六条），散落在各个 worker 的循环里会很难维护。真实系统常把它们收敛到一个统一的调度器（APScheduler、Celery Beat、K8s CronJob），由它负责「什么时候触发哪个任务」，worker 只负责执行。

本节验收清单：

• 周期任务用 SET NX PX 抢锁，多个副本同一轮只有一个执行
• 释放与续期都先比对 token，绝不 DEL / 延长别人的锁
• 锁带过期时间，持有者崩溃后能自动释放
• 长任务在持有期内续租，避免任务没完锁先过期
• 延迟任务用 ZSET 的 score 表达到期时间，认领用 ZREM 原子完成

一条值得记住的原则：调度负责「什么时候做」，锁负责「谁来做」，幂等负责「做几次都安全」。三者分工明确，任何一层缺失都会让定时任务从「自动化」变成「自动出事故」。

到这里，第 7 章把「一次 run」从同步请求彻底改造成了异步任务：7.1 让它可投递、可恢复，7.2 让它可重试、有归宿、可安全重放，7.3 补上了不由用户触发的周期与延迟任务。三节合起来，ResearchCopilot 的「后台」才算真正立住——请求只负责投递，执行、重试、调度都在 worker 侧自洽运转。但队列只解决了「任务之间」的并发——单个 run 内部（一次跑多个工具、拉多个子 Agent）还是串行的。下一节 8.1 引擎内并发：并行工具与并行子 Agent 会把并发下沉到引擎内部。想回看重试与死信是怎么给这条队列兜底的，翻 7.2 重试、死信与幂等 。

阅读导航：上一节：7.2 重试、死信与幂等 · 下一节：8.1 引擎内并发：并行工具与并行子 Agent 。