给 Python 配置文件做原子替换:写到一半失败时,旧文件还能用吗?

举报
yd_234095983 发表于 2026/11/09 08:59:47 2026/11/09
【摘要】 用 open(path, "w") 保存配置时,文件一打开就会被截断。如果序列化失败,或者另一个进程恰好在写入途中读取,它可能只拿到一段不能解析的 JSON。对一个由单个进程更新、多个进程读取的本地配置,更实用的发布点是一次同目录的 os.replace:先把新内容写完整,再替换目标路径。 把写入与发布分开下面的函数接受一个已有父目录下的文件路径。它先序列化,避免不支持的对象触及磁盘;临时文...

用 open(path, "w") 保存配置时,文件一打开就会被截断。如果序列化失败,或者另一个进程恰好在写入途中读取,它可能只拿到一段不能解析的 JSON。对一个由单个进程更新、多个进程读取的本地配置,更实用的发布点是一次同目录的 os.replace:先把新内容写完整,再替换目标路径。

把写入与发布分开

下面的函数接受一个已有父目录下的文件路径。它先序列化,避免不支持的对象触及磁盘;临时文件和目标位于同一目录,避免跨文件系统替换。先 flush 把 Python 缓冲区的数据交给系统,再 fsync 临时文件;两个步骤都成功后才关闭临时文件并替换目标。这段示例只讨论本地文件系统,不需要云资源或第三方库。

import json
import os
from pathlib import Path
from tempfile import NamedTemporaryFile


def write_config(path: Path, config: dict) -> None:
    text = json.dumps(config, ensure_ascii=False, allow_nan=False, indent=2) + "\n"
    temporary_path = None
    try:
        with NamedTemporaryFile(
            mode="w", encoding="utf-8", newline="\n", dir=path.parent,
            prefix=f".{path.name}.", suffix=".tmp", delete=False,
        ) as stream:
            temporary_path = Path(stream.name)
            stream.write(text)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary_path, path)
    finally:
        if temporary_path is not None:
            temporary_path.unlink(missing_ok=True)

在支持同目录原子替换的文件系统上,从目标路径重新打开文件的读者看到的是旧文件或新文件,不会看到临时文件写到一半的内容。已经持有旧文件句柄的读者不会自动切换到新版本;需要刷新时应重新打开路径。若序列化、写入或替换在发布前失败,函数会报告异常,不应在调用方把这次保存标为成功。

用失败注入验证旧文件没有被截断

下面的测试可接在同一文件末尾运行。先验证 UTF-8 往返,再模拟替换失败,检查旧字节不变且临时文件已清理;最后验证正常更新和非法值被拒绝。模拟替换失败不等价于模拟整机断电。

from tempfile import TemporaryDirectory
from unittest.mock import patch


with TemporaryDirectory() as directory:
    path = Path(directory) / "settings.json"
    write_config(path, {"name": "相册", "version": 1})
    original = path.read_bytes()
    assert json.loads(original) == {"name": "相册", "version": 1}
    assert "相册".encode("utf-8") in original

    with patch.object(os, "replace", side_effect=PermissionError("模拟拒绝替换")):
        try:
            write_config(path, {"version": 2})
        except PermissionError:
            pass
        else:
            raise AssertionError("应传播保存失败")
    assert path.read_bytes() == original
    assert list(path.parent.iterdir()) == [path]

    write_config(path, {"version": 2})
    assert json.loads(path.read_bytes()) == {"version": 2}
    try:
        write_config(path, {"version": float("nan")})
    except ValueError:
        pass
    else:
        raise AssertionError("应拒绝非标准 JSON 数值")
    assert json.loads(path.read_bytes()) == {"version": 2}

原子可见性不等于断电后一定持久

  • 持久性:示例同步了新文件内容,没有同步父目录。在支持目录同步的 POSIX 文件系统上,如果还要求断电后尽可能保留新目录项,需要在替换后对父目录执行 fsync 并处理失败;此时替换已经发生,目录同步失败不能被当作“旧版本仍在”。不要把本地实验的结果直接推广到网络文件系统。
  • 并发写入:两名写入者都从旧配置读取并各自更新时,原子替换只保证读者不会看到半份文件,不能防止后写者覆盖先写者。若要保护“读取—修改—保存”全过程,应使用跨进程锁、版本检查或数据库事务。
  • 权限与清理:替换的是新文件,不保证继承旧文件的权限、所有者或 ACL;如果配置要由其他账户读取,须明确设计权限。进程被强杀时 finally 不会运行,清理残留临时文件也要避免误删另一个正在写入的进程的文件。

本文示例由 AI 辅助整理,使用本地 Python 标准库验证;没有操作真实业务配置。

【声明】本内容来自华为云开发者社区博主,不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0)

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。