让定时任务“断点续跑”：一个 JSON 队列文件的幂等调度实践

让定时任务"断点续跑"：一个 JSON 队列文件的幂等调度实践

我有一批凌晨定时跑的无人值守脚本，职责很朴素：去一个任务平台挑一件事做，做完把结果记下来。听起来十几行代码的事，真跑起来却连续踩坑：进程每次都是全新的、任务做一半网络抖了、有些任务一周只能做一次、有的任务压根做不了却每轮都去撞墙。最后解决问题的不是框架，而是一个几百行的 JSON 队列文件加一个很小的状态机。

这篇文章记录这套设计，以及我在真实运行中补进去的几个细节。所有代码只用标准库。

一、为什么不能靠内存记住进度

定时任务的第一直觉是"跑完就结束"，但真实情况往往是：

• 每轮都是全新进程，上一轮的变量全部蒸发；
• 一轮里做不完，下一轮得知道是"接着做"还是"重新做"；
• 有些动作有配额（每天 1 次、每周 1 次），重复执行不加分，还占用额度；
• 有的任务这一轮就是做不了（要人脸核验、要付费），必须记住"别再试了"。

这四点合起来只有一个诉求：进度必须落盘，而且要带上"为什么停在这里"的原因。

二、队列结构：一个 JSON 就够

{
  "updated": "2026-11-03 03:10",
  "queue": [
    {
      "id": "blog",
      "title": "每日发布文章",
      "points": 60,
      "cycle": "DAILY",
      "status": "todo",
      "note": "每日任务，上次完成 2026-11-02",
      "done_at": ""
    }
  ]
}

字段没有一个是装饰用的：cycle 决定"这一轮还能不能再做"，status 决定"要不要做"，done_at 决定"上次是什么时候做的"，note 存的是人（也就是下一轮的我）能看懂的原因。JSON 而不是 SQLite，是因为它可以直接 cat、可以 diff、可以用文本编辑器手工修，出错时用肉眼就能定位。

三、四个状态，够用就好

• todo：可做，等着做；
• done：本周期已完成，带 done_at；
• blocked：做不了，附原因（需要人脸核验 / 需要付费 / 要填个人信息 / 依赖的上游功能坏了）；
• passive：被动等待，比如提交的建议等平台审核，主动做也没用。

我一开始只设了 todo 和 done，结果每轮都在明知做不了的任务上浪费十分钟，加上 blocked 之后，扫描一轮的时间从十几分钟降到一分钟以内。把"已知的不可行"也持久化下来，是这套设计里收益最大的一条。

四、区分周期：日任务比日期，周任务比 ISO 周

import datetime

def is_due(item, today):
    cycle = item.get("cycle", "ONCE")
    done = item.get("done_at", "")
    if not done:
        return True
    d = datetime.date.fromisoformat(done[:10])
    if cycle == "DAILY":
        return d != today
    if cycle == "WEEKLY":
        return d.isocalendar()[:2] != today.isocalendar()[:2]
    if cycle == "MONTHLY":
        return (d.year, d.month) != (today.year, today.month)
    return False

这里有个真实的坑：周任务不是"隔七天"。周一做完后周二又跑一次，按"间隔天数"判断会放行，但平台按自然周重置，这一轮白做。isocalendar() 返回的是 ISO 周，跨年那一周也正确，比自己算天数稳。

五、每轮只做一个

我给自己定的硬规则是：一轮最多推进一个任务。原因很实际——失败要能定位到具体动作；多个任务并发时，一个页面掉登录会污染后面所有任务的判定；而且慢一点不影响最终结果，快一点却容易把"做了一半"的状态写坏。跑得久的任务（比如写文章）本来就该单独占一轮。

六、写回必须原子

import json, os, tempfile

def save(path, data):
    d = os.path.dirname(path) or "."
    fd, tmp = tempfile.mkstemp(dir=d, suffix=".tmp")
    with os.fdopen(fd, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=1)
    os.replace(tmp, path)

os.replace 在同一文件系统内是原子的：要么看到旧文件，要么看到新文件，不会出现"写了一半"的 JSON。这一步省不得——进程被 OOM 杀掉、被超时打断，都正好发生在写文件的中间。我吃过一次亏，半个 JSON 让下一轮直接崩在读文件阶段，之后所有写回都走这个封装。

七、判定"做成了"要看外部指标，不要看按钮

最容易自欺的一环。页面上的按钮很诚实地写着"去完成"，但后台的积分是批处理入账的，可能晚几十分钟甚至更久。我第一版脚本的结论是"任务没完成，白做了"，实际上是采样太早。

可靠的做法是把可观测的外部指标当判据：

before = read_points()
do_task()
after = read_points()          # 立刻读，可能还没入账
record_delta(before, after)    # 记下来，下一轮用差额复核

看不到即时结果时，不要立刻下"失败"的结论，把差额留到下一轮复核。 这一条把我的误判率降到了零。

八、几个真踩过的坑

• pkill -f 会匹配到自己：模式写得太宽时，pkill 会把执行它的那条命令行也算进去，结果把当前 shell 一起杀掉。先 pgrep 拿 PID 再 kill，别偷懒。
• 不要为了凑数提交空内容：任务平台要求"提交一个真实问题"，找不到就不提。虚假提交短期看是多拿了一次分，长期看是把账号的信用做没了，不划算。
• 掉登录要先确认判定方式：只认页面右上角出现"登录/注册"这一个信号，其他位置的状态条经常是缓存的旧值。

九、复盘清单

跑了一段时间后，我保留下来的检查项只有五条：

1. 进度是否全部落盘（内存里不许有"唯一真相"）；
2. 周期判断用的是自然周/自然日，不是间隔时长；
3. 写回走原子替换；
4. 成功与否用外部指标判定，延迟入账留给下一轮复核；
5. 做不了的任务要留下原因，而不是留空。

这套东西没有用到任何框架，换来的是：任意一轮被杀掉，下一轮都能从正确的位置继续，而且我能看懂它为什么停在那儿。对无人值守脚本来说，"可解释地停住"比"拼命跑完"重要得多。