Python 定时任务如何安全保存 JSON 状态：临时文件与原子替换

定时脚本经常需要保存“上一次处理的位置”“已经发布的记录”或“任务执行结果”。直接打开状态文件并覆盖写入虽然简单，但如果进程在写到一半时退出，下次运行可能读到不完整的 JSON。

本文给出一个只依赖 Python 标准库的实现，并说明它能解决什么、不能解决什么。示例适合单进程定时任务，不涉及任何云服务凭据。

一、不要直接覆盖唯一一份状态

下面的代码会先截断旧文件，再写入新内容：

with open("state.json", "w", encoding="utf-8") as f:
    json.dump(state, f, ensure_ascii=False)

一旦序列化失败或写入中断，原本可用的旧状态也可能丢失。更稳妥的思路是：先在目标目录写临时文件，写完后再替换目标文件。

二、一个可复用的实现

import json
import os
import tempfile
from pathlib import Path


def save_json(path, data):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(
        prefix=path.name + ".",
        suffix=".tmp",
        dir=path.parent,
    )
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            json.dump(
                data, f,
                ensure_ascii=False,
                allow_nan=False,
            )
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp, path)
    finally:
        if os.path.exists(tmp):
            os.unlink(tmp)

使用方式：

save_json("state.json", {"状态": "完成", "次数": 1})

临时文件放在目标文件所在目录，避免跨文件系统替换。flush() 将 Python 缓冲区提交给操作系统，fsync() 请求同步文件数据，最后通过 os.replace() 切换到新文件。

allow_nan=False 用来拒绝标准 JSON 不支持的 NaN、Infinity 等数值，避免输出只能被部分解析器接受的内容。

三、验证失败时不会破坏旧状态

以下检查应在放有上述函数的同一脚本中运行。为了不覆盖业务文件，使用独立的临时目录。

with tempfile.TemporaryDirectory() as directory:
    target = Path(directory) / "state.json"
    expected = {"状态": "完成", "次数": 1}
    save_json(target, expected)
    assert json.loads(target.read_text(encoding="utf-8")) == expected

    try:
        save_json(target, {"bad": float("nan")})
    except ValueError:
        pass
    else:
        raise AssertionError("NaN should be rejected")

    assert json.loads(target.read_text(encoding="utf-8")) == expected
    assert not list(Path(directory).glob("*.tmp"))

本文核心保存函数已经在 Linux 环境实际执行检查：中文往返读取通过；非法数值触发失败后旧文件内容保持不变；本次异常路径没有遗留临时文件。这些检查不是断电或所有文件系统行为的证明。

四、四个容易忽略的边界

1. 原子替换不等于并发锁。 两个进程同时读旧状态、各自修改再保存，仍可能出现后写覆盖先写的问题。多进程场景可以使用文件锁，或改用 SQLite 事务。
2. 文件同步不等于完整的断电持久性保证。 某些 POSIX 场景还需要同步父目录，具体行为也受文件系统和存储设备影响。上面的简化实现不作绝对断电安全承诺。
3. 异常清理不覆盖强制终止。 finally 可以处理通常的 Python 异常，但 SIGKILL 等情况下不会执行。需要长期运行时，可增加有明确范围、不会误删活动文件的临时文件清理机制。
4. 新文件权限需要明确。 mkstemp() 创建的文件通常仅当前用户可读写，替换后不会自动继承旧目标文件的权限。需要其他服务读取时，应单独设计权限策略。

五、定时任务还需要业务去重

安全保存状态只解决文件写入问题，并不能保证外部操作“只执行一次”。例如发布文章后网络超时，脚本可能不知道服务端是否已经接收。

正确的重试方式是先查询外部实际状态，再决定是否重新提交；如果服务提供幂等键，应使用稳定的业务标识。不要因为本地状态没有更新，就直接重复发布、重复领取或重复扣费。

总结

单进程脚本保存少量 JSON 状态时，可以采用“同目录临时文件 → 写入并同步 → 原子替换”的方式，降低状态文件半写入的风险。涉及并发、事务、严格持久性或外部副作用时，应继续补充锁、数据库和幂等机制，而不是把一个文件写入函数当成完整的可靠性方案。