从原理到实战详解Python中文件并发读写的避坑指南

import os
import tempfile
import time

def safe_write(filepath, content):
    """安全写入文件：临时文件+原子替换 🛡️"""
    # 创建自动清理的临时文件（suffix指定后缀，dir确保同文件系统）
    with tempfile.NamedTemporaryFile(
        suffix=".tmp",
        dir=os.path.dirname(filepath),
        delete=False,  # 手动控制删除时机 ⏳
        mode="w",
        encoding="utf-8"
    ) as tmp_f:
        # 模拟耗时写入过程
        for char in content:
            tmp_f.write(char)
            tmp_f.flush()  # 强制刷入磁盘 💾
            print(f"[写进程] 已写入: {tmp_f.tell()}/{len(content)}", end="\r")
            time.sleep(0.1)
    
    try:
        # 确保数据完全落盘（避免内存缓存导致的替换异常）
        with open(tmp_f.name, "r") as f:
            os.fsync(f.fileno())
        
        # 原子替换原文件 🔄
        os.replace(tmp_f.name, filepath)
        print(f"\n[写进程] 替换完成，{filepath} 已更新 ✅")
    except Exception as e:
        # 异常时清理临时文件 🧹
        os.unlink(tmp_f.name)
        raise RuntimeError(f"写入失败: {e} ❌")

# 测试：同时运行读写进程
if __name__ == "__main__":
    target = "data.txt"
    # 初始化原文件
    if not os.path.exists(target):
        with open(target, "w") as f:
            f.write("初始内容")
        print(f"[写进程] 已创建初始文件 {target} 📄")
    
    # 模拟写入
    safe_write(target, "这是完整的新内容，不会被读残缺！")

import time

def continuous_read(filepath):
    """持续读取文件，观察内容变化 📖"""
    print("=== 读进程启动 ===")
    try:
        while True:
            with open(filepath, "r", encoding="utf-8") as f:
                content = f.read()
                print(f"[读进程] 内容: {repr(content)}")
            time.sleep(0.5)  # 间隔0.5秒读取一次 ⏱️
    except KeyboardInterrupt:
        print("\n读进程退出 🚪")

if __name__ == "__main__":
    continuous_read("data.txt")

from filelock import FileLock
import time

def write_with_lock(filepath, content):
    """加排他锁写入文件 🔐"""
    # 锁文件与目标文件同目录，便于管理
    lock_path = f"{filepath}.lock"
    with FileLock(lock_path, timeout=5):  # 超时5秒避免死锁 ⏱️
        print("[写进程] 获得锁，开始写入 ✍️")
        with open(filepath, "w", encoding="utf-8") as f:
            for char in content:
                f.write(char)
                f.flush()  # 强制刷盘 💾
                time.sleep(0.1)
        print("[写进程] 写入完成，释放锁 🗝️")

def read_with_lock(filepath):
    """加共享锁读取文件 📖"""
    lock_path = f"{filepath}.lock"
    with FileLock(lock_path, shared=True, timeout=5):  # 共享锁允许多进程读 🤝
        print("[读进程] 获得共享锁，开始读取 👀")
        with open(filepath, "r", encoding="utf-8") as f:
            content = f.read()
        print(f"[读进程] 读取内容: {repr(content)}")
        time.sleep(1)  # 模拟读取耗时 ⏳