Pathlib 原子写入临时文件

示例说明

直接以写模式打开目标文件，进程崩溃时会留下半成品——读者看到不完整的 JSON 或空文件。原子写入的公式是三步：写临时文件、fsync 强制数据落盘、rename 原子替换目标。rename 在同一文件系统内是原子操作，读者要么看到旧文件要么看到新文件，永远没有中间态；崩溃时最坏情况是多一个临时文件，正式文件完好。

技术要点

• 临时文件与目标同目录保证 rename 原子
• fsync 先于 rename 防数据与元数据乱序
• 异常路径清理暂存文件不留垃圾

代码解读

atomic_write 先建父目录，再用 mkstemp 在目标目录创建带隐藏前缀的临时文件；写入后 flush 加 fsync 确保内容先于 rename 落盘；os.replace 在 Windows 与 Unix 语义一致地原子覆盖。except 分支删除临时文件后重抛异常。read_or_default 演示缺省初始化的容错读取。

使用建议

高并发写同一目标时 rename 只是原子不是互斥，需要文件锁配合。fsync 有性能成本，按数据重要性取舍。临时文件名加隐藏前缀便于巡检清理。

代码示例

import os
import tempfile
from pathlib import Path


def atomic_write(path: Path, data: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)

    # 临时文件必须与目标同目录，rename 才能保证原子性
    fd, tmp_name = tempfile.mkstemp(
        dir=path.parent, prefix=f'.{path.name}.', suffix='.tmp'
    )
    tmp_path = Path(tmp_name)

    try:
        with os.fdopen(fd, 'wb') as fh:
            fh.write(data)
            fh.flush()
            os.fsync(fh.fileno())  # 确保数据先于元数据落盘
        os.replace(tmp_path, path)  # 原子替换，读者永远看不到半成品
    except BaseException:
        tmp_path.unlink(missing_ok=True)
        raise


def read_or_default(path: Path, default: bytes) -> bytes:
    try:
        return path.read_bytes()
    except FileNotFoundError:
        atomic_write(path, default)
        return default


config = Path('state/registry.json')
payload = b'{"version": 2, "entries": 42}'

atomic_write(config, payload)
print('size on disk:', read_or_default(config, b'{}'))

partial = Path('state/huge.bin')
atomic_write(partial, b'x' * 1024 * 1024)
print('big write done:', partial.stat().st_size)